Respect mux.ErrReset

conn.go

@@ -0,0 +1,41 @@
+package sm_yamux
+
+import (
+	"github.com/libp2p/go-libp2p-core/mux"
+	"github.com/libp2p/go-yamux"
+)
+
+// conn implements mux.MuxedConn over yamux.Session.
+type conn yamux.Session
+
+// Close closes underlying yamux
+func (c *conn) Close() error {
+	return c.yamux().Close()
+}
+
+// IsClosed checks if yamux.Session is in closed state.
+func (c *conn) IsClosed() bool {
+	return c.yamux().IsClosed()
+}
+
+// OpenStream creates a new stream.
+func (c *conn) OpenStream() (mux.MuxedStream, error) {
+	s, err := c.yamux().OpenStream()
+	if err != nil {
+		return nil, err
+	}
+
+	return (*stream)(s), nil
+}
+
+// AcceptStream accepts a stream opened by the other side.
+func (c *conn) AcceptStream() (mux.MuxedStream, error) {
+	s, err := c.yamux().AcceptStream()
+	return (*stream)(s), err
+}
+
+func (c *conn) yamux() *yamux.Session {
+	return (*yamux.Session)(c)
+}
+
+var _ mux.MuxedConn = &conn{}

go.mod

@@ -1,7 +1,9 @@
 module github.com/libp2p/go-libp2p-yamux
 
+go 1.13
+
 require (
 	github.com/libp2p/go-libp2p-core v0.3.0
 	github.com/libp2p/go-libp2p-testing v0.1.1
-	github.com/libp2p/go-yamux v1.2.4
+	github.com/libp2p/go-yamux v1.3.0
 )

go.sum

@@ -81,6 +81,8 @@ github.com/libp2p/go-yamux v1.2.3 h1:xX8A36vpXb59frIzWFdEgptLMsOANMFq2K7fPRlunYI
 github.com/libp2p/go-yamux v1.2.3/go.mod h1:FGTiPvoV/3DVdgWpX+tM0OW3tsM+W5bSE3gZwqQTcow=
 github.com/libp2p/go-yamux v1.2.4 h1:tSkZdyEEwA++MeJ+r3bxIVpwsF6ygPmOs3xaIjrgJCw=
 github.com/libp2p/go-yamux v1.2.4/go.mod h1:FGTiPvoV/3DVdgWpX+tM0OW3tsM+W5bSE3gZwqQTcow=
+github.com/libp2p/go-yamux v1.3.0 h1:FsYzT16Wq2XqUGJsBbOxoz9g+dFklvNi7jN6YFPfl7U=
+github.com/libp2p/go-yamux v1.3.0/go.mod h1:FGTiPvoV/3DVdgWpX+tM0OW3tsM+W5bSE3gZwqQTcow=
 github.com/mailru/easyjson v0.0.0-20180823135443-60711f1a8329 h1:2gxZ0XQIU/5z3Z3bUBu+FXuk2pFbkN6tcwi/pjyaDic=
 github.com/mailru/easyjson v0.0.0-20180823135443-60711f1a8329/go.mod h1:C1wdFJiN94OJF2b5HbByQZoLdCWB1Yqtg26g4irojpc=
 github.com/mattn/go-colorable v0.0.9/go.mod h1:9vuHe8Xs5qXnSaW/c/ABM9alt+Vo+STaOChaDxuIBZU=

stream.go

@@ -0,0 +1,55 @@
+package sm_yamux
+
+import (
+	"time"
+
+	"github.com/libp2p/go-libp2p-core/mux"
+	"github.com/libp2p/go-yamux"
+)
+
+// stream implements mux.MuxedStream over yamux.Stream.
+type stream yamux.Stream
+
+func (s *stream) Read(b []byte) (n int, err error) {
+	n, err = s.yamux().Read(b)
+	if err == yamux.ErrStreamReset {
+		err = mux.ErrReset
+	}
+
+	return n, err
+}
+
+func (s *stream) Write(b []byte) (n int, err error) {
+	n, err = s.yamux().Write(b)
+	if err == yamux.ErrStreamReset {
+		err = mux.ErrReset
+	}
+
+	return n, err
+}
+
+func (s *stream) Close() error {
+	return s.yamux().Close()
+}
+
+func (s *stream) Reset() error {
+	return s.yamux().Reset()
+}
+
+func (s *stream) SetDeadline(t time.Time) error {
+	return s.yamux().SetDeadline(t)
+}
+
+func (s *stream) SetReadDeadline(t time.Time) error {
+	return s.yamux().SetReadDeadline(t)
+}
+
+func (s *stream) SetWriteDeadline(t time.Time) error {
+	return s.yamux().SetWriteDeadline(t)
+}
+
+func (s *stream) yamux() *yamux.Stream {
+	return (*yamux.Stream)(s)
+}
+
+var _ mux.MuxedStream = &stream{}

yamux.go → transport.go

@@ -8,41 +8,6 @@ import (
 	yamux "github.com/libp2p/go-yamux"
 )
 
-// Conn is a connection to a remote peer.
-type conn yamux.Session
-
-func (c *conn) yamuxSession() *yamux.Session {
-	return (*yamux.Session)(c)
-}
-
-func (c *conn) Close() error {
-	return c.yamuxSession().Close()
-}
-
-func (c *conn) IsClosed() bool {
-	return c.yamuxSession().IsClosed()
-}
-
-// OpenStream creates a new stream.
-func (c *conn) OpenStream() (mux.MuxedStream, error) {
-	s, err := c.yamuxSession().OpenStream()
-	if err != nil {
-		return nil, err
-	}
-
-	return s, nil
-}
-
-// AcceptStream accepts a stream opened by the other side.
-func (c *conn) AcceptStream() (mux.MuxedStream, error) {
-	s, err := c.yamuxSession().AcceptStream()
-	return s, err
-}
-
-// Transport is a go-peerstream transport that constructs
-// yamux-backed connections.
-type Transport yamux.Config
-
 var DefaultTransport *Transport
 
 func init() {
@@ -61,6 +26,10 @@ func init() {
 	DefaultTransport = (*Transport)(config)
 }
 
+// Transport implements mux.Multiplexer that constructs
+// yamux-backed muxed connections.
+type Transport yamux.Config
+
 func (t *Transport) NewConn(nc net.Conn, isServer bool) (mux.MuxedConn, error) {
 	var s *yamux.Session
 	var err error
@@ -75,3 +44,5 @@ func (t *Transport) NewConn(nc net.Conn, isServer bool) (mux.MuxedConn, error) {
 func (t *Transport) Config() *yamux.Config {
 	return (*yamux.Config)(t)
 }
+
+var _ mux.Multiplexer = &Transport{}

yamux_test.go → transport_test.go

@@ -6,6 +6,6 @@ import (
 	tmux "github.com/libp2p/go-libp2p-testing/suites/mux"
 )
 
-func TestYamuxTransport(t *testing.T) {
+func TestDefaultTransport(t *testing.T) {
 	tmux.SubtestAll(t, DefaultTransport)
 }