use go-wakurelay-pubsub instead of go-libp2p-pubsub

go.mod

@@ -2,8 +2,6 @@ module github.com/status-im/go-waku
 
 go 1.15
 
-replace github.com/libp2p/go-libp2p-pubsub => github.com/status-im/go-libp2p-pubsub v0.4.1-customProtocols
-
 require (
 	github.com/cruxic/go-hmac-drbg v0.0.0-20170206035330-84c46983886d
 	github.com/ethereum/go-ethereum v1.9.5
@@ -13,7 +11,6 @@ require (
 	github.com/libp2p/go-libp2p v0.13.0
 	github.com/libp2p/go-libp2p-connmgr v0.2.4
 	github.com/libp2p/go-libp2p-core v0.8.5
-	github.com/libp2p/go-libp2p-pubsub v0.4.1
 	github.com/magiconair/properties v1.8.4 // indirect
 	github.com/mitchellh/mapstructure v1.4.1 // indirect
 	github.com/multiformats/go-multiaddr v0.3.1
@@ -24,6 +21,7 @@ require (
 	github.com/spf13/cobra v1.1.3
 	github.com/spf13/jwalterweatherman v1.1.0 // indirect
 	github.com/spf13/viper v1.7.1
+	github.com/status-im/go-wakurelay-pubsub v0.4.2
 	golang.org/x/sys v0.0.0-20210317225723-c4fcb01b228e // indirect
 	golang.org/x/text v0.3.5 // indirect
 	google.golang.org/protobuf v1.25.0

go.sum

@@ -342,6 +342,8 @@ github.com/libp2p/go-libp2p-peerstore v0.2.6 h1:2ACefBX23iMdJU9Ke+dcXt3w86MIryes
 github.com/libp2p/go-libp2p-peerstore v0.2.6/go.mod h1:ss/TWTgHZTMpsU/oKVVPQCGuDHItOpf2W8RxAi50P2s=
 github.com/libp2p/go-libp2p-pnet v0.2.0 h1:J6htxttBipJujEjz1y0a5+eYoiPcFHhSYHH6na5f0/k=
 github.com/libp2p/go-libp2p-pnet v0.2.0/go.mod h1:Qqvq6JH/oMZGwqs3N1Fqhv8NVhrdYcO0BW4wssv21LA=
+github.com/libp2p/go-libp2p-pubsub v0.4.1 h1:j4umIg5nyus+sqNfU+FWvb9aeYFQH/A+nDFhWj+8yy8=
+github.com/libp2p/go-libp2p-pubsub v0.4.1/go.mod h1:izkeMLvz6Ht8yAISXjx60XUQZMq9ZMe5h2ih4dLIBIQ=
 github.com/libp2p/go-libp2p-secio v0.1.0/go.mod h1:tMJo2w7h3+wN4pgU2LSYeiKPrfqBgkOsdiKK77hE7c8=
 github.com/libp2p/go-libp2p-secio v0.2.0/go.mod h1:2JdZepB8J5V9mBp79BmwsaPQhRPNN2NrnB2lKQcdy6g=
 github.com/libp2p/go-libp2p-secio v0.2.1/go.mod h1:cWtZpILJqkqrSkiYcDBh5lA3wbT2Q+hz3rJQq3iftD8=
@@ -622,8 +624,8 @@ github.com/spf13/viper v1.7.0/go.mod h1:8WkrPz2fc9jxqZNCJI/76HCieCp4Q8HaLFoCha5q
 github.com/spf13/viper v1.7.1 h1:pM5oEahlgWv/WnHXpgbKz7iLIxRf65tye2Ci+XFK5sk=
 github.com/spf13/viper v1.7.1/go.mod h1:8WkrPz2fc9jxqZNCJI/76HCieCp4Q8HaLFoCha5qpdg=
 github.com/src-d/envconfig v1.0.0/go.mod h1:Q9YQZ7BKITldTBnoxsE5gOeB5y66RyPXeue/R4aaNBc=
-github.com/status-im/go-libp2p-pubsub v0.4.1-customProtocols h1:xLGSO/46TTcwu0LhqchpJUvAa31gWxW49Snwr20Q108=
+github.com/status-im/go-wakurelay-pubsub v0.4.2 h1:F4UGcP80H0PGaeJ0mRMzA1Ux3DKYiyv/qu3bOR/efTg=
-github.com/status-im/go-libp2p-pubsub v0.4.1-customProtocols/go.mod h1:izkeMLvz6Ht8yAISXjx60XUQZMq9ZMe5h2ih4dLIBIQ=
+github.com/status-im/go-wakurelay-pubsub v0.4.2/go.mod h1:LSCVYR7mnBBsxVJghrGpQ3yJAAATEe6XeQQqGCZhwrE=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=

waku/v2/node/wakunode2.go

@@ -16,11 +16,11 @@ import (
 	"github.com/libp2p/go-libp2p-core/crypto"
 	"github.com/libp2p/go-libp2p-core/host"
 	"github.com/libp2p/go-libp2p-core/peer"
-	pubsub "github.com/libp2p/go-libp2p-pubsub"
 	ma "github.com/multiformats/go-multiaddr"
 	manet "github.com/multiformats/go-multiaddr-net"
 	"github.com/status-im/go-waku/waku/v2/protocol"
 	store "github.com/status-im/go-waku/waku/v2/protocol/waku_store"
+	wakurelay "github.com/status-im/go-wakurelay-pubsub"
 )
 
 var log = logging.Logger("wakunode")
@@ -49,16 +49,16 @@ type Subscription struct {
 	C               chan *protocol.WakuMessage
 	closed          bool
 	mutex           sync.Mutex
-	pubSubscription *pubsub.Subscription
+	pubSubscription *wakurelay.Subscription
 	quit            chan struct{}
 }
 
 type WakuNode struct {
 	host   host.Host
-	pubsub *pubsub.PubSub
+	pubsub *wakurelay.PubSub
 	store  *store.WakuStore
 
-	topics      map[Topic]*pubsub.Topic
+	topics      map[Topic]*wakurelay.Topic
 	topicsMutex sync.Mutex
 
 	subscriptions      []*Subscription
@@ -115,7 +115,7 @@ func New(ctx context.Context, privKey *ecdsa.PrivateKey, hostAddr net.Addr, extA
 	w.cancel = cancel
 	w.privKey = nodeKey
 	w.ctx = ctx
-	w.topics = make(map[Topic]*pubsub.Topic)
+	w.topics = make(map[Topic]*wakurelay.Topic)
 
 	hostInfo, _ := ma.NewMultiaddr(fmt.Sprintf("/p2p/%s", host.ID().Pretty()))
 	for _, addr := range host.Addrs() {
@@ -146,16 +146,16 @@ func (w *WakuNode) ID() string {
 	return w.host.ID().Pretty()
 }
 
-func (w *WakuNode) PubSub() *pubsub.PubSub {
+func (w *WakuNode) PubSub() *wakurelay.PubSub {
 	return w.pubsub
 }
 
-func (w *WakuNode) SetPubSub(pubSub *pubsub.PubSub) {
+func (w *WakuNode) SetPubSub(pubSub *wakurelay.PubSub) {
 	w.pubsub = pubSub
 }
 
 func (w *WakuNode) MountRelay() error {
-	ps, err := protocol.NewWakuRelay(w.ctx, w.host)
+	ps, err := wakurelay.NewWakuRelaySub(w.ctx, w.host)
 	if err != nil {
 		return err
 	}
@@ -250,7 +250,7 @@ func (node *WakuNode) Subscribe(topic *Topic) (*Subscription, error) {
 	subscription.C = make(chan *protocol.WakuMessage)
 	subscription.quit = make(chan struct{})
 
-	go func(ctx context.Context, sub *pubsub.Subscription) {
+	go func(ctx context.Context, sub *wakurelay.Subscription) {
 		nextMsgTicker := time.NewTicker(time.Millisecond * 10)
 		defer nextMsgTicker.Stop()
 
@@ -304,7 +304,7 @@ func (subs *Subscription) IsClosed() bool {
 	return subs.closed
 }
 
-func (node *WakuNode) upsertTopic(topic *Topic) (*pubsub.Topic, error) {
+func (node *WakuNode) upsertTopic(topic *Topic) (*wakurelay.Topic, error) {
 	defer node.topicsMutex.Unlock()
 	node.topicsMutex.Lock()
 

waku/v2/protocol/waku_relay.go

@@ -1,43 +0,0 @@
-// Waku Relay module. Thin layer on top of GossipSub.
-//
-// See https://github.com/vacp2p/specs/blob/master/specs/waku/v2/waku-relay.md
-// for spec.
-package protocol
-
-import (
-	"context"
-	"crypto/sha256"
-
-	"github.com/libp2p/go-libp2p-core/host"
-	libp2pProtocol "github.com/libp2p/go-libp2p-core/protocol"
-	pubsub "github.com/libp2p/go-libp2p-pubsub"
-	pubsub_pb "github.com/libp2p/go-libp2p-pubsub/pb"
-)
-
-const WakuRelayProtocol = libp2pProtocol.ID("/vac/waku/relay/2.0.0-beta2")
-
-type WakuRelay struct {
-	p *pubsub.PubSub
-}
-
-// Once https://github.com/status-im/nim-waku/issues/420 is fixed, implement a custom messageIdFn
-func msgIdFn(pmsg *pubsub_pb.Message) string {
-	hash := sha256.Sum256(pmsg.Data)
-	return string(hash[:])
-}
-
-func NewWakuRelay(ctx context.Context, h host.Host, opts ...pubsub.Option) (*pubsub.PubSub, error) {
-	opts = append(opts, pubsub.WithMessageSignaturePolicy(pubsub.StrictNoSign))
-	opts = append(opts, pubsub.WithNoAuthor())
-	opts = append(opts, pubsub.WithMessageIdFn(msgIdFn))
-
-	gossipSub, err := pubsub.NewGossipSub(ctx, h, []libp2pProtocol.ID{WakuRelayProtocol}, opts...)
-
-	if err != nil {
-		return nil, err
-	}
-
-	w := new(WakuRelay)
-	w.p = gossipSub
-	return gossipSub, nil
-}