Merge pull request #50 from libp2p/fix/reset-closed

fix: reset a stream even if closed remotely

multiplex_test.go

@@ -8,6 +8,8 @@ import (
 	"net"
 	"testing"
 	"time"
+
+	streammux "github.com/libp2p/go-stream-muxer"
 )
 
 func init() {
@@ -362,6 +364,38 @@ func TestReset(t *testing.T) {
 	}
 }
 
+func TestResetAfterEOF(t *testing.T) {
+	a, b := net.Pipe()
+
+	mpa := NewMultiplex(a, false)
+	mpb := NewMultiplex(b, true)
+
+	defer mpa.Close()
+	defer mpb.Close()
+
+	sa, err := mpa.NewStream()
+	if err != nil {
+		t.Fatal(err)
+	}
+	sb, err := mpb.Accept()
+
+	if err := sa.Close(); err != nil {
+		t.Fatal(err)
+	}
+
+	n, err := sb.Read([]byte{0})
+	if n != 0 || err != io.EOF {
+		t.Fatal(err)
+	}
+
+	sb.Reset()
+
+	n, err = sa.Read([]byte{0})
+	if n != 0 || err != streammux.ErrReset {
+		t.Fatal(err)
+	}
+}
+
 func TestOpenAfterClose(t *testing.T) {
 	a, b := net.Pipe()
 

stream.go

@@ -203,22 +203,27 @@ func (s *Stream) Close() error {
 
 func (s *Stream) Reset() error {
 	s.clLock.Lock()
-	isClosed := s.isClosed()
-	if s.closedRemote && isClosed {
+
+	// Don't reset when fully closed.
+	if s.closedRemote && s.isClosed() {
 		s.clLock.Unlock()
 		return nil
 	}
 
-	if !s.closedRemote {
-		close(s.reset)
-		// We generally call this to tell the other side to go away. No point in waiting around.
-		go s.mp.sendMsg(context.Background(), s.id.header(resetTag), nil)
+	// Don't reset twice.
+	select {
+	case <-s.reset:
+		s.clLock.Unlock()
+		return nil
+	default:
 	}
 
+	close(s.reset)
 	s.doCloseLocal()
-
 	s.closedRemote = true
 
+	go s.mp.sendMsg(context.Background(), s.id.header(resetTag), nil)
+
 	s.clLock.Unlock()
 
 	s.mp.chLock.Lock()