feat(waku utils): Add function that init and start `waku` instance
This commit is contained in:
parent
e9c728c64d
commit
7714f6d492
|
@ -1,6 +1,4 @@
|
||||||
import
|
import
|
||||||
std/[tables, times, sequtils],
|
|
||||||
stew/shims/net,
|
|
||||||
chronicles,
|
chronicles,
|
||||||
chronos,
|
chronos,
|
||||||
confutils,
|
confutils,
|
||||||
|
@ -16,14 +14,7 @@ import
|
||||||
waku/common/[logging, protobuf]
|
waku/common/[logging, protobuf]
|
||||||
|
|
||||||
import ../../waku_handshake_utils
|
import ../../waku_handshake_utils
|
||||||
|
import ../../waku_node
|
||||||
const bootstrapNode = "enr:-P-4QGVNANzbhCI49du6Moyw98AjuMhKoOpE_Jges9JlCq-I" &
|
|
||||||
"CAVadktjfcNpuhQgT0g1cu86_S3nbM7eYkCsqDAQG7UBgmlkgnY0" &
|
|
||||||
"gmlwhI_G-a6KbXVsdGlhZGRyc7hgAC02KG5vZGUtMDEuZG8tYW1z" &
|
|
||||||
"My5zdGF0dXMucHJvZC5zdGF0dXNpbS5uZXQGdl8ALzYobm9kZS0w" &
|
|
||||||
"MS5kby1hbXMzLnN0YXR1cy5wcm9kLnN0YXR1c2ltLm5ldAYBu94D" &
|
|
||||||
"iXNlY3AyNTZrMaECoVyonsTGEQvVioM562Q1fjzTb_vKD152PPId" &
|
|
||||||
"sV7sM6SDdGNwgnZfg3VkcIIjKIV3YWt1Mg8"
|
|
||||||
|
|
||||||
const
|
const
|
||||||
wakuPort = 60000
|
wakuPort = 60000
|
||||||
|
@ -60,61 +51,9 @@ proc exampleJSWaku(rng: ref HmacDrbgContext) {.async.} =
|
||||||
initiatorHS = initHS(initiatorInfo, qr, true)
|
initiatorHS = initHS(initiatorInfo, qr, true)
|
||||||
initiatorHSResult: HandshakeResult
|
initiatorHSResult: HandshakeResult
|
||||||
|
|
||||||
# Setup the Waku node
|
# Start nwaku instance
|
||||||
let
|
let node = await startWakuNode(rng, wakuPort, discv5Port,
|
||||||
nodeKey = crypto.PrivateKey.random(Secp256k1, rng[]).get()
|
requiredConnectedPeers)
|
||||||
ip = parseIpAddress("0.0.0.0")
|
|
||||||
flags = CapabilitiesBitfield.init(lightpush = false, filter = false,
|
|
||||||
store = false, 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()
|
|
||||||
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.peerStore[
|
|
||||||
ConnectionBook].book.values().countIt(it == Connected)
|
|
||||||
if numConnectedPeers >= requiredConnectedPeers:
|
|
||||||
notice "Node is ready", connectedPeers = numConnectedPeers,
|
|
||||||
required = requiredConnectedPeers
|
|
||||||
break
|
|
||||||
notice "Waiting for the node to be ready",
|
|
||||||
connectedPeers = numConnectedPeers, required = requiredConnectedPeers
|
|
||||||
await sleepAsync(5000)
|
|
||||||
|
|
||||||
# Perform the handshake
|
# Perform the handshake
|
||||||
initiatorHSResult = await initiatorHandshake(rng, node, pubSubTopic,
|
initiatorHSResult = await initiatorHandshake(rng, node, pubSubTopic,
|
||||||
|
|
|
@ -1,12 +1,9 @@
|
||||||
import
|
import
|
||||||
std/[tables, times, sequtils],
|
|
||||||
stew/shims/net,
|
|
||||||
chronicles,
|
chronicles,
|
||||||
chronos,
|
chronos,
|
||||||
confutils,
|
confutils,
|
||||||
libp2p/crypto/crypto,
|
libp2p/crypto/crypto,
|
||||||
eth/[keys, p2p/discoveryv5/enr],
|
eth/[keys, p2p/discoveryv5/enr]
|
||||||
nimcrypto/utils
|
|
||||||
|
|
||||||
import
|
import
|
||||||
waku/[waku_core, waku_node, waku_enr, waku_discv5],
|
waku/[waku_core, waku_node, waku_enr, waku_discv5],
|
||||||
|
@ -16,15 +13,8 @@ import
|
||||||
waku/common/[logging, protobuf]
|
waku/common/[logging, protobuf]
|
||||||
|
|
||||||
import ../../waku_handshake_utils
|
import ../../waku_handshake_utils
|
||||||
|
import ../../waku_node
|
||||||
|
|
||||||
const bootstrapNode = "enr:-P-4QGVNANzbhCI49du6Moyw98AjuMhKoOpE_Jges9JlCq-I" &
|
|
||||||
"CAVadktjfcNpuhQgT0g1cu86_S3nbM7eYkCsqDAQG7UBgmlkgnY0" &
|
|
||||||
"gmlwhI_G-a6KbXVsdGlhZGRyc7hgAC02KG5vZGUtMDEuZG8tYW1z" &
|
|
||||||
"My5zdGF0dXMucHJvZC5zdGF0dXNpbS5uZXQGdl8ALzYobm9kZS0w" &
|
|
||||||
"MS5kby1hbXMzLnN0YXR1cy5wcm9kLnN0YXR1c2ltLm5ldAYBu94D" &
|
|
||||||
"iXNlY3AyNTZrMaECoVyonsTGEQvVioM562Q1fjzTb_vKD152PPId" &
|
|
||||||
"sV7sM6SDdGNwgnZfg3VkcIIjKIV3YWt1Mg8"
|
|
||||||
# careful if running pub and sub in the same machine
|
|
||||||
const
|
const
|
||||||
wakuPort = 60000
|
wakuPort = 60000
|
||||||
discv5Port = 9000
|
discv5Port = 9000
|
||||||
|
@ -59,61 +49,9 @@ proc exampleNwakuAgentA(rng: ref HmacDrbgContext) {.async.} =
|
||||||
agentAHS = initHS(agentAInfo, qr, true)
|
agentAHS = initHS(agentAInfo, qr, true)
|
||||||
agentAHSResult: HandshakeResult
|
agentAHSResult: HandshakeResult
|
||||||
|
|
||||||
# Setup the Waku node
|
# Start nwaku instance
|
||||||
let
|
let node = await startWakuNode(rng, wakuPort, discv5Port,
|
||||||
nodeKey = crypto.PrivateKey.random(Secp256k1, rng[]).get()
|
requiredConnectedPeers)
|
||||||
ip = parseIpAddress("0.0.0.0")
|
|
||||||
flags = CapabilitiesBitfield.init(lightpush = false, filter = false,
|
|
||||||
store = false, 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()
|
|
||||||
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.peerStore[
|
|
||||||
ConnectionBook].book.values().countIt(it == Connected)
|
|
||||||
if numConnectedPeers >= requiredConnectedPeers:
|
|
||||||
notice "Node is ready", connectedPeers = numConnectedPeers,
|
|
||||||
required = requiredConnectedPeers
|
|
||||||
break
|
|
||||||
notice "Waiting for the node to be ready",
|
|
||||||
connectedPeers = numConnectedPeers, required = requiredConnectedPeers
|
|
||||||
await sleepAsync(5000)
|
|
||||||
|
|
||||||
# Perform the handshake
|
# Perform the handshake
|
||||||
agentAHSResult = await initiatorHandshake(rng, node, pubSubTopic, contentTopic,
|
agentAHSResult = await initiatorHandshake(rng, node, pubSubTopic, contentTopic,
|
||||||
|
|
|
@ -1,12 +1,10 @@
|
||||||
import
|
import
|
||||||
std/[tables, times, sequtils],
|
|
||||||
stew/shims/net,
|
stew/shims/net,
|
||||||
chronicles,
|
chronicles,
|
||||||
chronos,
|
chronos,
|
||||||
confutils,
|
confutils,
|
||||||
libp2p/crypto/crypto,
|
libp2p/crypto/crypto,
|
||||||
eth/[keys, p2p/discoveryv5/enr],
|
eth/[keys, p2p/discoveryv5/enr]
|
||||||
nimcrypto/utils
|
|
||||||
|
|
||||||
import
|
import
|
||||||
waku/[waku_core, waku_node, waku_enr, waku_discv5],
|
waku/[waku_core, waku_node, waku_enr, waku_discv5],
|
||||||
|
@ -16,14 +14,8 @@ import
|
||||||
waku/common/[logging, protobuf]
|
waku/common/[logging, protobuf]
|
||||||
|
|
||||||
import ../../waku_handshake_utils
|
import ../../waku_handshake_utils
|
||||||
|
import ../../waku_node
|
||||||
|
|
||||||
const bootstrapNode = "enr:-P-4QGVNANzbhCI49du6Moyw98AjuMhKoOpE_Jges9JlCq-I" &
|
|
||||||
"CAVadktjfcNpuhQgT0g1cu86_S3nbM7eYkCsqDAQG7UBgmlkgnY0" &
|
|
||||||
"gmlwhI_G-a6KbXVsdGlhZGRyc7hgAC02KG5vZGUtMDEuZG8tYW1z" &
|
|
||||||
"My5zdGF0dXMucHJvZC5zdGF0dXNpbS5uZXQGdl8ALzYobm9kZS0w" &
|
|
||||||
"MS5kby1hbXMzLnN0YXR1cy5wcm9kLnN0YXR1c2ltLm5ldAYBu94D" &
|
|
||||||
"iXNlY3AyNTZrMaECoVyonsTGEQvVioM562Q1fjzTb_vKD152PPId" &
|
|
||||||
"sV7sM6SDdGNwgnZfg3VkcIIjKIV3YWt1Mg8"
|
|
||||||
const
|
const
|
||||||
wakuPort = 50000
|
wakuPort = 50000
|
||||||
discv5Port = 8000
|
discv5Port = 8000
|
||||||
|
@ -68,68 +60,9 @@ proc exampleNwakuAgentB(rng: ref HmacDrbgContext) {.async.} =
|
||||||
agentBMessageNametag: MessageNametag
|
agentBMessageNametag: MessageNametag
|
||||||
agentBHSResult: HandshakeResult
|
agentBHSResult: HandshakeResult
|
||||||
|
|
||||||
# Setup the Waku node
|
# Start nwaku instance
|
||||||
let
|
let node = await startWakuNode(rng, wakuPort, discv5Port,
|
||||||
nodeKey = crypto.PrivateKey.random(Secp256k1, rng[])[]
|
requiredConnectedPeers)
|
||||||
ip = parseIpAddress("0.0.0.0")
|
|
||||||
flags = CapabilitiesBitfield.init(lightpush = false, filter = false,
|
|
||||||
store = false, 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()
|
|
||||||
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.peerStore[
|
|
||||||
ConnectionBook].book.values().countIt(it == Connected)
|
|
||||||
if numConnectedPeers >= requiredConnectedPeers:
|
|
||||||
notice "subscriber is ready", connectedPeers = numConnectedPeers,
|
|
||||||
required = requiredConnectedPeers
|
|
||||||
break
|
|
||||||
notice "waiting to be ready", connectedPeers = numConnectedPeers,
|
|
||||||
required = requiredConnectedPeers
|
|
||||||
await sleepAsync(5000)
|
|
||||||
|
|
||||||
# Make sure it matches the publisher. Use default value
|
# Make sure it matches the publisher. Use default value
|
||||||
# see spec: https://rfc.vac.dev/spec/23/
|
# see spec: https://rfc.vac.dev/spec/23/
|
||||||
|
|
|
@ -0,0 +1,80 @@
|
||||||
|
import
|
||||||
|
std/[tables, sequtils],
|
||||||
|
stew/shims/net,
|
||||||
|
chronicles,
|
||||||
|
chronos,
|
||||||
|
confutils,
|
||||||
|
libp2p/crypto/crypto,
|
||||||
|
eth/[keys, p2p/discoveryv5/enr]
|
||||||
|
|
||||||
|
import
|
||||||
|
waku/[waku_core, waku_node, waku_enr, waku_discv5],
|
||||||
|
waku/node/peer_manager,
|
||||||
|
waku/common/[logging, protobuf]
|
||||||
|
|
||||||
|
proc startWakuNode*(rng: ref HmacDrbgContext,
|
||||||
|
wakuPort, discv5Port, requiredConnectedPeers: int
|
||||||
|
):Future[WakuNode] {.async.} =
|
||||||
|
let
|
||||||
|
bootstrapNode = "enr:-P-4QGVNANzbhCI49du6Moyw98AjuMhKoOpE_Jges9JlCq-I" &
|
||||||
|
"CAVadktjfcNpuhQgT0g1cu86_S3nbM7eYkCsqDAQG7UBgmlkgnY0" &
|
||||||
|
"gmlwhI_G-a6KbXVsdGlhZGRyc7hgAC02KG5vZGUtMDEuZG8tYW1z" &
|
||||||
|
"My5zdGF0dXMucHJvZC5zdGF0dXNpbS5uZXQGdl8ALzYobm9kZS0w" &
|
||||||
|
"MS5kby1hbXMzLnN0YXR1cy5wcm9kLnN0YXR1c2ltLm5ldAYBu94D" &
|
||||||
|
"iXNlY3AyNTZrMaECoVyonsTGEQvVioM562Q1fjzTb_vKD152PPId" &
|
||||||
|
"sV7sM6SDdGNwgnZfg3VkcIIjKIV3YWt1Mg8"
|
||||||
|
nodeKey = crypto.PrivateKey.random(Secp256k1, rng[]).get()
|
||||||
|
ip = parseIpAddress("0.0.0.0")
|
||||||
|
flags = CapabilitiesBitfield.init(lightpush = false, filter = false,
|
||||||
|
store = false, 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()
|
||||||
|
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.peerStore[
|
||||||
|
ConnectionBook].book.values().countIt(it == Connected)
|
||||||
|
if numConnectedPeers >= requiredConnectedPeers:
|
||||||
|
notice "Node is ready", connectedPeers = numConnectedPeers,
|
||||||
|
required = requiredConnectedPeers
|
||||||
|
break
|
||||||
|
notice "Waiting for the node to be ready",
|
||||||
|
connectedPeers = numConnectedPeers, required = requiredConnectedPeers
|
||||||
|
await sleepAsync(5000)
|
||||||
|
|
||||||
|
return node
|
Loading…
Reference in New Issue