diff --git a/frame.go b/frame.go index ac7e6fe..4825ee3 100644 --- a/frame.go +++ b/frame.go @@ -75,15 +75,11 @@ func (h RawHeader) StreamID() uint32 { return binary.BigEndian.Uint32(h[2:]) } -func (h RawHeader) Wnd() uint32 { +func (h RawHeader) Length() uint32 { return binary.BigEndian.Uint32(h[6:]) } -func (h RawHeader) Length() uint32 { - return binary.BigEndian.Uint32(h[10:]) -} - func (h RawHeader) String() string { - return fmt.Sprintf("Version:%d Cmd:%d StreamID:%d Wnd:%d Length:%d", - h.Version(), h.Cmd(), h.StreamID(), h.Wnd(), h.Length()) + return fmt.Sprintf("Version:%d Cmd:%d StreamID:%d Length:%d", + h.Version(), h.Cmd(), h.StreamID(), h.Length()) } diff --git a/mux.go b/mux.go new file mode 100644 index 0000000..81951ed --- /dev/null +++ b/mux.go @@ -0,0 +1,11 @@ +package smux + +import "io" + +func Server(conn io.ReadWriteCloser) (*Session, error) { + return newSession(4096, conn, false), nil +} + +func Client(conn io.ReadWriteCloser) (*Session, error) { + return newSession(4096, conn, true), nil +} diff --git a/session.go b/session.go index 345cac7..a37774f 100644 --- a/session.go +++ b/session.go @@ -9,7 +9,8 @@ import ( ) const ( - defaultFrameSize = 4096 + defaultFrameSize = 4096 + defaultAcceptBacklog = 1024 ) type Session struct { @@ -28,9 +29,9 @@ type Session struct { rdEvents map[uint32]chan struct{} // session control - die chan struct{} - chAccept chan *Stream - mu sync.Mutex + die chan struct{} + chAccepts chan *Stream + mu sync.Mutex } type lockedWriter struct { @@ -52,6 +53,7 @@ func newSession(maxframes uint32, conn io.ReadWriteCloser, client bool) *Session s.tokens = make(chan struct{}, maxframes) s.streamLines = make(map[uint32][]Frame) s.rdEvents = make(map[uint32]chan struct{}) + s.chAccepts = make(chan *Stream, defaultAcceptBacklog) s.die = make(chan struct{}) for i := uint32(0); i < maxframes; i++ { s.tokens <- struct{}{} @@ -101,7 +103,7 @@ func (s *Session) Accept() (net.Conn, error) { // is ready to be accepted. func (s *Session) AcceptStream() (*Stream, error) { select { - case stream := <-s.chAccept: + case stream := <-s.chAccepts: return stream, nil case <-s.die: return nil, errors.New("session shutdown") @@ -155,6 +157,7 @@ func (s *Session) recvLoop() { chNotifyReader := make(chan struct{}, 1) s.streams[f.sid] = newStream(f.sid, defaultFrameSize, chNotifyReader, s) s.rdEvents[f.sid] = chNotifyReader + s.chAccepts <- s.streams[f.sid] } } case <-s.die: diff --git a/session_test.go b/session_test.go new file mode 100644 index 0000000..94230a3 --- /dev/null +++ b/session_test.go @@ -0,0 +1,65 @@ +package smux + +import ( + "fmt" + "net" + "testing" +) + +func init() { + go func() { + ln, err := net.Listen("tcp", "127.0.0.1:19999") + if err != nil { + // handle error + panic(err) + } + for { + conn, err := ln.Accept() + if err != nil { + // handle error + } + go handleConnection(conn) + } + }() +} + +func handleConnection(conn net.Conn) { + session, _ := Server(conn) + buf := make([]byte, 65536) + count := 0 + for { + stream, _ := session.Accept() + go func() { + for { + n, err := stream.Read(buf) + if err != nil { + panic(err) + } + count++ + stream.Write(buf[:n]) + } + }() + } + +} + +func TestEcho(t *testing.T) { + cli, err := net.Dial("tcp", "127.0.0.1:19999") + if err != nil { + t.Fatal(err) + } + session, _ := Client(cli) + stream, _ := session.OpenStream() + const N = 100 + buf := make([]byte, 10) + for i := 0; i < N; i++ { + msg := fmt.Sprintf("hello%v", i) + fmt.Println("sent:", msg) + stream.Write([]byte(msg)) + if n, err := stream.Read(buf); err == nil { + fmt.Println("recv:", string(buf[:n])) + } else { + panic(err) + } + } +} diff --git a/stream.go b/stream.go index 7acab7e..98f9f5d 100644 --- a/stream.go +++ b/stream.go @@ -18,6 +18,7 @@ type Stream struct { frameSize uint32 die chan struct{} mu sync.Mutex + buffer []byte } func newStream(id uint32, frameSize uint32, chNotifyReader chan struct{}, sess *Session) *Stream { @@ -35,11 +36,25 @@ func newStream(id uint32, frameSize uint32, chNotifyReader chan struct{}, sess * // Read implements io.ReadWriteCloser func (s *Stream) Read(b []byte) (n int, err error) { + if len(s.buffer) > 0 { + n = copy(b, s.buffer) + s.buffer = s.buffer[n:] + return n, nil + } + if f := s.sess.read(s.id); f != nil { switch f.cmd { - case cmdSYN: + case cmdPSH: + n = copy(b, f.data) + if len(f.data) > n { + s.buffer = make([]byte, len(f.data)-n) + copy(s.buffer, f.data[n:]) + } + return n, nil } + println("cmd:", f.cmd) } + return 0, nil }