Merge pull request #81 from libp2p/feat/rw-close

feat: use new stream interfaces from go-libp2p-core 0.7.0

fetch.go

@@ -5,7 +5,6 @@ import (
 	"errors"
 	"time"
 
-	"github.com/libp2p/go-libp2p-core/helpers"
 	"github.com/libp2p/go-libp2p-core/host"
 	"github.com/libp2p/go-libp2p-core/network"
 	"github.com/libp2p/go-libp2p-core/peer"
@@ -38,7 +37,7 @@ func newFetchProtocol(ctx context.Context, host host.Host, getData getValue) *fe
 }
 
 func (p *fetchProtocol) receive(s network.Stream, getData getValue) {
-	defer helpers.FullClose(s)
+	defer s.Close()
 
 	msg := &pb.FetchRequest{}
 	if err := readMsg(p.ctx, s, msg); err != nil {
@@ -57,6 +56,7 @@ func (p *fetchProtocol) receive(s network.Stream, getData getValue) {
 	}
 
 	if err := writeMsg(p.ctx, s, &respProto); err != nil {
+		s.Reset()
 		return
 	}
 }
@@ -69,17 +69,23 @@ func (p *fetchProtocol) Fetch(ctx context.Context, pid peer.ID, key string) ([]b
 	if err != nil {
 		return nil, err
 	}
-	defer helpers.FullClose(s)
+	defer s.Close()
 
 	msg := &pb.FetchRequest{Identifier: key}
 
 	if err := writeMsg(ctx, s, msg); err != nil {
+		_ = s.Reset()
+		return nil, err
+	}
+
+	if err := s.CloseWrite(); err != nil {
+		_ = s.Reset()
 		return nil, err
 	}
-	s.Close()
 
 	response := &pb.FetchResponse{}
 	if err := readMsg(ctx, s, response); err != nil {
+		_ = s.Reset()
 		return nil, err
 	}
 
@@ -114,7 +120,6 @@ func writeMsg(ctx context.Context, s network.Stream, msg proto.Message) error {
 	}
 
 	if retErr != nil {
-		s.Reset()
 		log.Infof("error writing response to %s: %s", s.Conn().RemotePeer(), retErr)
 	}
 	return retErr

go.mod

@@ -8,10 +8,10 @@ require (
 	github.com/ipfs/go-ipfs-ds-help v0.1.1
 	github.com/ipfs/go-log/v2 v2.1.1
 	github.com/libp2p/go-libp2p-blankhost v0.2.0
-	github.com/libp2p/go-libp2p-core v0.6.0
-	github.com/libp2p/go-libp2p-pubsub v0.3.2
+	github.com/libp2p/go-libp2p-core v0.7.0
+	github.com/libp2p/go-libp2p-pubsub v0.4.0
 	github.com/libp2p/go-libp2p-record v0.1.3
-	github.com/libp2p/go-libp2p-swarm v0.2.8
+	github.com/libp2p/go-libp2p-swarm v0.3.1
 	github.com/libp2p/go-msgio v0.0.6
 	golang.org/x/sync v0.0.0-20200317015054-43a5402ce75a
 )