Make bridge topics configurable (#629)

* Make bridge topics configurable

* Less ambiguous config naming

tests/v2/test_waku_bridge.nim

@@ -26,6 +26,8 @@ procSuite "WakuBridge":
   # Suite setup #
   ###############
 
+  const DefaultBridgeTopic = "/waku/2/default-bridge/proto"
+
   let
     rng = keys.newRng()
 
@@ -38,7 +40,8 @@ procSuite "WakuBridge":
         powRequirement = 0.002,
         rng = rng,
         nodev2Key = nodev2Key,
-        nodev2BindIp = ValidIpAddress.init("0.0.0.0"), nodev2BindPort= Port(60000))
+        nodev2BindIp = ValidIpAddress.init("0.0.0.0"), nodev2BindPort= Port(60000),
+        nodev2PubsubTopic = DefaultBridgeTopic)
     
     # Waku v1 node
     v1Node = setupTestNode(rng, Waku)

waku/common/config_bridge.nim

@@ -93,10 +93,15 @@ type
       defaultValue: keys.KeyPair.random(keys.newRng()[])
       name: "nodekey-v1" .}: keys.KeyPair
 
-    wakuPow* {.
-      desc: "PoW requirement of Waku node.",
+    wakuV1Pow* {.
+      desc: "PoW requirement of Waku v1 node.",
       defaultValue: 0.002
-      name: "waku-pow" .}: float64
+      name: "waku-v1-pow" .}: float64
+    
+    wakuV1TopicInterest* {.
+      desc: "Run as Waku v1 node with a topic-interest",
+      defaultValue: false
+      name: "waku-v1-topic-interest" .}: bool
 
     ### Waku v2 options
 
@@ -109,11 +114,6 @@ type
       defaultValue: crypto.PrivateKey.random(Secp256k1, keys.newRng()[]).tryGet()
       name: "nodekey-v2" }: crypto.PrivateKey
 
-    topics* {.
-      desc: "Default topics to subscribe to (space separated list)"
-      defaultValue: "/waku/2/default-waku/proto"
-      name: "topics" .}: string
-
     store* {.
       desc: "Flag whether to start store protocol",
       defaultValue: true
@@ -138,6 +138,13 @@ type
       desc: "Multiaddr of peer to connect with for waku filter protocol"
       defaultValue: ""
       name: "filternode" }: string
+    
+    ### Bridge options
+
+    bridgePubsubTopic* {.
+      desc: "Waku v2 Pubsub topic to bridge to/from"
+      defaultValue: "/waku/2/default-waku/proto"
+      name: "bridge-pubsub-topic" }: string
 
 proc parseCmdArg*(T: type keys.KeyPair, p: TaintedString): T =
   try:

waku/common/wakubridge.nim

@@ -26,7 +26,6 @@ logScope:
 ##################
 
 const
-  DefaultBridgeTopic* = "/waku/2/default-bridge/proto"
   ClientIdV1 = "nim-waku v1 node"
   DefaultTTL = 5'u32
   DeduplQSize = 20  # Maximum number of seen messages to keep in deduplication queue
@@ -39,6 +38,7 @@ type
   WakuBridge* = ref object of RootObj
     nodev1*: EthereumNode
     nodev2*: WakuNode
+    nodev2PubsubTopic: wakunode2.Topic # Pubsub topic to bridge to/from
     seen: seq[hashes.Hash] # FIFO queue of seen WakuMessages. Used for deduplication.
 
 ###################
@@ -76,7 +76,7 @@ proc toWakuV2(bridge: WakuBridge, env: Envelope) {.async.} =
 
   waku_bridge_transfers.inc(labelValues = ["v1_to_v2"])
   
-  await bridge.nodev2.publish(DefaultBridgeTopic, msg)
+  await bridge.nodev2.publish(bridge.nodev2PubsubTopic, msg)
 
 proc toWakuV1(bridge: WakuBridge, msg: WakuMessage) {.gcsafe.} =
   if bridge.seen.containsOrAdd(msg.encode().buffer.hash()):
