feat : autoshard relay api

cmd/waku/node.go

@@ -344,10 +344,10 @@ func Execute(options NodeOptions) {
 	if options.Relay.Enable {
 		for nodeTopic := range pubSubTopicMap {
 			nodeTopic := nodeTopic
-			sub, err := wakuNode.Relay().SubscribeToTopic(ctx, nodeTopic)
+			sub, err := wakuNode.Relay().Subscribe(ctx, wprotocol.NewContentFilter(nodeTopic))
 			failOnErr(err, "Error subscring to topic")
-			sub.Unsubscribe()
-
+			//Calling unsubscribe only closes broadcaster sub and not the underlying pubSub subscription.
+			sub[0].Unsubscribe()
 			if len(options.Rendezvous.Nodes) != 0 {
 				// Register the node in rendezvous point
 				iter := rendezvous.NewRendezvousPointIterator(options.Rendezvous.Nodes)

cmd/waku/server/rest/relay.go

@@ -107,7 +107,7 @@ func (r *RelayService) deleteV1Subscriptions(w http.ResponseWriter, req *http.Re
 
 	var err error
 	for _, topic := range topics {
-		err = r.node.Relay().Unsubscribe(req.Context(), topic)
+		err = r.node.Relay().Unsubscribe(req.Context(), protocol.NewContentFilter(topic))
 		if err != nil {
 			r.log.Error("unsubscribing from topic", zap.String("topic", strings.Replace(strings.Replace(topic, "\n", "", -1), "\r", "", -1)), zap.Error(err))
 		} else {
@@ -129,18 +129,20 @@ func (r *RelayService) postV1Subscriptions(w http.ResponseWriter, req *http.Requ
 
 	var err error
 	var sub *relay.Subscription
+	var subs []*relay.Subscription
 	var topicToSubscribe string
 	for _, topic := range topics {
 		if topic == "" {
-			sub, err = r.node.Relay().Subscribe(req.Context())
+			subs, err = r.node.Relay().Subscribe(req.Context(), protocol.NewContentFilter(relay.DefaultWakuTopic))
 			topicToSubscribe = relay.DefaultWakuTopic
 		} else {
-			sub, err = r.node.Relay().SubscribeToTopic(req.Context(), topic)
+			subs, err = r.node.Relay().Subscribe(req.Context(), protocol.NewContentFilter(topic))
 			topicToSubscribe = topic
 		}
 		if err != nil {
 			r.log.Error("subscribing to topic", zap.String("topic", strings.Replace(topicToSubscribe, "\n", "", -1)), zap.Error(err))
 		} else {
+			sub = subs[0]
 			sub.Unsubscribe()
 			r.messagesMutex.Lock()
 			r.messages[topic] = []*pb.WakuMessage{}

cmd/waku/server/rest/runner.go

@@ -11,7 +11,7 @@ type Adder func(msg *protocol.Envelope)
 
 type runnerService struct {
 	broadcaster relay.Broadcaster
-	sub         relay.Subscription
+	sub         *relay.Subscription
 	cancel      context.CancelFunc
 	adder       Adder
 }

cmd/waku/server/rpc/filter_test.go

@@ -13,6 +13,7 @@ import (
 	"github.com/waku-org/go-waku/tests"
 	"github.com/waku-org/go-waku/waku/v2/node"
 	"github.com/waku-org/go-waku/waku/v2/peerstore"
+	"github.com/waku-org/go-waku/waku/v2/protocol"
 	"github.com/waku-org/go-waku/waku/v2/protocol/legacy_filter"
 	"github.com/waku-org/go-waku/waku/v2/protocol/legacy_filter/pb"
 	wpb "github.com/waku-org/go-waku/waku/v2/protocol/pb"
@@ -37,9 +38,9 @@ func makeFilterService(t *testing.T, isFullNode bool) *FilterService {
 	require.NoError(t, err)
 
 	if isFullNode {
-		sub, err := n.Relay().SubscribeToTopic(context.Background(), testTopic)
+		sub, err := n.Relay().Subscribe(context.Background(), protocol.NewContentFilter(testTopic))
 		go func() {
-			for range sub.Ch {
+			for range sub[0].Ch {
 			}
 		}()
 		require.NoError(t, err)
@@ -62,14 +63,15 @@ func TestFilterSubscription(t *testing.T) {
 	err = node.Start(context.Background())
 	require.NoError(t, err)
 
-	_, err = node.SubscribeToTopic(context.Background(), testTopic)
+	_, err = node.Subscribe(context.Background(), protocol.NewContentFilter(testTopic))
 	require.NoError(t, err)
 
 	b2 := relay.NewBroadcaster(10)
 	require.NoError(t, b2.Start(context.Background()))
 	f := legacy_filter.NewWakuFilter(b2, false, timesource.NewDefaultClock(), prometheus.DefaultRegisterer, utils.Logger())
 	f.SetHost(host)
-	err = f.Start(context.Background(), relay.NoopSubscription())
+	sub := relay.NewSubscription(protocol.NewContentFilter(relay.DefaultWakuTopic))
+	err = f.Start(context.Background(), sub)
 	require.NoError(t, err)
 
 	d := makeFilterService(t, true)

cmd/waku/server/rpc/relay.go

@@ -127,15 +127,21 @@ func (r *RelayService) PostV1Subscription(req *http.Request, args *TopicsArgs, r
 		var err error
 		if topic == "" {
 			var sub *relay.Subscription
-			sub, err = r.node.Relay().Subscribe(ctx)
+			subs, err := r.node.Relay().Subscribe(ctx, protocol.NewContentFilter(relay.DefaultWakuTopic))
+			if err != nil {
+				r.log.Error("subscribing to topic", zap.String("topic", topic), zap.Error(err))
+				return err
+			}
+			sub = subs[0]
 			sub.Unsubscribe()
 		} else {
 			var sub *relay.Subscription
-			sub, err = r.node.Relay().SubscribeToTopic(ctx, topic)
+			subs, err := r.node.Relay().Subscribe(ctx, protocol.NewContentFilter(topic))
 			if err != nil {
 				r.log.Error("subscribing to topic", zap.String("topic", topic), zap.Error(err))
 				return err
 			}
+			sub = subs[0]
 			sub.Unsubscribe()
 		}
 		if err != nil {
@@ -155,7 +161,7 @@ func (r *RelayService) PostV1Subscription(req *http.Request, args *TopicsArgs, r
 func (r *RelayService) DeleteV1Subscription(req *http.Request, args *TopicsArgs, reply *SuccessReply) error {
 	ctx := req.Context()
 	for _, topic := range args.Topics {
-		err := r.node.Relay().Unsubscribe(ctx, topic)
+		err := r.node.Relay().Unsubscribe(ctx, protocol.NewContentFilter(topic))
 		if err != nil {
 			r.log.Error("unsubscribing from topic", zap.String("topic", topic), zap.Error(err))
 			return err

cmd/waku/server/rpc/runner.go

@@ -9,7 +9,7 @@ type Adder func(msg *protocol.Envelope)
 
 type runnerService struct {
 	broadcaster relay.Broadcaster
-	sub         relay.Subscription
+	sub         *relay.Subscription
 	adder       Adder
 }
 

examples/basic2/main.go

@@ -16,6 +16,7 @@ import (
 	"github.com/waku-org/go-waku/waku/v2/payload"
 	"github.com/waku-org/go-waku/waku/v2/protocol"
 	"github.com/waku-org/go-waku/waku/v2/protocol/pb"
+	"github.com/waku-org/go-waku/waku/v2/protocol/relay"
 	"github.com/waku-org/go-waku/waku/v2/utils"
 	"go.uber.org/zap"
 )
@@ -116,13 +117,13 @@ func writeLoop(ctx context.Context, wakuNode *node.WakuNode, contentTopic string
 }
 
 func readLoop(ctx context.Context, wakuNode *node.WakuNode, contentTopic string) {
-	sub, err := wakuNode.Relay().Subscribe(ctx)
+	sub, err := wakuNode.Relay().Subscribe(ctx, protocol.NewContentFilter(relay.DefaultWakuTopic))
 	if err != nil {
 		log.Error("Could not subscribe", zap.Error(err))
 		return
 	}
 
-	for envelope := range sub.Ch {
+	for envelope := range sub[0].Ch {
 		if envelope.Message().ContentTopic != contentTopic {
 			continue
 		}

examples/chat2/chat.go

@@ -3,7 +3,6 @@ package main
 import (
 	"chat2/pb"
 	"context"
-	"crypto/sha256"
 	"encoding/hex"
 	"errors"
 	"fmt"
@@ -24,7 +23,6 @@ import (
 	"github.com/waku-org/go-waku/waku/v2/protocol/store"
 	"github.com/waku-org/go-waku/waku/v2/utils"
 	"github.com/waku-org/go-zerokit-rln/rln"
-	"golang.org/x/crypto/pbkdf2"
 	"google.golang.org/protobuf/proto"
 )
 
@@ -84,13 +82,13 @@ func NewChat(ctx context.Context, node *node.WakuNode, connNotifier <-chan node.
 	} else {
 
 		for _, topic := range topics {
-			sub, err := node.Relay().SubscribeToTopic(ctx, topic)
+			sub, err := node.Relay().Subscribe(ctx, protocol.NewContentFilter(topic))
 			if err != nil {
 				chat.ui.ErrorMessage(err)
 			} else {
 				chat.C = make(chan *protocol.Envelope)
 				go func() {
-					for e := range sub.Ch {
+					for e := range sub[0].Ch {
 						chat.C <- e
 					}
 				}()
@@ -356,12 +354,6 @@ func decodeMessage(contentTopic string, wakumsg *wpb.WakuMessage) (*pb.Chat2Mess
 	return msg, nil
 }
 
-func generateSymKey(password string) []byte {
-	// AesKeyLength represents the length (in bytes) of an private key
-	AESKeyLength := 256 / 8
-	return pbkdf2.Key([]byte(password), nil, 65356, AESKeyLength, sha256.New)
-}
-
 func (c *Chat) retrieveHistory(connectionWg *sync.WaitGroup) {
 	defer c.wg.Done()
 

examples/filter2/main.go

@@ -172,13 +172,13 @@ func writeLoop(ctx context.Context, wakuNode *node.WakuNode) {
 
 func readLoop(ctx context.Context, wakuNode *node.WakuNode) {
 	pubsubTopic := pubSubTopic.String()
-	sub, err := wakuNode.Relay().SubscribeToTopic(ctx, pubsubTopic)
+	sub, err := wakuNode.Relay().Subscribe(ctx, protocol.NewContentFilter(pubsubTopic))
 	if err != nil {
 		log.Error("Could not subscribe: ", err)
 		return
 	}
 
-	for value := range sub.Ch {
+	for value := range sub[0].Ch {
 		payload, err := payload.DecodePayload(value.Message(), &payload.KeyInfo{Kind: payload.None})
 		if err != nil {
 			fmt.Println(err)

examples/rln/main.go

@@ -142,13 +142,13 @@ func writeLoop(ctx context.Context, wakuNode *node.WakuNode) {
 }
 
 func readLoop(ctx context.Context, wakuNode *node.WakuNode) {
-	sub, err := wakuNode.Relay().SubscribeToTopic(ctx, pubsubTopic.String())
+	sub, err := wakuNode.Relay().Subscribe(ctx, protocol.NewContentFilter(pubsubTopic.String()))
 	if err != nil {
 		log.Error("Could not subscribe", zap.Error(err))
 		return
 	}
 
-	for envelope := range sub.Ch {
+	for envelope := range sub[0].Ch {
 		if envelope.Message().ContentTopic != contentTopic.String() {
 			continue
 		}

library/relay.go

@@ -69,18 +69,18 @@ func relaySubscribe(topic string) error {
 		return nil
 	}
 
-	subscription, err := wakuState.node.Relay().SubscribeToTopic(context.Background(), topicToSubscribe)
+	subscription, err := wakuState.node.Relay().Subscribe(context.Background(), protocol.NewContentFilter(topicToSubscribe))
 	if err != nil {
 		return err
 	}
 
-	relaySubscriptions[topicToSubscribe] = subscription
+	relaySubscriptions[topicToSubscribe] = subscription[0]
 
 	go func(subscription *relay.Subscription) {
 		for envelope := range subscription.Ch {
 			send("message", toSubscriptionMessage(envelope))
 		}
-	}(subscription)
+	}(subscription[0])
 
 	return nil
 }
@@ -123,5 +123,5 @@ func RelayUnsubscribe(topic string) error {
 
 	delete(relaySubscriptions, topicToUnsubscribe)
 
-	return wakuState.node.Relay().Unsubscribe(context.Background(), topicToUnsubscribe)
+	return wakuState.node.Relay().Unsubscribe(context.Background(), protocol.NewContentFilter(topicToUnsubscribe))
 }

tests/connection_test.go

@@ -10,7 +10,9 @@ import (
 
 	"github.com/waku-org/go-waku/waku/v2/node"
 	"github.com/waku-org/go-waku/waku/v2/payload"
+	"github.com/waku-org/go-waku/waku/v2/protocol"
 	"github.com/waku-org/go-waku/waku/v2/protocol/pb"
+	"github.com/waku-org/go-waku/waku/v2/protocol/relay"
 	"github.com/waku-org/go-waku/waku/v2/utils"
 )
 
@@ -39,10 +41,10 @@ func TestBasicSendingReceiving(t *testing.T) {
 
 	require.NoError(t, write(ctx, wakuNode, "test"))
 
-	sub, err := wakuNode.Relay().Subscribe(ctx)
+	sub, err := wakuNode.Relay().Subscribe(ctx, protocol.NewContentFilter(relay.DefaultWakuTopic))
 	require.NoError(t, err)
 
-	value := <-sub.Ch
+	value := <-sub[0].Ch
 	payload, err := payload.DecodePayload(value.Message(), &payload.KeyInfo{Kind: payload.None})
 	require.NoError(t, err)
 

waku/v2/node/service.go

@@ -16,5 +16,5 @@ type Service interface {
 type ReceptorService interface {
 	SetHost(h host.Host)
 	Stop()
-	Start(context.Context, relay.Subscription) error
+	Start(context.Context, *relay.Subscription) error
 }

waku/v2/node/wakunode2.go

@@ -679,7 +679,7 @@ func (w *WakuNode) mountDiscV5() error {
 	return err
 }
 
-func (w *WakuNode) startStore(ctx context.Context, sub relay.Subscription) error {
+func (w *WakuNode) startStore(ctx context.Context, sub *relay.Subscription) error {
 	err := w.store.Start(ctx, sub)
 	if err != nil {
 		w.log.Error("starting store", zap.Error(err))

waku/v2/node/wakunode2_test.go

@@ -18,6 +18,7 @@ import (
 	"github.com/waku-org/go-waku/waku/persistence/sqlite"
 	"github.com/waku-org/go-waku/waku/v2/dnsdisc"
 	"github.com/waku-org/go-waku/waku/v2/peerstore"
+	"github.com/waku-org/go-waku/waku/v2/protocol"
 	"github.com/waku-org/go-waku/waku/v2/protocol/legacy_filter"
 	"github.com/waku-org/go-waku/waku/v2/protocol/pb"
 	"github.com/waku-org/go-waku/waku/v2/protocol/relay"
@@ -54,11 +55,11 @@ func TestWakuNode2(t *testing.T) {
 	err = wakuNode.Start(ctx)
 	require.NoError(t, err)
 
-	_, err = wakuNode.Relay().SubscribeToTopic(ctx, "waku/rs/1/1")
+	_, err = wakuNode.Relay().Subscribe(ctx, protocol.NewContentFilter("waku/rs/1/1"))
 	require.NoError(t, err)
 	time.Sleep(time.Second * 1)
 
-	err = wakuNode.Relay().Unsubscribe(ctx, "waku/rs/1/1")
+	err = wakuNode.Relay().Unsubscribe(ctx, protocol.NewContentFilter("waku/rs/1/1"))
 	require.NoError(t, err)
 
 	defer wakuNode.Stop()
@@ -151,9 +152,9 @@ func Test500(t *testing.T) {
 
 	time.Sleep(2 * time.Second)
 
-	sub1, err := wakuNode1.Relay().Subscribe(ctx)
+	sub1, err := wakuNode1.Relay().Subscribe(ctx, protocol.NewContentFilter(relay.DefaultWakuTopic))
 	require.NoError(t, err)
-	sub2, err := wakuNode1.Relay().Subscribe(ctx)
+	sub2, err := wakuNode1.Relay().Subscribe(ctx, protocol.NewContentFilter(relay.DefaultWakuTopic))
 	require.NoError(t, err)
 
 	wg := sync.WaitGroup{}
@@ -168,7 +169,7 @@ func Test500(t *testing.T) {
 			select {
 			case <-ticker.C:
 				require.Fail(t, "Timeout Sub1")
-			case msg := <-sub1.Ch:
+			case msg := <-sub1[0].Ch:
 				if msg == nil {
 					return
 				}
@@ -189,7 +190,7 @@ func Test500(t *testing.T) {
 			select {
 			case <-ticker.C:
 				require.Fail(t, "Timeout Sub2")
-			case msg := <-sub2.Ch:
+			case msg := <-sub2[0].Ch:
 				if msg == nil {
 					return
 				}
@@ -234,9 +235,9 @@ func TestDecoupledStoreFromRelay(t *testing.T) {
 	require.NoError(t, err)
 	defer wakuNode1.Stop()
 
-	subs, err := wakuNode1.Relay().Subscribe(ctx)
+	subs, err := wakuNode1.Relay().Subscribe(ctx, protocol.NewContentFilter(relay.DefaultWakuTopic))
 	require.NoError(t, err)
-	subs.Unsubscribe()
+	defer subs[0].Unsubscribe()
 
 	// NODE2: Filter Client/Store
 	db, err := sqlite.NewDB(":memory:", false, utils.Logger())

waku/v2/node/wakuoptions.go

@@ -24,7 +24,6 @@ import (
 	"github.com/libp2p/go-libp2p/p2p/transport/tcp"
 	libp2pwebtransport "github.com/libp2p/go-libp2p/p2p/transport/webtransport"
 	"github.com/multiformats/go-multiaddr"
-	ma "github.com/multiformats/go-multiaddr"
 	manet "github.com/multiformats/go-multiaddr/net"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/waku-org/go-waku/waku/v2/protocol/filter"
@@ -232,7 +231,7 @@ func WithHostAddress(hostAddr *net.TCPAddr) WakuNodeOption {
 }
 
 // WithAdvertiseAddresses is a WakuNodeOption that allows overriding the address used in the waku node with custom value
-func WithAdvertiseAddresses(advertiseAddrs ...ma.Multiaddr) WakuNodeOption {
+func WithAdvertiseAddresses(advertiseAddrs ...multiaddr.Multiaddr) WakuNodeOption {
 	return func(params *WakuNodeParameters) error {
 		params.advertiseAddrs = advertiseAddrs
 		return WithMultiaddress(advertiseAddrs...)(params)