Subscribe RPC and quicksim WIP; hacky in-line handler

waku/node/v2/quicksim.nim

@@ -34,7 +34,11 @@ waitFor node2.connect("localhost", Port(8548))
 
 let version = waitFor node1.wakuVersion()
 
-let res = waitFor node1.wakuPost("hello world")
+proc handler(topic: string, data: seq[byte]) {.async, gcsafe.} =
+  debug "Hit handler", topic=topic, data=data
+
+# TODO: Implement handler logic
+let res1 = waitFor node2.wakuSubscribe("foobar")
 let res2 = waitFor node1.wakuPublish("foobar", "hello world")
 
 info "Version is", version

waku/node/v2/rpc/wakucallsigs.nim

@@ -1,6 +1,8 @@
 # Alpha - Currently implemented in v2
 proc waku_version(): string
 proc waku_publish(topic: string, message: string): bool
+proc waku_subscribe(topic: string): bool
+#proc waku_subscribe(topic: string, handler: Topichandler): bool
 
 # NYI
 proc waku_info(): WakuInfo

waku/node/v2/rpc/wakurpc.nim

@@ -4,7 +4,8 @@ import
   ../../../protocol/v2/waku_protocol,
   nimcrypto/[sysrand, hmac, sha2, pbkdf2],
   ../../v1/rpc/[rpc_types, hexstrings, key_storage],
-  ../waku_types
+  ../waku_types,
+  libp2p/protocols/pubsub/pubsub
 
 from stew/byteutils import hexToSeqByte, hexToByteArray
 
@@ -34,6 +35,16 @@ proc setupWakuRPC*(wakuProto: WakuProto, rpcsrv: RpcServer) =
     return true
     #if not result:
     #  raise newException(ValueError, "Message could not be posted")
+
+  # TODO: Handler / Identifier logic
+  rpcsrv.rpc("waku_subscribe") do(topic: string) -> bool:
+    let wakuSub = cast[WakuSub](wakuProto.switch.pubSub.get())
+
+    # XXX: Hacky in-line handler
+    proc handler(topic: string, data: seq[byte]) {.async, gcsafe.} =
+      info "Hit subscribe handler", topic=topic, data=data
+
+    discard wakuSub.subscribe(topic, handler)
     return true
     #if not result:
     #  raise newException(ValueError, "Message could not be posted")

waku/protocol/v2/waku_protocol.nim

@@ -47,6 +47,14 @@ method initPubSub*(w: WakuSub) =
   procCall FloodSub(w).initPubSub()
   w.init()
 
+method subscribe*(w: WakuSub,
+                  topic: string,
+                  handler: TopicHandler) {.async.} =
+  debug "subscribe", topic=topic
+  # XXX: Pubsub really
+  await procCall FloodSub(w).subscribe(topic, handler)
+
+# Subscribing a peer to a specified topic
 method subscribeTopic*(w: WakuSub,
                        topic: string,
                        subscribe: bool,