introduce stream callback in session, move http2 stuff to main.go

This commit is contained in:
Lucas Clemente
2016-04-15 14:31:31 +02:00
parent ec509fb98f
commit 3c16e8c9ee
2 changed files with 52 additions and 22 deletions

View File

@@ -6,6 +6,9 @@ import (
"net" "net"
"os" "os"
"golang.org/x/net/http2"
"golang.org/x/net/http2/hpack"
"github.com/lucas-clemente/quic-go" "github.com/lucas-clemente/quic-go"
"github.com/lucas-clemente/quic-go/crypto" "github.com/lucas-clemente/quic-go/crypto"
"github.com/lucas-clemente/quic-go/protocol" "github.com/lucas-clemente/quic-go/protocol"
@@ -77,7 +80,7 @@ func main() {
session, ok := sessions[publicHeader.ConnectionID] session, ok := sessions[publicHeader.ConnectionID]
if !ok { if !ok {
session = quic.NewSession(conn, publicHeader.ConnectionID, serverConfig) session = quic.NewSession(conn, publicHeader.ConnectionID, serverConfig, handleStream)
sessions[publicHeader.ConnectionID] = session sessions[publicHeader.ConnectionID] = session
} }
err = session.HandlePacket(remoteAddr, data[0:n-r.Len()], publicHeader, r) err = session.HandlePacket(remoteAddr, data[0:n-r.Len()], publicHeader, r)
@@ -86,3 +89,30 @@ func main() {
} }
} }
} }
func handleStream(frame *quic.StreamFrame) *quic.StreamFrame {
h2r := bytes.NewReader(frame.Data)
var reply bytes.Buffer
h2framer := http2.NewFramer(&reply, h2r)
h2framer.ReadMetaHeaders = hpack.NewDecoder(1024, nil)
h2frame, err := h2framer.ReadFrame()
if err != nil {
return nil
}
h2headersFrame := h2frame.(*http2.MetaHeadersFrame)
fmt.Printf("%#v\n", h2headersFrame)
var replyHeaders bytes.Buffer
enc := hpack.NewEncoder(&replyHeaders)
enc.WriteField(hpack.HeaderField{Name: ":status", Value: "204"})
h2framer.WriteHeaders(http2.HeadersFrameParam{
StreamID: h2frame.Header().StreamID,
EndHeaders: true,
BlockFragment: replyHeaders.Bytes(),
})
return &quic.StreamFrame{
StreamID: frame.StreamID,
Data: reply.Bytes(),
}
}

View File

@@ -6,14 +6,14 @@ import (
"fmt" "fmt"
"net" "net"
"golang.org/x/net/http2"
"golang.org/x/net/http2/hpack"
"github.com/lucas-clemente/quic-go/crypto" "github.com/lucas-clemente/quic-go/crypto"
"github.com/lucas-clemente/quic-go/handshake" "github.com/lucas-clemente/quic-go/handshake"
"github.com/lucas-clemente/quic-go/protocol" "github.com/lucas-clemente/quic-go/protocol"
) )
// StreamCallback gets a stream frame and returns a reply frame
type StreamCallback func(*StreamFrame) *StreamFrame
// A Session is a QUIC session // A Session is a QUIC session
type Session struct { type Session struct {
ConnectionID protocol.ConnectionID ConnectionID protocol.ConnectionID
@@ -27,15 +27,18 @@ type Session struct {
Entropy EntropyAccumulator Entropy EntropyAccumulator
lastSentPacketNumber protocol.PacketNumber lastSentPacketNumber protocol.PacketNumber
streamCallback StreamCallback
} }
// NewSession makes a new session // NewSession makes a new session
func NewSession(conn *net.UDPConn, connectionID protocol.ConnectionID, sCfg *ServerConfig) *Session { func NewSession(conn *net.UDPConn, connectionID protocol.ConnectionID, sCfg *ServerConfig, streamCallback StreamCallback) *Session {
return &Session{ return &Session{
Connection: conn, Connection: conn,
ConnectionID: connectionID, ConnectionID: connectionID,
ServerConfig: sCfg, ServerConfig: sCfg,
aead: &crypto.NullAEAD{}, aead: &crypto.NullAEAD{},
streamCallback: streamCallback,
} }
} }
@@ -70,7 +73,7 @@ func (s *Session) HandlePacket(addr *net.UDPAddr, publicHeaderBinary []byte, pub
frameCounter++ frameCounter++
fmt.Printf("Reading frame %d\n", frameCounter) fmt.Printf("Reading frame %d\n", frameCounter)
if typeByte&0x80 == 0x80 { // STREAM if typeByte&0x80 > 0 { // STREAM
fmt.Println("Detected STREAM") fmt.Println("Detected STREAM")
frame, err := ParseStreamFrame(r, typeByte) frame, err := ParseStreamFrame(r, typeByte)
if err != nil { if err != nil {
@@ -82,21 +85,17 @@ func (s *Session) HandlePacket(addr *net.UDPAddr, publicHeaderBinary []byte, pub
return errors.New("Session: 0 is not a valid Stream ID") return errors.New("Session: 0 is not a valid Stream ID")
} }
// TODO: Switch stream here
if frame.StreamID == 1 { if frame.StreamID == 1 {
s.HandleCryptoHandshake(frame) s.HandleCryptoHandshake(frame)
} else { } else {
h2r := bytes.NewReader(frame.Data) replyFrame := s.streamCallback(frame)
h2framer := http2.NewFramer(nil, h2r) replyFrames := []Frame{&AckFrame{Entropy: s.Entropy.Get(), LargestObserved: 3}}
h2framer.ReadMetaHeaders = hpack.NewDecoder(1024, nil) if replyFrame != nil {
h2frame, err := h2framer.ReadFrame() replyFrames = append(replyFrames, replyFrame)
if err != nil {
return err
} }
h2headersFrame := h2frame.(*http2.MetaHeadersFrame) s.SendFrames(replyFrames)
fmt.Printf("%#v\n", h2headersFrame)
panic("streamid not 1")
} }
continue
} else if typeByte&0xC0 == 0x40 { // ACK } else if typeByte&0xC0 == 0x40 { // ACK
fmt.Println("Detected ACK") fmt.Println("Detected ACK")
frame, err := ParseAckFrame(r, typeByte) frame, err := ParseAckFrame(r, typeByte)
@@ -106,18 +105,19 @@ func (s *Session) HandlePacket(addr *net.UDPAddr, publicHeaderBinary []byte, pub
fmt.Printf("%#v\n", frame) fmt.Printf("%#v\n", frame)
continue // not yet implemented continue
} else if typeByte&0xE0 == 0x20 { // CONGESTION_FEEDBACK } else if typeByte&0xE0 == 0x20 { // CONGESTION_FEEDBACK
return errors.New("Detected CONGESTION_FEEDBACK") return errors.New("Detected CONGESTION_FEEDBACK")
} else if typeByte&0x06 == 0x06 { // STOP_WAITING } else if typeByte&0x06 == 0x06 { // STOP_WAITING
fmt.Println("Detected STOP_WAITING") fmt.Println("Detected STOP_WAITING")
r.ReadByte() r.ReadByte()
r.ReadByte() r.ReadByte()
continue
} else if typeByte == 0 { } else if typeByte == 0 {
// PAD // PAD
return nil return nil
} }
return errors.New("Session: invalid Frame Type Field") return fmt.Errorf("Session: invalid frame type %x", typeByte)
} }
return nil return nil
} }