@@ -105,10 +105,14 @@ proc new*(T: type WakuBridge,
           nodev1Address: Address,
           powRequirement = 0.002,
           rng: ref BrHmacDrbgContext,
+          topicInterest = none(seq[waku_protocol.Topic]),
+          bloom = some(fullBloom()),
           # NodeV2 initialisation
           nodev2Key: crypto.PrivateKey,
           nodev2BindIp: ValidIpAddress, nodev2BindPort: Port,
-          nodev2ExtIp = none[ValidIpAddress](), nodev2ExtPort = none[Port]()): T =
+          nodev2ExtIp = none[ValidIpAddress](), nodev2ExtPort = none[Port](),
+          # Bridge configuration
+          nodev2PubsubTopic: wakunode2.Topic): T =
 
   # Setup Waku v1 node
   var
@@ -119,13 +123,15 @@ proc new*(T: type WakuBridge,
   nodev1.addCapability Waku # Always enable Waku protocol
 
   # Setup the Waku configuration.
-  # This node is being set up as a bridge so it gets configured as a node with
+  # This node is being set up as a bridge. By default it gets configured as a node with
   # a full bloom filter so that it will receive and forward all messages.
+  # It is, however, possible to configure a topic interest to bridge only
+  # selected messages.
   # TODO: What is the PoW setting now?
   let wakuConfig = WakuConfig(powRequirement: powRequirement,
-                              bloom: some(fullBloom()), isLightNode: false,
+                              bloom: bloom, isLightNode: false,
                               maxMsgSize: waku_protocol.defaultMaxMsgSize,
-                              topics: none(seq[waku_protocol.Topic]))
+                              topics: topicInterest)
   nodev1.configureWaku(wakuConfig)
 
   # Setup Waku v2 node
@@ -134,7 +140,7 @@ proc new*(T: type WakuBridge,
                            nodev2BindIp, nodev2BindPort,
                            nodev2ExtIp, nodev2ExtPort)
   
-  return WakuBridge(nodev1: nodev1, nodev2: nodev2)
+  return WakuBridge(nodev1: nodev1, nodev2: nodev2, nodev2PubsubTopic: nodev2PubsubTopic)
 
 proc start*(bridge: WakuBridge) {.async.} =
   info "Starting WakuBridge"
@@ -174,7 +180,7 @@ proc start*(bridge: WakuBridge) {.async.} =
       trace "Bridging message from V2 to V1", msg=msg[]
       bridge.toWakuV1(msg[])
   
-  bridge.nodev2.subscribe(DefaultBridgeTopic, relayHandler)
+  bridge.nodev2.subscribe(bridge.nodev2PubsubTopic, relayHandler)
 
 proc stop*(bridge: WakuBridge) {.async.} =
   await bridge.nodev2.stop()
@@ -234,15 +240,28 @@ when isMainModule:
     (nodev2ExtIp, nodev2ExtPort, _) = setupNat(conf.nat, clientId,
                                                Port(uint16(conf.libp2pTcpPort) + conf.portsShift),
                                                Port(uint16(conf.udpPort) + conf.portsShift))
+  
+  # Topic interest and bloom
+  var topicInterest: Option[seq[waku_protocol.Topic]]
+  var bloom: Option[Bloom]
+  
+  if conf.wakuV1TopicInterest:
+    var topics: seq[waku_protocol.Topic]
+    topicInterest = some(topics)
+  else:
+    bloom = some(fullBloom())
 
   let
     bridge = WakuBridge.new(nodev1Key = conf.nodekeyV1,
                             nodev1Address = nodev1Address,
-                            powRequirement = conf.wakuPow,
+                            powRequirement = conf.wakuV1Pow,
                             rng = rng,
+                            topicInterest = topicInterest,
+                            bloom = bloom,
                             nodev2Key = conf.nodekeyV2,
                             nodev2BindIp = conf.listenAddress, nodev2BindPort = Port(uint16(conf.libp2pTcpPort) + conf.portsShift),
-                            nodev2ExtIp = nodev2ExtIp, nodev2ExtPort = nodev2ExtPort)
+                            nodev2ExtIp = nodev2ExtIp, nodev2ExtPort = nodev2ExtPort,
+                            nodev2PubsubTopic = conf.bridgePubsubTopic)
   
   waitFor bridge.start()