Add functionality to unsubscribe from pubsub topic (#241)

docs/api/v2/node.md

@@ -40,11 +40,15 @@ proc subscribe*(node: WakuNode, request: FilterRequest, handler: ContentFilterHa
   ##
   ## Status: Implemented.
 
-proc unsubscribe*(w: WakuNode, topic: Topic) =
-  ## Unsubscribe from a topic.
+proc unsubscribe*(node: WakuNode, topic: Topic, handler: TopicHandler) {.async.} =
+  ## Unsubscribes a handler from a PubSub topic.
   ##
-  ## Status: Not yet implemented.
-  ## TODO Implement.
+  ## Status: Implemented.
+
+proc unsubscribeAll*(node: WakuNode, topic: Topic) {.async.} =
+  ## Unsubscribes all handlers registered on a specific PubSub topic.
+  ##
+  ## Status: Implemented.
 
 proc unsubscribe*(w: WakuNode, contentFilter: ContentFilter) =
   ## Unsubscribe from a content filter.

waku/node/v2/wakunode2.nim

@@ -132,12 +132,24 @@ proc subscribe*(node: WakuNode, request: FilterRequest, handler: ContentFilterHa
     id = await node.wakuFilter.subscribe(request)
   node.filters[id] = Filter(contentFilters: request.contentFilters, handler: handler)
 
-proc unsubscribe*(w: WakuNode, topic: Topic) =
-  echo "NYI"
-  ## Unsubscribe from a topic.
+proc unsubscribe*(node: WakuNode, topic: Topic, handler: TopicHandler) {.async.} =
+  ## Unsubscribes a handler from a PubSub topic.
   ##
-  ## Status: Not yet implemented.
-  ## TODO Implement.
+  ## Status: Implemented.
+  info "unsubscribe", topic=topic
+
+  let wakuRelay = node.wakuRelay
+  await wakuRelay.unsubscribe(@[(topic, handler)])
+
+proc unsubscribeAll*(node: WakuNode, topic: Topic) {.async.} =
+  ## Unsubscribes all handlers registered on a specific PubSub topic.
+  ##
+  ## Status: Implemented.
+  info "unsubscribeAll", topic=topic
+
+  let wakuRelay = node.wakuRelay
+  await wakuRelay.unsubscribeAll(topic)
+  
 
 proc unsubscribe*(w: WakuNode, contentFilter: waku_types.ContentFilter) =
   echo "NYI"

waku/protocol/v2/waku_relay.nim

@@ -74,6 +74,14 @@ method unsubscribe*(w: WakuRelay,
   else:
     await procCall FloodSub(w).unsubscribe(topics)
 
+method unsubscribeAll*(w: WakuRelay,
+                       pubSubTopic: string) {.async.} =
+  debug "unsubscribeAll"
+  if w.gossipEnabled:
+    await procCall GossipSub(w).unsubscribeAll(pubSubTopic)
+  else:
+    await procCall FloodSub(w).unsubscribeAll(pubSubTopic)
+
 # GossipSub specific methods --------------------------------------------------
 method start*(w: WakuRelay) {.async.} =
   debug "start"