upd

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())
 }

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
+}

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:

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)
+		}
+	}
+}

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
 }