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{
-		PubsubTopic:   pubSubTopic.String(),
 		ContentTopics: filter.NewContentTopicSet(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 when using contentTopics as per Autosharding. In case of named or static-sharding, pubSub topic is mandatory.
 
 
 ### `LegacyFilterSubscription` type
@@ -884,15 +884,21 @@ Creates a subscription to a filter full node matching a content filter..
 
 A status code. Refer to the [`Status codes`](#status-codes) section for possible values.
 
-If the function is executed succesfully, `onOkCb` will receive the subscription details.
+If the function is executed succesfully, `onOkCb` will receive the following subscription details along with any partial errors.
 
 For example:
 
 ```json
 {
-  "peerID": "....",
-  "pubsubTopic": "...",
-  "contentTopics": [...]
+  "subscriptions" : [
+    {
+      "ID": "<subscriptionID>",
+      "peerID": "....",
+      "pubsubTopic": "...",
+      "contentTopics": [...]
+    }
+  ],
+  "error" : "subscriptions failed for contentTopics:<topicA>,.." // Empty if all subscriptions are succesful
 }
 ```
 

library/c/api_filter.go

@@ -10,14 +10,14 @@ 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 if using autosharding, mandatory if using static or named sharding.
 //	      "contentTopics": ["the content topic"] // mandatory, at least one required, with a max of 10
 //		}
 //
 // peerID should contain the ID of a peer supporting the filter protocol. Use NULL to automatically select a node
 // If ms is greater than 0, the subscription must happen before the timeout
 // (in milliseconds) is reached, or an error will be returned
-// It returns a json object containing the peerID to which we are subscribed to and the details of the subscription
+// It returns a json object containing the details of the subscriptions along with any errors in case of partial failures
 //
 //export waku_filter_subscribe
 func waku_filter_subscribe(filterJSON *C.char, peerID *C.char, ms C.int, onOkCb C.WakuCallBack, onErrCb C.WakuCallBack) C.int {
@@ -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 if using autosharding, mandatory if using static or named sharding.
 //	      "contentTopics": ["the content topic"] // mandatory, at least one required, with a max of 10
 //		}
 //

library/filter.go

@@ -28,6 +28,11 @@ func toContentFilter(filterJSON string) (filter.ContentFilter, error) {
 	}, nil
 }
 
+type subscribeResult struct {
+	Subscriptions []*filter.SubscriptionDetails `json:"subscriptions"`
+	Error         string                        `json:"error,omitempty"`
+}
+
 // FilterSubscribe is used to create a subscription to a filter node to receive messages
 func FilterSubscribe(filterJSON string, peerID string, ms int) (string, error) {
 	cf, err := toContentFilter(filterJSON)
@@ -60,18 +65,22 @@ func FilterSubscribe(filterJSON string, peerID string, ms int) (string, error) {
 		fOptions = append(fOptions, filter.WithAutomaticPeerSelection())
 	}
 
-	subscriptionDetails, err := wakuState.node.FilterLightnode().Subscribe(ctx, cf, fOptions...)
-	if err != nil {
+	subscriptions, err := wakuState.node.FilterLightnode().Subscribe(ctx, cf, fOptions...)
+	if err != nil && subscriptions == nil {
 		return "", err
 	}
 
-	go func(subscriptionDetails *filter.SubscriptionDetails) {
-		for envelope := range subscriptionDetails.C {
-			send("message", toSubscriptionMessage(envelope))
-		}
-	}(subscriptionDetails)
-
-	return marshalJSON(subscriptionDetails)
+	for _, subscriptionDetails := range subscriptions {
+		go func(subscriptionDetails *filter.SubscriptionDetails) {
+			for envelope := range subscriptionDetails.C {
+				send("message", toSubscriptionMessage(envelope))
+			}
+		}(subscriptionDetails)
+	}
+	var subResult subscribeResult
+	subResult.Subscriptions = subscriptions
+	subResult.Error = err.Error()
+	return marshalJSON(subResult)
 }
 
 // FilterPing is used to determine if a peer has an active subscription