From 4460adf7bf7fc9d3bf537e1a3fe3d3168fb0eb86 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Oskar=20Thor=C3=A9n?= <ot@oskarthoren.com>
Date: Tue, 28 Jul 2020 16:17:50 +0800
Subject: [PATCH] Node API: Implement basic subscribe topic (#89)

* Fix init signature according to Node API

See https://github.com/status-im/nim-waku/blob/master/docs/api/v2/node.md

* Update docs and example

* Node API: Basic subscribe

* Modify example with subscribe handler
---
 examples/v2/basic2.nim     | 17 ++++++++++-------
 waku/node/v2/wakunode2.nim | 22 ++++++++++++++++------
 2 files changed, 26 insertions(+), 13 deletions(-)

diff --git a/examples/v2/basic2.nim b/examples/v2/basic2.nim
index d007e01c1..0ffe527cb 100644
--- a/examples/v2/basic2.nim
+++ b/examples/v2/basic2.nim
@@ -15,15 +15,18 @@ import ../../waku/node/v2/wakunode2
 # Loads the config in `waku/node/v2/config.nim`
 let conf = WakuNodeConf.load()
 
-# Create and start the node
-#proc runBackground(conf: WakuNodeConf) {.async.} =
-#  init(conf)
-#  runForever()
+# Node operations happens asynchronously
+proc runBackground(conf: WakuNodeConf) {.async.} =
+  # Create and start the node
+  let node = await init(conf)
 
-discard init(conf)
-
-echo("Do stuff after with node")
+  # Subscribe to a topic
+  let topic = "foobar"
+  proc handler(topic: string, data: seq[byte]) {.async, gcsafe.} =
+    info "Hit subscribe handler", topic=topic, data=data
+  node.subscribe(topic, handler)
 
+discard runBackground(conf)
 runForever()
 
 # TODO Lets start with Nim Node API first
diff --git a/waku/node/v2/wakunode2.nim b/waku/node/v2/wakunode2.nim
index b5dfb37e8..14159ae0d 100644
--- a/waku/node/v2/wakunode2.nim
+++ b/waku/node/v2/wakunode2.nim
@@ -6,6 +6,8 @@ import
   libp2p/multiaddress,
   libp2p/crypto/crypto,
   libp2p/protocols/protocol,
+  # NOTE For TopicHandler, solve with exports?
+  libp2p/protocols/pubsub/pubsub,
   libp2p/peerinfo,
   stew/shims/net as stewNet,
   rpc/wakurpc,
@@ -231,8 +233,13 @@ method init*(conf: WakuNodeConf): Future[WakuNode] {.async.} =
 type Topic* = string
 type Message* = seq[byte]
 type ContentFilter* = object
-    contentTopic*: string
-type TopicHandler* = proc(topic: Topic, message: Message)
+  contentTopic*: string
+
+# TODO Update TopicHandler to take Message, not seq[byte] data
+#type TopicHandler* = proc(topic: Topic, message: Message)
+# Currently this is using the one in pubsub.nim, roughly:
+#type TopicHandler* = proc(topic: string, data: seq[byte])
+
 type ContentFilterHandler* = proc(contentFilter: ContentFilter, message: Message)
 
 type HistoryQuery = object
@@ -242,13 +249,16 @@ type HistoryResponse = object
     xxx*: seq[byte]
 
 method subscribe*(w: WakuNode, topic: Topic, handler: TopicHandler) =
-  echo "NYI"
   ## Subscribes to a PubSub topic. Triggers handler when receiving messages on
   ## this topic. TopicHandler is a method that takes a topic and a `Message`.
   ##
-  ## Status: Not yet implemented.
-  ## TODO Implement as wrapper around `waku_protocol`, and ensure Message is
-  ## passed, not `data` field.
+  ## Status: Partially implemented.
+  ## TODO Ensure Message is passed, not `data` field. This means modifying
+  ## TopicHandler.
+
+  let wakuSub = w.switch.pubSub.get()
+  # XXX Consider awaiting here
+  discard wakuSub.subscribe(topic, handler)
 
 method subscribe*(w: WakuNode, contentFilter: ContentFilter, handler: ContentFilterHandler) =
   echo "NYI"