Feat/autoshard filter

examples/chat2/chat.go

@@ -75,11 +75,11 @@ func NewChat(ctx context.Context, node *node.WakuNode, connNotifier <-chan node.
 			filterOpt = filter.WithPeer(peerID)
 			chat.ui.InfoMessage(fmt.Sprintf("Subscribing to filter node %s", peerID))
 		}
-		theFilter, err := node.FilterLightnode().Subscribe(ctx, cf, filterOpt)
+		theFilters, err := node.FilterLightnode().Subscribe(ctx, cf, filterOpt)
 		if err != nil {
 			chat.ui.ErrorMessage(err)
 		} else {
-			chat.C = theFilter.C
+			chat.C = theFilters[0].C //Picking first subscription since there is only 1 contentTopic specified.
 		}
 	} else {
 

examples/filter2/README.md

@@ -21,28 +21,20 @@ The app will run 2 nodes ("full" node and "light" node), with light node subscri
 
 ## Flow description
 
-1. Light node submits a FilterRequest through WakuNode.SubscribeFilter. This request is submitted to a particular peer.
-Filter is stored in WakuNode.filters map. That's it.
-DONE
-2. Full node: we read incoming messages in WakuFilter.onRequest(). It is set as a stream handler on wakunode.Host for WakuFilterProtocolId.
-3. In WakuFilter.onRequest():
-  3.1. We check whether it's a MessagePush or FilterRequest.
-  3.2. If it's a MessagePush, then we're on a light node. Invoke pushHandler coming from WakuNode.mountFilter()
-  3.3. If it's a FilterRequest, add a subscriber.
-4. WakuNode.Subscribe has a message loop extracting WakuMessages from a wakurelay.Subscription object.
-It denotes a pubsub topic subscription.
-All envelopes are then submitted to node.broadcaster.
-
-
-## Nim code flow
-1. Light node: WakuFilter.subscribe(). Find a peer, wrileLP(FilterRequest). Store requestId in WakuNode.filters along with a ContentFilterHandler proc.
-2. Full node: WakuFilter inherits LPProtocol. LPProtocol.handler invokes readLP() to read FilterRPC messages
-3. this handler function has a signature (conn: Connection, proto: string).
-  3.1. it checks whether a MessagePush or FilterRequest is received.
-  3.2. (light node) if it's a MessagePush, then we're on a light node. Invoke pushHandler of MessagePushHandler type. This pushHandler comes from WakuNode.mountFilter(). It iterates through all registered WakuNode.filters (stored in step 1) and invokes their ContentFilterHandler proc.
-  3.3. (full node) if it's a FilterRequest, create a Subscriber and add to WakuFilter.subscribers seq
-4. (full node) Each time a message is received through GossipSub in wakunode.subscribe.defaultHandler(), we iterate through subscriptions.
-5. (full node) One of these subscriptions is a filter subscription added by WakuNode.mountFilter(), which in turn is returned from WakuFilter.subscription()
-6. (full node) This subscription iterates through subscribers added by WakuFilter.handler() fn (subscribers being light nodes)
-7. (full node) Once subscriber peer is found, a message is pushed directly to the peer (go to step 3.2)
-
+### Light Node
+1. A light node is created with option WithWakuFilterLightNode.
+2. Starting this node sets stream handler on wakunode.Host for WakuFilterProtocolId.
+3. Light node submits a FilterSubscribeRequest through WakuFilterLightNode.Subscribe. This request is submitted to a particular peer.
+Filter is stored in WakuFilterLightNode.subscriptions map. That's it.
+4. Now we wait on WakuFilterLightNode.onRequest to process any further messages.
+5. On receiving a message check and notify all subscribers on relevant channel (which is part of subscription obbject).
+6. If a broadcaster is specified, 
+  WakuNode.Subscribe has a message loop extracting WakuMessages from a wakurelay.Subscription object.It denotes a pubsub topic subscription.All envelopes are then submitted to node.broadcaster.
+### Full Node
+1. Full node is created with option WithWakuFilterFullNode.
+2. We read incoming messages in WithWakuFilterFullNode.onRequest(). It is set as a stream handler on wakunode.Host for WakuFilterProtocolId.
+3. In WakuFilter.onRequest
+  * We check the type of FilterRequest and handle accordingly.
+  * If it's a FilterRequest for subscribe, add a subscriber.
+  * If it is a SubscriberPing request, check if subscriptions exists or not and respond accordingly.
+  *  If it is an unsubscribe/unsubscribeAll request, check and remove relevant subscriptions.

examples/filter2/main.go

@@ -26,7 +26,7 @@ var log = logging.Logger("filter2")
 
 var pubSubTopic = protocol.DefaultPubsubTopic()
 
-const contentTopic = "test"
+const contentTopic = "/filter2test/1/testTopic/proto"
 
 func main() {
 	hostAddr1, _ := net.ResolveTCPAddr("tcp", "0.0.0.0:60000")
@@ -98,7 +98,6 @@ func main() {
 
 	// Send FilterRequest from light node to full node
 	cf := filter.ContentFilter{
-		Topic:         pubSubTopic.String(),
 		ContentTopics: []string{contentTopic},
 	}
 
@@ -108,7 +107,7 @@ func main() {
 	}
 
 	go func() {
-		for env := range theFilter.C {
+		for env := range theFilter[0].C { //Safely picking first subscriptions since only 1 contentTopic is subscribed
 			log.Info("Light node received msg, ", string(env.Message().Payload))
 		}
 		log.Info("Message channel closed!")

library/c/README.md

@@ -88,7 +88,7 @@ The criteria to create subscription to a filter full node in JSON Format:
 Fields:
 
 - `contentTopics`: Array of content topics.
-- `topic`: pubsub topic.
+- `topic`: Optional pubsub topic.
 
 
 ### `LegacyFilterSubscription` type

library/c/api_filter.go

@@ -10,7 +10,7 @@ import "github.com/waku-org/go-waku/library"
 // filterJSON must contain a JSON with this format:
 //
 //		{
-//		  "pubsubTopic": "the pubsub topic" // mandatory
+//		  "pubsubTopic": "the pubsub topic" // optional
 //	      "contentTopics": ["the content topic"] // mandatory, at least one required, with a max of 10
 //		}
 //
@@ -42,7 +42,7 @@ func waku_filter_ping(peerID *C.char, ms C.int, onErrCb C.WakuCallBack) C.int {
 // criteria
 //
 //		{
-//		  "pubsubTopic": "the pubsub topic" // mandatory
+//		  "pubsubTopic": "the pubsub topic" // optional
 //	      "contentTopics": ["the content topic"] // mandatory, at least one required, with a max of 10
 //		}
 //

library/filter.go

@@ -60,18 +60,20 @@ func FilterSubscribe(filterJSON string, peerID string, ms int) (string, error) {
 		fOptions = append(fOptions, filter.WithAutomaticPeerSelection())
 	}
 
-	subscriptionDetails, err := wakuState.node.FilterLightnode().Subscribe(ctx, cf, fOptions...)
+	subscriptions, err := wakuState.node.FilterLightnode().Subscribe(ctx, cf, fOptions...)
 	if err != nil {
 		return "", err
 	}
 
-	go func(subscriptionDetails *filter.SubscriptionDetails) {
-		for envelope := range subscriptionDetails.C {
-			send("message", toSubscriptionMessage(envelope))
-		}
-	}(subscriptionDetails)
+	for _, subscriptionDetails := range subscriptions {
+		go func(subscriptionDetails *filter.SubscriptionDetails) {
+			for envelope := range subscriptionDetails.C {
+				send("message", toSubscriptionMessage(envelope))
+			}
+		}(subscriptionDetails)
+	}
 
-	return marshalJSON(subscriptionDetails)
+	return marshalJSON(subscriptions)
 }
 
 // FilterPing is used to determine if a peer has an active subscription