diff --git a/frame.go b/frame.go index dacddc6..300473e 100644 --- a/frame.go +++ b/frame.go @@ -60,25 +60,25 @@ func (f *Frame) UnmarshalBinary(bts []byte) error { return nil } -type RawHeader []byte +type rawHeader []byte -func (h RawHeader) Version() byte { +func (h rawHeader) Version() byte { return h[0] } -func (h RawHeader) Cmd() byte { +func (h rawHeader) Cmd() byte { return h[1] } -func (h RawHeader) StreamID() uint32 { +func (h rawHeader) StreamID() uint32 { return binary.LittleEndian.Uint32(h[2:]) } -func (h RawHeader) Length() uint32 { +func (h rawHeader) Length() uint32 { return binary.LittleEndian.Uint32(h[6:]) } -func (h RawHeader) String() string { +func (h rawHeader) String() string { return fmt.Sprintf("Version:%d Cmd:%d StreamID:%d Length:%d", h.Version(), h.Cmd(), h.StreamID(), h.Length()) } diff --git a/session.go b/session.go index 025f388..c17fced 100644 --- a/session.go +++ b/session.go @@ -2,7 +2,6 @@ package smux import ( "io" - "log" "sync" "github.com/pkg/errors" @@ -73,11 +72,18 @@ func newSession(maxframes uint32, conn io.ReadWriteCloser, client bool) *Session func (s *Session) OpenStream() (*Stream, error) { chNotifyReader := make(chan struct{}, 1) stream := newStream(s.nextStreamID, defaultFrameSize, chNotifyReader, s) + + // track stream s.mu.Lock() - defer s.mu.Unlock() s.rdEvents[s.nextStreamID] = chNotifyReader s.nextStreamID += 2 s.streams[stream.id] = stream + s.mu.Unlock() + + // send SYN packet + f := newFrame(cmdSYN, stream.id) + bts, _ := f.MarshalBinary() + s.lw.Write(bts) return stream, nil } @@ -130,7 +136,7 @@ func (s *Session) readFrame() (f Frame, err error) { return f, errors.Wrap(err, "readFrame") } - dec := RawHeader(h) + dec := rawHeader(h) data := h if dec.Length() > 0 { data = make([]byte, headerSize+dec.Length()) @@ -143,6 +149,7 @@ func (s *Session) readFrame() (f Frame, err error) { return f, err } +// recvLoop keeps on reading from underlying connection if tokens are available func (s *Session) recvLoop() { for { select { @@ -163,7 +170,7 @@ func (s *Session) recvLoop() { } s.mu.Unlock() } else { - log.Println(err) + return } case <-s.die: return diff --git a/session_test.go b/session_test.go index b30a1bd..7235124 100644 --- a/session_test.go +++ b/session_test.go @@ -36,7 +36,7 @@ func handleConnection(conn net.Conn) { for { n, err := stream.Read(buf) if err != nil { - panic(err) + return } count++ stream.Write(buf[:n]) @@ -62,7 +62,7 @@ func TestEcho(t *testing.T) { if n, err := stream.Read(buf); err == nil { fmt.Println("recv:", string(buf[:n])) } else { - panic(err) + return } } } diff --git a/stream.go b/stream.go index 2a2c2db..db97c9b 100644 --- a/stream.go +++ b/stream.go @@ -1,6 +1,7 @@ package smux import ( + "io" "sync" "github.com/pkg/errors" @@ -24,9 +25,6 @@ func newStream(id uint32, frameSize uint32, chNotifyReader chan struct{}, sess * s.frameSize = frameSize s.sess = sess s.die = make(chan struct{}) - f := newFrame(cmdSYN, s.id) - bts, _ := f.MarshalBinary() - sess.lw.Write(bts) return s } @@ -42,7 +40,6 @@ READ: f := s.sess.read(s.id) if f != nil { switch f.cmd { - case cmdRST: case cmdPSH: n = copy(b, f.data) if len(f.data) > n { @@ -50,6 +47,8 @@ READ: copy(s.buffer, f.data[n:]) } return n, nil + default: + return 0, io.EOF } }