From 9f184ceddaab8d8c44b7f365421ad02b137d7da4 Mon Sep 17 00:00:00 2001 From: xtaci Date: Thu, 26 Dec 2019 12:30:53 +0800 Subject: [PATCH] config protocol Version in Config --- frame.go | 6 +----- mux.go | 7 +++++++ mux_test.go | 8 ++++++++ session.go | 6 +++--- session_test.go | 20 ++++++++++---------- stream.go | 4 ++-- 6 files changed, 31 insertions(+), 20 deletions(-) diff --git a/frame.go b/frame.go index 71d3d44..e7b338d 100644 --- a/frame.go +++ b/frame.go @@ -5,10 +5,6 @@ import ( "fmt" ) -const ( - version = 1 -) - const ( // cmds cmdSYN byte = iota // stream open cmdFIN // stream close, a.k.a EOF mark @@ -32,7 +28,7 @@ type Frame struct { data []byte } -func newFrame(cmd byte, sid uint32) Frame { +func newFrame(version byte, cmd byte, sid uint32) Frame { return Frame{ver: version, cmd: cmd, sid: sid} } diff --git a/mux.go b/mux.go index bbe07a7..bd9efe4 100644 --- a/mux.go +++ b/mux.go @@ -13,6 +13,9 @@ import ( // Config is used to tune the Smux session type Config struct { + // Version config + Version byte + // KeepAliveInterval is how often to send a NOP command to the remote KeepAliveInterval time.Duration @@ -32,6 +35,7 @@ type Config struct { // DefaultConfig is used to return a default configuration func DefaultConfig() *Config { return &Config{ + Version: 1, KeepAliveInterval: 10 * time.Second, KeepAliveTimeout: 30 * time.Second, MaxFrameSize: 32768, @@ -41,6 +45,9 @@ func DefaultConfig() *Config { // VerifyConfig is used to verify the sanity of configuration func VerifyConfig(config *Config) error { + if config.Version > 2 { + return errors.New("max supported smux version is 2") + } if config.KeepAliveInterval == 0 { return errors.New("keep-alive interval must be positive") } diff --git a/mux_test.go b/mux_test.go index 638e67c..f2963bd 100644 --- a/mux_test.go +++ b/mux_test.go @@ -25,6 +25,14 @@ func TestConfig(t *testing.T) { t.Fatal(err) } + config = DefaultConfig() + config.Version = 3 + err = VerifyConfig(config) + t.Log(err) + if err == nil { + t.Fatal(err) + } + config = DefaultConfig() config.KeepAliveInterval = 10 config.KeepAliveTimeout = 5 diff --git a/session.go b/session.go index d51e365..da6d4fe 100644 --- a/session.go +++ b/session.go @@ -141,7 +141,7 @@ func (s *Session) OpenStream() (*Stream, error) { stream := newStream(sid, s.config.MaxFrameSize, s) - if _, err := s.writeFrame(newFrame(cmdSYN, sid)); err != nil { + if _, err := s.writeFrame(newFrame(s.config.Version, cmdSYN, sid)); err != nil { return nil, err } @@ -367,7 +367,7 @@ func (s *Session) recvLoop() { // read header first if _, err := io.ReadFull(s.conn, hdr[:]); err == nil { atomic.StoreInt32(&s.dataReady, 1) - if hdr.Version() != version { + if hdr.Version() != s.config.Version { s.notifyProtoError(ErrInvalidProtocol) return } @@ -427,7 +427,7 @@ func (s *Session) keepalive() { for { select { case <-tickerPing.C: - s.writeFrameInternal(newFrame(cmdNOP, 0), tickerPing.C, 0) + s.writeFrameInternal(newFrame(s.config.Version, cmdNOP, 0), tickerPing.C, 0) s.notifyBucket() // force a signal to the recvLoop case <-tickerTimeout.C: if !atomic.CompareAndSwapInt32(&s.dataReady, 1, 0) { diff --git a/session_test.go b/session_test.go index 4029720..698ecc9 100644 --- a/session_test.go +++ b/session_test.go @@ -657,7 +657,7 @@ func TestRandomFrame(t *testing.T) { } session, _ = Client(cli, nil) for i := 0; i < 100; i++ { - f := newFrame(cmdSYN, 1000) + f := newFrame(1, cmdSYN, 1000) session.writeFrame(f) } cli.Close() @@ -670,7 +670,7 @@ func TestRandomFrame(t *testing.T) { allcmds := []byte{cmdSYN, cmdFIN, cmdPSH, cmdNOP} session, _ = Client(cli, nil) for i := 0; i < 100; i++ { - f := newFrame(allcmds[rand.Int()%len(allcmds)], rand.Uint32()) + f := newFrame(1, allcmds[rand.Int()%len(allcmds)], rand.Uint32()) session.writeFrame(f) } cli.Close() @@ -682,7 +682,7 @@ func TestRandomFrame(t *testing.T) { } session, _ = Client(cli, nil) for i := 0; i < 100; i++ { - f := newFrame(byte(rand.Uint32()), rand.Uint32()) + f := newFrame(1, byte(rand.Uint32()), rand.Uint32()) session.writeFrame(f) } cli.Close() @@ -694,7 +694,7 @@ func TestRandomFrame(t *testing.T) { } session, _ = Client(cli, nil) for i := 0; i < 100; i++ { - f := newFrame(byte(rand.Uint32()), rand.Uint32()) + f := newFrame(1, byte(rand.Uint32()), rand.Uint32()) f.ver = byte(rand.Uint32()) session.writeFrame(f) } @@ -707,7 +707,7 @@ func TestRandomFrame(t *testing.T) { } session, _ = Client(cli, nil) - f := newFrame(byte(rand.Uint32()), rand.Uint32()) + f := newFrame(1, byte(rand.Uint32()), rand.Uint32()) rnd := make([]byte, rand.Uint32()%1024) io.ReadFull(crand.Reader, rnd) f.data = rnd @@ -731,7 +731,7 @@ func TestRandomFrame(t *testing.T) { //close first session.Close() for i := 0; i < 100; i++ { - f := newFrame(byte(rand.Uint32()), rand.Uint32()) + f := newFrame(1, byte(rand.Uint32()), rand.Uint32()) session.writeFrame(f) } } @@ -760,7 +760,7 @@ func TestWriteFrameInternal(t *testing.T) { //close first session.Close() for i := 0; i < 100; i++ { - f := newFrame(byte(rand.Uint32()), rand.Uint32()) + f := newFrame(1, byte(rand.Uint32()), rand.Uint32()) session.writeFrameInternal(f, time.After(session.config.KeepAliveTimeout), 0) } @@ -772,14 +772,14 @@ func TestWriteFrameInternal(t *testing.T) { allcmds := []byte{cmdSYN, cmdFIN, cmdPSH, cmdNOP} session, _ = Client(cli, nil) for i := 0; i < 100; i++ { - f := newFrame(allcmds[rand.Int()%len(allcmds)], rand.Uint32()) + f := newFrame(1, allcmds[rand.Int()%len(allcmds)], rand.Uint32()) session.writeFrameInternal(f, time.After(session.config.KeepAliveTimeout), 0) } //deadline occur { c := make(chan time.Time) close(c) - f := newFrame(allcmds[rand.Int()%len(allcmds)], rand.Uint32()) + f := newFrame(1, allcmds[rand.Int()%len(allcmds)], rand.Uint32()) _, err := session.writeFrameInternal(f, c, 0) if !strings.Contains(err.Error(), "timeout") { t.Fatal("write frame with deadline failed", err) @@ -796,7 +796,7 @@ func TestWriteFrameInternal(t *testing.T) { config.KeepAliveInterval = time.Second config.KeepAliveTimeout = 2 * time.Second session, _ = Client(&blockWriteConn{cli}, config) - f := newFrame(byte(rand.Uint32()), rand.Uint32()) + f := newFrame(1, byte(rand.Uint32()), rand.Uint32()) c := make(chan time.Time) go func() { //die first, deadline second, better for coverage diff --git a/stream.go b/stream.go index 35e3a3c..7c487ad 100644 --- a/stream.go +++ b/stream.go @@ -175,7 +175,7 @@ func (s *Stream) Write(b []byte) (n int, err error) { // frame split and transmit sent := 0 - frame := newFrame(cmdPSH, s.id) + frame := newFrame(s.sess.config.Version, cmdPSH, s.id) bts := b for len(bts) > 0 { sz := len(bts) @@ -205,7 +205,7 @@ func (s *Stream) Close() error { }) if once { - _, err = s.sess.writeFrame(newFrame(cmdFIN, s.id)) + _, err = s.sess.writeFrame(newFrame(s.sess.config.Version, cmdFIN, s.id)) s.sess.streamClosed(s.id) return err } else {