upgrade deps + interoperable uvarint delimited writer/reader.

comm.go

@@ -9,10 +9,11 @@ import (
 	"github.com/libp2p/go-libp2p-core/network"
 	"github.com/libp2p/go-libp2p-core/peer"
 
-	ggio "github.com/gogo/protobuf/io"
-	proto "github.com/gogo/protobuf/proto"
 	pb "github.com/libp2p/go-libp2p-pubsub/pb"
 
+	"github.com/libp2p/go-msgio/protoio"
+
+	"github.com/gogo/protobuf/proto"
 	ms "github.com/multiformats/go-multistream"
 )
 
@@ -41,7 +42,7 @@ func (p *PubSub) getHelloPacket() *RPC {
 }
 
 func (p *PubSub) handleNewStream(s network.Stream) {
-	r := ggio.NewDelimitedReader(s, p.maxMessageSize)
+	r := protoio.NewDelimitedReader(s, p.maxMessageSize)
 	for {
 		rpc := new(RPC)
 		err := r.ReadMsg(&rpc.RPC)
@@ -96,7 +97,7 @@ func (p *PubSub) handleNewPeer(ctx context.Context, pid peer.ID, outgoing <-chan
 }
 
 func (p *PubSub) handlePeerEOF(ctx context.Context, s network.Stream) {
-	r := ggio.NewDelimitedReader(s, p.maxMessageSize)
+	r := protoio.NewDelimitedReader(s, p.maxMessageSize)
 	rpc := new(RPC)
 	for {
 		err := r.ReadMsg(&rpc.RPC)
@@ -113,7 +114,7 @@ func (p *PubSub) handlePeerEOF(ctx context.Context, s network.Stream) {
 
 func (p *PubSub) handleSendingMessages(ctx context.Context, s network.Stream, outgoing <-chan *RPC) {
 	bufw := bufio.NewWriter(s)
-	wc := ggio.NewDelimitedWriter(bufw)
+	wc := protoio.NewDelimitedWriter(bufw)
 
 	writeMsg := func(msg proto.Message) error {
 		err := wc.WriteMsg(msg)

floodsub_test.go

@@ -23,7 +23,7 @@ import (
 	bhost "github.com/libp2p/go-libp2p-blankhost"
 	swarmt "github.com/libp2p/go-libp2p-swarm/testing"
 
-	ggio "github.com/gogo/protobuf/io"
+	"github.com/libp2p/go-msgio/protoio"
 )
 
 func checkMessageRouting(t *testing.T, topic string, pubs []*PubSub, subs []*Subscription) {
@@ -1057,7 +1057,7 @@ type announceWatcher struct {
 func (aw *announceWatcher) handleStream(s network.Stream) {
 	defer s.Close()
 
-	r := ggio.NewDelimitedReader(s, 1<<20)
+	r := protoio.NewDelimitedReader(s, 1<<20)
 
 	var rpc pb.RPC
 	for {

go.mod

 module github.com/libp2p/go-libp2p-pubsub
 
+go 1.13
+
 require (
-	github.com/benbjohnson/clock v1.0.2
+	github.com/benbjohnson/clock v1.0.3
 	github.com/gogo/protobuf v1.3.1
 	github.com/ipfs/go-log v1.0.4
-	github.com/libp2p/go-eventbus v0.2.1 // indirect
-	github.com/libp2p/go-libp2p-blankhost v0.1.6
+	github.com/libp2p/go-libp2p-blankhost v0.2.0
 	github.com/libp2p/go-libp2p-connmgr v0.2.4
-	github.com/libp2p/go-libp2p-core v0.5.7
-	github.com/libp2p/go-libp2p-discovery v0.4.0
-	github.com/libp2p/go-libp2p-swarm v0.2.4
+	github.com/libp2p/go-libp2p-core v0.6.1
+	github.com/libp2p/go-libp2p-discovery v0.5.0
+	github.com/libp2p/go-libp2p-swarm v0.2.8
+	github.com/libp2p/go-msgio v0.0.6
 	github.com/multiformats/go-multiaddr v0.2.2
 	github.com/multiformats/go-multiaddr-net v0.1.5
-	github.com/multiformats/go-multibase v0.0.2 // indirect
 	github.com/multiformats/go-multistream v0.1.1
 	github.com/whyrusleeping/timecache v0.0.0-20160911033111-cfcb2f1abfee
-	go.uber.org/zap v1.15.0 // indirect
-	golang.org/x/crypto v0.0.0-20200510223506-06a226fb4e37 // indirect
-	golang.org/x/sys v0.0.0-20200509044756-6aff5f38e54f // indirect
 )
-
-go 1.13

gossipsub_spam_test.go

@@ -8,13 +8,15 @@ import (
 	"testing"
 	"time"
 
-	ggio "github.com/gogo/protobuf/io"
-	logging "github.com/ipfs/go-log"
 	"github.com/libp2p/go-libp2p-core/host"
 	"github.com/libp2p/go-libp2p-core/network"
 	"github.com/libp2p/go-libp2p-core/peer"
 	"github.com/libp2p/go-libp2p-core/protocol"
+
 	pb "github.com/libp2p/go-libp2p-pubsub/pb"
+
+	logging "github.com/ipfs/go-log"
+	"github.com/libp2p/go-msgio/protoio"
 )
 
 // Test that when Gossipsub receives too many IWANT messages from a peer
@@ -718,8 +720,8 @@ func newMockGS(ctx context.Context, t *testing.T, attacker host.Host, onReadMsg
 			t.Fatal(err)
 		}
 
-		r := ggio.NewDelimitedReader(stream, maxMessageSize)
-		w := ggio.NewDelimitedWriter(ostream)
+		r := protoio.NewDelimitedReader(stream, maxMessageSize)
+		w := protoio.NewDelimitedWriter(ostream)
 
 		var irpc pb.RPC
 

gossipsub_test.go

@@ -21,7 +21,7 @@ import (
 	bhost "github.com/libp2p/go-libp2p-blankhost"
 	swarmt "github.com/libp2p/go-libp2p-swarm/testing"
 
-	ggio "github.com/gogo/protobuf/io"
+	"github.com/libp2p/go-msgio/protoio"
 )
 
 func getGossipsub(ctx context.Context, h host.Host, opts ...Option) *PubSub {
@@ -1611,8 +1611,8 @@ func (sq *sybilSquatter) handleStream(s network.Stream) {
 
 	// send a subscription for test in the output stream to become candidate for GRAFT
 	// and then just read and ignore the incoming RPCs
-	r := ggio.NewDelimitedReader(s, 1<<20)
-	w := ggio.NewDelimitedWriter(os)
+	r := protoio.NewDelimitedReader(s, 1<<20)
+	w := protoio.NewDelimitedWriter(os)
 	truth := true
 	topic := "test"
 	err = w.WriteMsg(&pb.RPC{Subscriptions: []*pb.RPC_SubOpts{&pb.RPC_SubOpts{Subscribe: &truth, Topicid: &topic}}})
@@ -1792,8 +1792,8 @@ func (iwe *iwantEverything) handleStream(s network.Stream) {
 	gossipMsgIdsReceived := make(map[string]struct{})
 
 	// send a subscription for test in the output stream to become candidate for gossip
-	r := ggio.NewDelimitedReader(s, 1<<20)
-	w := ggio.NewDelimitedWriter(os)
+	r := protoio.NewDelimitedReader(s, 1<<20)
+	w := protoio.NewDelimitedWriter(os)
 	truth := true
 	topic := "test"
 	err = w.WriteMsg(&pb.RPC{Subscriptions: []*pb.RPC_SubOpts{&pb.RPC_SubOpts{Subscribe: &truth, Topicid: &topic}}})

trace_test.go

@@ -20,7 +20,7 @@ import (
 	bhost "github.com/libp2p/go-libp2p-blankhost"
 	swarmt "github.com/libp2p/go-libp2p-swarm/testing"
 
-	ggio "github.com/gogo/protobuf/io"
+	"github.com/libp2p/go-msgio/protoio"
 )
 
 func testWithTracer(t *testing.T, tracer EventTracer) {
@@ -244,7 +244,7 @@ func TestPBTracer(t *testing.T) {
 	}
 	defer f.Close()
 
-	r := ggio.NewDelimitedReader(f, 1<<20)
+	r := protoio.NewDelimitedReader(f, 1<<20)
 	for {
 		evt.Reset()
 		err := r.ReadMsg(&evt)
@@ -271,7 +271,7 @@ func (mrt *mockRemoteTracer) handleStream(s network.Stream) {
 		panic(err)
 	}
 
-	r := ggio.NewDelimitedReader(gzr, 1<<24)
+	r := protoio.NewDelimitedReader(gzr, 1<<24)
 
 	var batch pb.TraceEventBatch
 	for {

tracer.go

@@ -18,7 +18,7 @@ import (
 	"github.com/libp2p/go-libp2p-core/peerstore"
 	"github.com/libp2p/go-libp2p-core/protocol"
 
-	ggio "github.com/gogo/protobuf/io"
+	"github.com/libp2p/go-msgio/protoio"
 )
 
 var TraceBufferSize = 1 << 16 // 64K ought to be enough for everyone; famous last words.
@@ -154,7 +154,7 @@ func OpenPBTracer(file string, flags int, perm os.FileMode) (*PBTracer, error) {
 
 func (t *PBTracer) doWrite() {
 	var buf []*pb.TraceEvent
-	w := ggio.NewDelimitedWriter(t.w)
+	w := protoio.NewDelimitedWriter(t.w)
 	for {
 		_, ok := <-t.ch
 
@@ -211,7 +211,7 @@ func (t *RemoteTracer) doWrite() {
 	var batch pb.TraceEventBatch
 
 	gzipW := gzip.NewWriter(s)
-	w := ggio.NewDelimitedWriter(gzipW)
+	w := protoio.NewDelimitedWriter(gzipW)
 
 	for {
 		_, ok := <-t.ch

validation_test.go

@@ -15,7 +15,7 @@ import (
 	"github.com/libp2p/go-libp2p-core/network"
 	"github.com/libp2p/go-libp2p-core/peer"
 
-	ggio "github.com/gogo/protobuf/io"
+	"github.com/libp2p/go-msgio/protoio"
 )
 
 func TestRegisterUnregisterValidator(t *testing.T) {
@@ -731,7 +731,7 @@ func (p *multiTopicPublisher) handleStream(s network.Stream) {
 	p.out = append(p.out, os)
 	p.mx.Unlock()
 
-	r := ggio.NewDelimitedReader(s, 1<<20)
+	r := protoio.NewDelimitedReader(s, 1<<20)
 	var rpc pb.RPC
 	for {
 		rpc.Reset()
@@ -761,7 +761,7 @@ func (p *multiTopicPublisher) publish(msg string, topics ...string) {
 	p.mx.Lock()
 	defer p.mx.Unlock()
 	for _, os := range p.out {
-		w := ggio.NewDelimitedWriter(os)
+		w := protoio.NewDelimitedWriter(os)
 		err := w.WriteMsg(rpc)
 		if err != nil {
 			panic(err)