diff --git a/apps/waku_publisher.nim b/apps/waku_publisher.nim new file mode 100644 index 0000000..8c2d036 --- /dev/null +++ b/apps/waku_publisher.nim @@ -0,0 +1,142 @@ +import + std/[tables, times, sequtils], + stew/byteutils, + chronicles, + chronos, + confutils, + libp2p/crypto/crypto, + eth/keys, + eth/p2p/discoveryv5/enr + +import + waku/[ + common/logging, + node/peer_manager, + waku_core, + waku_node, + waku_enr, + discovery/waku_discv5, + factory/builder, + ] + +proc now*(): Timestamp = + getNanosecondTime(getTime().toUnixFloat()) + +# An accesible bootstrap node. See waku.sandbox fleets.status.im +const bootstrapNode = + "enr:-QEkuEB3WHNS-xA3RDpfu9A2Qycr3bN3u7VoArMEiDIFZJ6" & + "6F1EB3d4wxZN1hcdcOX-RfuXB-MQauhJGQbpz3qUofOtLAYJpZI" & + "J2NIJpcIQI2SVcim11bHRpYWRkcnO4bgA0Ni9ub2RlLTAxLmFjL" & + "WNuLWhvbmdrb25nLWMud2FrdS5zYW5kYm94LnN0YXR1cy5pbQZ2" & + "XwA2Ni9ub2RlLTAxLmFjLWNuLWhvbmdrb25nLWMud2FrdS5zYW5" & + "kYm94LnN0YXR1cy5pbQYfQN4DgnJzkwABCAAAAAEAAgADAAQABQ" & + "AGAAeJc2VjcDI1NmsxoQPK35Nnz0cWUtSAhBp7zvHEhyU_AqeQU" & + "lqzLiLxfP2L4oN0Y3CCdl-DdWRwgiMohXdha3UyDw" + +# careful if running pub and sub in the same machine +const wakuPort = 60000 +const discv5Port = 9000 + +proc setupAndPublish(rng: ref HmacDrbgContext) {.async.} = + # use notice to filter all waku messaging + setupLog(logging.LogLevel.NOTICE, logging.LogFormat.TEXT) + + notice "starting publisher", wakuPort = wakuPort, discv5Port = discv5Port + let + nodeKey = crypto.PrivateKey.random(Secp256k1, rng[]).get() + ip = parseIpAddress("0.0.0.0") + flags = CapabilitiesBitfield.init(relay = true) + + var enrBuilder = EnrBuilder.init(nodeKey) + + let recordRes = enrBuilder.build() + let record = + if recordRes.isErr(): + error "failed to create enr record", error = recordRes.error + quit(QuitFailure) + else: + recordRes.get() + + var builder = WakuNodeBuilder.init() + builder.withNodeKey(nodeKey) + builder.withRecord(record) + builder.withNetworkConfigurationDetails(ip, Port(wakuPort)).tryGet() + let node = builder.build().tryGet() + + var bootstrapNodeEnr: enr.Record + discard bootstrapNodeEnr.fromURI(bootstrapNode) + + let discv5Conf = WakuDiscoveryV5Config( + discv5Config: none(DiscoveryConfig), + address: ip, + port: Port(discv5Port), + privateKey: keys.PrivateKey(nodeKey.skkey), + bootstrapRecords: @[bootstrapNodeEnr], + autoupdateRecord: true, + ) + + # assumes behind a firewall, so not care about being discoverable + let wakuDiscv5 = WakuDiscoveryV5.new( + node.rng, + discv5Conf, + some(node.enr), + some(node.peerManager), + node.topicSubscriptionQueue, + ) + + await node.start() + (await node.mountRelay()).isOkOr: + error "failed to mount relay", error = error + quit(1) + + node.peerManager.start() + + (await wakuDiscv5.start()).isOkOr: + error "failed to start discv5", error = error + quit(1) + + # wait for a minimum of peers to be connected, otherwise messages wont be gossiped + while true: + let numConnectedPeers = node.peerManager.switch.peerStore[ConnectionBook].book + .values() + .countIt(it == Connected) + if numConnectedPeers >= 6: + notice "publisher is ready", connectedPeers = numConnectedPeers, required = 6 + break + notice "waiting to be ready", connectedPeers = numConnectedPeers, required = 6 + await sleepAsync(5000) + + # Make sure it matches the publisher. Use default value + # see spec: https://rfc.vac.dev/spec/23/ + let pubSubTopic = PubsubTopic("/waku/2/rs/0/0") + + # any content topic can be chosen + let contentTopic = ContentTopic("/examples/1/pubsub-example/proto") + + notice "publisher service started" + while true: + let text = "hi there i'm a publisher" + let message = WakuMessage( + payload: toBytes(text), # content of the message + contentTopic: contentTopic, # content topic to publish to + ephemeral: true, # tell store nodes to not store it + timestamp: now(), + ) # current timestamp + + let res = await node.publish(some(pubSubTopic), message) + + if res.isOk: + notice "published message", + text = text, + timestamp = message.timestamp, + psTopic = pubSubTopic, + contentTopic = contentTopic + else: + error "failed to publish message", error = res.error + + await sleepAsync(5000) + +when isMainModule: + let rng = crypto.newRng() + asyncSpawn setupAndPublish(rng) + runForever() diff --git a/apps/waku_subscriber.nim b/apps/waku_subscriber.nim new file mode 100644 index 0000000..fb040b0 --- /dev/null +++ b/apps/waku_subscriber.nim @@ -0,0 +1,129 @@ +import + std/[tables, sequtils], + stew/byteutils, + chronicles, + chronos, + confutils, + libp2p/crypto/crypto, + eth/keys, + eth/p2p/discoveryv5/enr + +import + waku/[ + common/logging, + node/peer_manager, + waku_core, + waku_node, + waku_enr, + discovery/waku_discv5, + factory/builder, + waku_relay, + ] + +# An accesible bootstrap node. See waku.sandbox fleets.status.im +const bootstrapNode = + "enr:-QEkuEB3WHNS-xA3RDpfu9A2Qycr3bN3u7VoArMEiDIFZJ6" & + "6F1EB3d4wxZN1hcdcOX-RfuXB-MQauhJGQbpz3qUofOtLAYJpZI" & + "J2NIJpcIQI2SVcim11bHRpYWRkcnO4bgA0Ni9ub2RlLTAxLmFjL" & + "WNuLWhvbmdrb25nLWMud2FrdS5zYW5kYm94LnN0YXR1cy5pbQZ2" & + "XwA2Ni9ub2RlLTAxLmFjLWNuLWhvbmdrb25nLWMud2FrdS5zYW5" & + "kYm94LnN0YXR1cy5pbQYfQN4DgnJzkwABCAAAAAEAAgADAAQABQ" & + "AGAAeJc2VjcDI1NmsxoQPK35Nnz0cWUtSAhBp7zvHEhyU_AqeQU" & + "lqzLiLxfP2L4oN0Y3CCdl-DdWRwgiMohXdha3UyDw" + +# careful if running pub and sub in the same machine +const wakuPort = 50000 +const discv5Port = 8000 + +proc setupAndSubscribe(rng: ref HmacDrbgContext) {.async.} = + # use notice to filter all waku messaging + setupLog(logging.LogLevel.NOTICE, logging.LogFormat.TEXT) + + notice "starting subscriber", wakuPort = wakuPort, discv5Port = discv5Port + let + nodeKey = crypto.PrivateKey.random(Secp256k1, rng[])[] + ip = parseIpAddress("0.0.0.0") + flags = CapabilitiesBitfield.init(relay = true) + + var enrBuilder = EnrBuilder.init(nodeKey) + + let recordRes = enrBuilder.build() + let record = + if recordRes.isErr(): + error "failed to create enr record", error = recordRes.error + quit(QuitFailure) + else: + recordRes.get() + + var builder = WakuNodeBuilder.init() + builder.withNodeKey(nodeKey) + builder.withRecord(record) + builder.withNetworkConfigurationDetails(ip, Port(wakuPort)).tryGet() + let node = builder.build().tryGet() + + var bootstrapNodeEnr: enr.Record + discard bootstrapNodeEnr.fromURI(bootstrapNode) + + let discv5Conf = WakuDiscoveryV5Config( + discv5Config: none(DiscoveryConfig), + address: ip, + port: Port(discv5Port), + privateKey: keys.PrivateKey(nodeKey.skkey), + bootstrapRecords: @[bootstrapNodeEnr], + autoupdateRecord: true, + ) + + # assumes behind a firewall, so not care about being discoverable + let wakuDiscv5 = WakuDiscoveryV5.new( + node.rng, + discv5Conf, + some(node.enr), + some(node.peerManager), + node.topicSubscriptionQueue, + ) + + await node.start() + (await node.mountRelay()).isOkOr: + error "failed to mount relay", error = error + quit(1) + node.peerManager.start() + + (await wakuDiscv5.start()).isOkOr: + error "failed to start discv5", error = error + quit(1) + + # wait for a minimum of peers to be connected, otherwise messages wont be gossiped + while true: + let numConnectedPeers = node.peerManager.switch.peerStore[ConnectionBook].book + .values() + .countIt(it == Connected) + if numConnectedPeers >= 6: + notice "subscriber is ready", connectedPeers = numConnectedPeers, required = 6 + break + notice "waiting to be ready", connectedPeers = numConnectedPeers, required = 6 + await sleepAsync(5000) + + # Make sure it matches the publisher. Use default value + # see spec: https://rfc.vac.dev/spec/23/ + let pubSubTopic = PubsubTopic("/waku/2/rs/0/0") + + # any content topic can be chosen. make sure it matches the publisher + let contentTopic = ContentTopic("/examples/1/pubsub-example/proto") + + proc handler(topic: PubsubTopic, msg: WakuMessage): Future[void] {.async, gcsafe.} = + let payloadStr = string.fromBytes(msg.payload) + if msg.contentTopic == contentTopic: + notice "message received", + payload = payloadStr, + pubsubTopic = pubsubTopic, + contentTopic = msg.contentTopic, + timestamp = msg.timestamp + + node.subscribe((kind: PubsubSub, topic: pubsubTopic), WakuRelayHandler(handler)).isOkOr: + error "failed to subscribe to pubsub topic", pubsubTopic, error + quit(1) + +when isMainModule: + let rng = crypto.newRng() + asyncSpawn setupAndSubscribe(rng) + runForever() diff --git a/chat_sdk.nimble b/chat_sdk.nimble index b1165f3..67c1204 100644 --- a/chat_sdk.nimble +++ b/chat_sdk.nimble @@ -7,7 +7,7 @@ license = "MIT" srcDir = "src" ### Dependencies -requires "nim >= 2.2.4", "chronicles", "chronos", "db_connector", "flatty" +requires "nim >= 2.2.4", "chronicles", "chronos", "db_connector", "flatty", "waku#nimble-fix-next" task buildSharedLib, "Build shared library for C bindings": exec "nim c --mm:refc --app:lib --out:../library/c-bindings/libchatsdk.so chat_sdk/chat_sdk.nim" @@ -20,3 +20,9 @@ task migrate, "Run database migrations": task segment, "Run segmentation": exec "nim c -r chat_sdk/segmentation.nim" + +task subscribe, "Run waku subscription service": + exec "nim c -r apps/waku_subscriber.nim" + +task publish, "Run waku publishing service": + exec "nim c -r apps/waku_publisher.nim"