logos-messaging/logos-messaging-nim
1
0
Fork 0
mirror of https://github.com/logos-messaging/logos-messaging-nim.git synced 2026-01-07 16:33:08 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor: big refactor to add waku component in libwaku instead of onlu waku node (#2658)

Browse Source
This commit is contained in:
Ivan FB 2024-05-03 14:07:15 +02:00 committed by GitHub
parent 853ec1869e
commit 2463527b24
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
16 changed files with 152 additions and 206 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
apps/wakunode2/wakunode2.nim
View File

@ -18,7 +18,7 @@ import
../../waku/common/logging, ../../waku/common/logging,
../../waku/factory/external_config, ../../waku/factory/external_config,
../../waku/factory/networks_config, ../../waku/factory/networks_config,
../../waku/factory/app, ../../waku/factory/waku,
../../waku/node/health_monitor, ../../waku/node/health_monitor,
../../waku/node/waku_metrics, ../../waku/node/waku_metrics,
../../waku/waku_api/rest/builder as rest_server_builder ../../waku/waku_api/rest/builder as rest_server_builder
@ -63,7 +63,7 @@ when isMainModule:
## 5. Start monitoring tools and external interfaces ## 5. Start monitoring tools and external interfaces
## 6. Setup graceful shutdown hooks ## 6. Setup graceful shutdown hooks
const versionString = "version / git commit hash: " & app.git_version const versionString = "version / git commit hash: " & waku.git_version
let confRes = WakuNodeConf.load(version = versionString) let confRes = WakuNodeConf.load(version = versionString)
if confRes.isErr(): if confRes.isErr():
@ -119,7 +119,7 @@ when isMainModule:
else: else:
discard discard
info "Running nwaku node", version = app.git_version info "Running nwaku node", version = waku.git_version
logConfig(conf) logConfig(conf)
# NOTE: {.threadvar.} is used to make the global variable GC safe for the closure uses it # NOTE: {.threadvar.} is used to make the global variable GC safe for the closure uses it
@ -135,25 +135,25 @@ when isMainModule:
error "Starting esential REST server failed.", error = $error error "Starting esential REST server failed.", error = $error
quit(QuitFailure) quit(QuitFailure)
var wakunode2 = App.init(conf).valueOr: var waku = Waku.init(conf).valueOr:
error "App initialization failed", error = error error "Waku initialization failed", error = error
quit(QuitFailure) quit(QuitFailure)
wakunode2.restServer = restServer waku.restServer = restServer
nodeHealthMonitor.setNode(wakunode2.node) nodeHealthMonitor.setNode(waku.node)
wakunode2.startApp().isOkOr: (waitFor startWaku(addr waku)).isOkOr:
error "Starting app failed", error = error error "Starting waku failed", error = error
quit(QuitFailure) quit(QuitFailure)
rest_server_builder.startRestServerProtocolSupport( rest_server_builder.startRestServerProtocolSupport(
restServer, wakunode2.node, wakunode2.wakuDiscv5, conf restServer, waku.node, waku.wakuDiscv5, conf
).isOkOr: ).isOkOr:
error "Starting protocols support REST server failed.", error = $error error "Starting protocols support REST server failed.", error = $error
quit(QuitFailure) quit(QuitFailure)
wakunode2.metricsServer = waku_metrics.startMetricsServerAndLogging(conf).valueOr: waku.metricsServer = waku_metrics.startMetricsServerAndLogging(conf).valueOr:
error "Starting monitoring and external interfaces failed", error = error error "Starting monitoring and external interfaces failed", error = error
quit(QuitFailure) quit(QuitFailure)
@ -163,7 +163,7 @@ when isMainModule:
## Setup shutdown hooks for this process. ## Setup shutdown hooks for this process.
## Stop node gracefully on shutdown. ## Stop node gracefully on shutdown.
proc asyncStopper(node: App) {.async: (raises: [Exception]).} = proc asyncStopper(node: Waku) {.async: (raises: [Exception]).} =
nodeHealthMonitor.setOverallHealth(HealthStatus.SHUTTING_DOWN) nodeHealthMonitor.setOverallHealth(HealthStatus.SHUTTING_DOWN)
await node.stop() await node.stop()
quit(QuitSuccess) quit(QuitSuccess)
@ -174,7 +174,7 @@ when isMainModule:
# workaround for https://github.com/nim-lang/Nim/issues/4057 # workaround for https://github.com/nim-lang/Nim/issues/4057
setupForeignThreadGc() setupForeignThreadGc()
notice "Shutting down after receiving SIGINT" notice "Shutting down after receiving SIGINT"
asyncSpawn asyncStopper(wakunode2) asyncSpawn asyncStopper(waku)
setControlCHook(handleCtrlC) setControlCHook(handleCtrlC)
@ -182,7 +182,7 @@ when isMainModule:
when defined(posix): when defined(posix):
proc handleSigterm(signal: cint) {.noconv.} = proc handleSigterm(signal: cint) {.noconv.} =
notice "Shutting down after receiving SIGTERM" notice "Shutting down after receiving SIGTERM"
asyncSpawn asyncStopper(wakunode2) asyncSpawn asyncStopper(waku)
c_signal(ansi_c.SIGTERM, handleSigterm) c_signal(ansi_c.SIGTERM, handleSigterm)
@ -195,7 +195,7 @@ when isMainModule:
# Not available in -d:release mode # Not available in -d:release mode
writeStackTrace() writeStackTrace()
waitFor wakunode2.stop() waitFor waku.stop()
quit(QuitFailure) quit(QuitFailure)
c_signal(ansi_c.SIGSEGV, handleSigsegv) c_signal(ansi_c.SIGSEGV, handleSigsegv)

55
examples/wakustealthcommitments/node_spec.nim
View File

@ -3,8 +3,8 @@ when (NimMajor, NimMinor) < (1, 4):
else: else:
{.push raises: [].} {.push raises: [].}
import ../../apps/wakunode2/[networks_config, app, external_config] import
import ../../waku/common/logging ../../waku/common/logging, ../../waku/factory/[waku, networks_config, external_config]
import import
std/[options, strutils, os, sequtils], std/[options, strutils, os, sequtils],
stew/shims/net as stewNet, stew/shims/net as stewNet,
@ -15,11 +15,11 @@ import
libp2p/crypto/crypto libp2p/crypto/crypto
export export
networks_config, app, logging, options, strutils, os, sequtils, stewNet, chronicles, networks_config, waku, logging, options, strutils, os, sequtils, stewNet, chronicles,
chronos, metrics, libbacktrace, crypto chronos, metrics, libbacktrace, crypto
proc setup*(): App = proc setup*(): Waku =
const versionString = "version / git commit hash: " & app.git_version const versionString = "version / git commit hash: " & waku.git_version
let rng = crypto.newRng() let rng = crypto.newRng()
let confRes = WakuNodeConf.load(version = versionString) let confRes = WakuNodeConf.load(version = versionString)
@ -48,48 +48,17 @@ proc setup*(): App =
conf.rlnEpochSizeSec = twnClusterConf.rlnEpochSizeSec conf.rlnEpochSizeSec = twnClusterConf.rlnEpochSizeSec
conf.rlnRelayUserMessageLimit = twnClusterConf.rlnRelayUserMessageLimit conf.rlnRelayUserMessageLimit = twnClusterConf.rlnRelayUserMessageLimit
var wakunode2 = App.init(rng, conf) debug "Starting node"
## Peer persistence var waku = Waku.init(conf).valueOr:
let res1 = wakunode2.setupPeerPersistence() error "Waku initialization failed", error = error
if res1.isErr():
error "1/5 Setting up storage failed", error = $res1.error
quit(QuitFailure) quit(QuitFailure)
debug "2/5 Retrieve dynamic bootstrap nodes" (waitFor startWaku(addr waku)).isOkOr:
error "Starting waku failed", error = error
let res3 = wakunode2.setupDyamicBootstrapNodes()
if res3.isErr():
error "2/5 Retrieving dynamic bootstrap nodes failed", error = $res3.error
quit(QuitFailure) quit(QuitFailure)
debug "3/5 Initializing node"
let res4 = wakunode2.setupWakuApp()
if res4.isErr():
error "3/5 Initializing node failed", error = $res4.error
quit(QuitFailure)
debug "4/5 Mounting protocols"
var res5: Result[void, string]
try:
res5 = waitFor wakunode2.setupAndMountProtocols()
if res5.isErr():
error "4/5 Mounting protocols failed", error = $res5.error
quit(QuitFailure)
except Exception:
error "4/5 Mounting protocols failed", error = getCurrentExceptionMsg()
quit(QuitFailure)
debug "5/5 Starting node and mounted protocols"
# set triggerSelf to false, we don't want to process our own stealthCommitments # set triggerSelf to false, we don't want to process our own stealthCommitments
wakunode2.node.wakuRelay.triggerSelf = false waku.node.wakuRelay.triggerSelf = false
let res6 = wakunode2.startApp()
if res6.isErr():
error "5/5 Starting node and protocols failed", error = $res6.error
quit(QuitFailure)
info "Node setup complete" info "Node setup complete"
return wakunode2 return waku

12
examples/wakustealthcommitments/stealth_commitment_protocol.nim
View File

@ -15,7 +15,7 @@ import
export wire_spec, logging export wire_spec, logging
type StealthCommitmentProtocol* = object type StealthCommitmentProtocol* = object
wakuApp: App waku: Waku
contentTopic: string contentTopic: string
spendingKeyPair: StealthCommitmentFFI.KeyPair spendingKeyPair: StealthCommitmentFFI.KeyPair
viewingKeyPair: StealthCommitmentFFI.KeyPair viewingKeyPair: StealthCommitmentFFI.KeyPair
@ -51,10 +51,10 @@ proc sendThruWaku*(
timestamp: getNanosecondTime(time), timestamp: getNanosecondTime(time),
) )
(self.wakuApp.node.wakuRlnRelay.appendRLNProof(message, float64(time))).isOkOr: (self.waku.node.wakuRlnRelay.appendRLNProof(message, float64(time))).isOkOr:
return err("could not append rate limit proof to the message: " & $error) return err("could not append rate limit proof to the message: " & $error)
(await self.wakuApp.node.publish(some(DefaultPubsubTopic), message)).isOkOr: (await self.waku.node.publish(some(DefaultPubsubTopic), message)).isOkOr:
return err("failed to publish message: " & $error) return err("failed to publish message: " & $error)
debug "rate limit proof is appended to the message" debug "rate limit proof is appended to the message"
@ -167,7 +167,7 @@ proc getSCPHandler(self: StealthCommitmentProtocol): SCPHandler =
return handler return handler
proc new*( proc new*(
wakuApp: App, contentTopic = ContentTopic("/wakustealthcommitments/1/app/proto") waku: Waku, contentTopic = ContentTopic("/wakustealthcommitments/1/app/proto")
): Result[StealthCommitmentProtocol, string] = ): Result[StealthCommitmentProtocol, string] =
let spendingKeyPair = StealthCommitmentFFI.generateKeyPair().valueOr: let spendingKeyPair = StealthCommitmentFFI.generateKeyPair().valueOr:
return err("could not generate spending key pair: " & $error) return err("could not generate spending key pair: " & $error)
@ -178,7 +178,7 @@ proc new*(
info "viewing public key", publicKey = viewingKeyPair.publicKey info "viewing public key", publicKey = viewingKeyPair.publicKey
let SCP = StealthCommitmentProtocol( let SCP = StealthCommitmentProtocol(
wakuApp: wakuApp, waku: waku,
contentTopic: contentTopic, contentTopic: contentTopic,
spendingKeyPair: spendingKeyPair, spendingKeyPair: spendingKeyPair,
viewingKeyPair: viewingKeyPair, viewingKeyPair: viewingKeyPair,
@ -192,5 +192,5 @@ proc new*(
except CatchableError: except CatchableError:
error "could not handle SCP message: ", err = getCurrentExceptionMsg() error "could not handle SCP message: ", err = getCurrentExceptionMsg()
wakuApp.node.subscribe((kind: PubsubSub, topic: DefaultPubsubTopic), some(handler)) waku.node.subscribe((kind: PubsubSub, topic: DefaultPubsubTopic), some(handler))
return ok(SCP) return ok(SCP)

6
library/waku_thread/inter_thread_communication/requests/debug_node_request.nim
View File

@ -1,6 +1,6 @@
import std/[options, sequtils, strutils, json] import std/[options, sequtils, strutils, json]
import chronicles, chronos, stew/results, stew/shims/net import chronicles, chronos, stew/results, stew/shims/net
import ../../../../waku/node/waku_node, ../../../alloc import ../../../../waku/factory/waku, ../../../../waku/node/waku_node, ../../../alloc
type DebugNodeMsgType* = enum type DebugNodeMsgType* = enum
RETRIEVE_LISTENING_ADDRESSES RETRIEVE_LISTENING_ADDRESSES
@ -20,13 +20,13 @@ proc getMultiaddresses(node: WakuNode): seq[string] =
return node.info().listenAddresses return node.info().listenAddresses
proc process*( proc process*(
self: ptr DebugNodeRequest, node: WakuNode self: ptr DebugNodeRequest, waku: Waku
): Future[Result[string, string]] {.async.} = ): Future[Result[string, string]] {.async.} =
defer: defer:
destroyShared(self) destroyShared(self)
case self.operation case self.operation
of RETRIEVE_LISTENING_ADDRESSES: of RETRIEVE_LISTENING_ADDRESSES:
return ok($(%*node.getMultiaddresses())) return ok($(%*waku.node.getMultiaddresses()))
return err("unsupported operation in DebugNodeRequest") return err("unsupported operation in DebugNodeRequest")

34
library/waku_thread/inter_thread_communication/requests/node_lifecycle_request.nim
View File

@ -12,7 +12,7 @@ import
../../../../waku/node/peer_manager/peer_manager, ../../../../waku/node/peer_manager/peer_manager,
../../../../waku/waku_core, ../../../../waku/waku_core,
../../../../waku/factory/external_config, ../../../../waku/factory/external_config,
../../../../waku/node/waku_node, ../../../../waku/factory/waku,
../../../../waku/node/config, ../../../../waku/node/config,
../../../../waku/waku_archive/driver/builder, ../../../../waku/waku_archive/driver/builder,
../../../../waku/waku_archive/driver, ../../../../waku/waku_archive/driver,
@ -48,16 +48,12 @@ proc destroyShared(self: ptr NodeLifecycleRequest) =
deallocShared(self[].configJson) deallocShared(self[].configJson)
deallocShared(self) deallocShared(self)
proc createNode(configJson: cstring): Future[Result[WakuNode, string]] {.async.} = proc createWaku(configJson: cstring): Future[Result[Waku, string]] {.async.} =
var conf: WakuNodeConf var conf: WakuNodeConf
var errorResp: string var errorResp: string
try: try:
if not parseConfig( if not parseConfig($configJson, conf, errorResp):
$configJson,
conf,
errorResp,
):
return err(errorResp) return err(errorResp)
except Exception: except Exception:
return err("exception calling parseConfig: " & getCurrentExceptionMsg()) return err("exception calling parseConfig: " & getCurrentExceptionMsg())
@ -69,6 +65,7 @@ proc createNode(configJson: cstring): Future[Result[WakuNode, string]] {.async.}
# The Waku Network config (cluster-id=1) # The Waku Network config (cluster-id=1)
if conf.clusterId == 1: if conf.clusterId == 1:
## TODO: This section is duplicated in wakunode2.nim. We need to move this to a common module
let twnClusterConf = ClusterConf.TheWakuNetworkConf() let twnClusterConf = ClusterConf.TheWakuNetworkConf()
if len(conf.shards) != 0: if len(conf.shards) != 0:
conf.pubsubTopics = conf.shards.mapIt(twnClusterConf.pubsubTopics[it.uint16]) conf.pubsubTopics = conf.shards.mapIt(twnClusterConf.pubsubTopics[it.uint16])
@ -88,31 +85,28 @@ proc createNode(configJson: cstring): Future[Result[WakuNode, string]] {.async.}
conf.rlnEpochSizeSec = twnClusterConf.rlnEpochSizeSec conf.rlnEpochSizeSec = twnClusterConf.rlnEpochSizeSec
conf.rlnRelayUserMessageLimit = twnClusterConf.rlnRelayUserMessageLimit conf.rlnRelayUserMessageLimit = twnClusterConf.rlnRelayUserMessageLimit
let wakuRes = Waku.init(conf).valueOr:
error "waku initialization failed", error = error
return err("Failed setting up Waku: " & $error)
let nodeRes = setupNode(conf).valueOr(): return ok(wakuRes)
error "Failed setting up node", error = error
return err("Failed setting up node: " & $error)
return ok(nodeRes)
proc process*( proc process*(
self: ptr NodeLifecycleRequest, node: ptr WakuNode self: ptr NodeLifecycleRequest, waku: ptr Waku
): Future[Result[string, string]] {.async.} = ): Future[Result[string, string]] {.async.} =
defer: defer:
destroyShared(self) destroyShared(self)
case self.operation case self.operation
of CREATE_NODE: of CREATE_NODE:
let newNodeRes = await createNode(self.configJson) waku[] = (await createWaku(self.configJson)).valueOr:
if newNodeRes.isErr(): return err("error processing createWaku request: " & $error)
return err(newNodeRes.error)
node[] = newNodeRes.get()
of START_NODE: of START_NODE:
await node[].start() (await waku.startWaku()).isOkOr:
return err("problem starting waku: " & $error)
of STOP_NODE: of STOP_NODE:
try: try:
await node[].stop() await waku[].stop()
except Exception: except Exception:
return err("exception stopping node: " & getCurrentExceptionMsg()) return err("exception stopping node: " & getCurrentExceptionMsg())

6
library/waku_thread/inter_thread_communication/requests/peer_manager_request.nim
View File

@ -1,6 +1,6 @@
import std/[options, sequtils, strutils] import std/[options, sequtils, strutils]
import chronicles, chronos, stew/results, stew/shims/net import chronicles, chronos, stew/results, stew/shims/net
import ../../../../waku/node/waku_node, ../../../alloc import ../../../../waku/factory/waku, ../../../../waku/node/waku_node, ../../../alloc
type PeerManagementMsgType* = enum type PeerManagementMsgType* = enum
CONNECT_TO CONNECT_TO
@ -43,14 +43,14 @@ proc connectTo(
return ok() return ok()
proc process*( proc process*(
self: ptr PeerManagementRequest, node: WakuNode self: ptr PeerManagementRequest, waku: Waku
): Future[Result[string, string]] {.async.} = ): Future[Result[string, string]] {.async.} =
defer: defer:
destroyShared(self) destroyShared(self)
case self.operation case self.operation
of CONNECT_TO: of CONNECT_TO:
let ret = node.connectTo($self[].peerMultiAddr, self[].dialTimeout) let ret = waku.node.connectTo($self[].peerMultiAddr, self[].dialTimeout)
if ret.isErr(): if ret.isErr():
return err(ret.error) return err(ret.error)

12
library/waku_thread/inter_thread_communication/requests/protocols/relay_request.nim
View File

@ -2,7 +2,7 @@ import std/[options, sequtils, strutils]
import chronicles, chronos, stew/byteutils, stew/results, stew/shims/net import chronicles, chronos, stew/byteutils, stew/results, stew/shims/net
import import
../../../../../waku/waku_core/message/message, ../../../../../waku/waku_core/message/message,
../../../../../waku/node/waku_node, ../../../../../waku/factory/waku,
../../../../../waku/waku_core/message, ../../../../../waku/waku_core/message,
../../../../../waku/waku_core/time, # Timestamp ../../../../../waku/waku_core/time, # Timestamp
../../../../../waku/waku_core/topics/pubsub_topic, ../../../../../waku/waku_core/topics/pubsub_topic,
@ -79,26 +79,26 @@ proc toWakuMessage(m: ThreadSafeWakuMessage): WakuMessage =
return wakuMessage return wakuMessage
proc process*( proc process*(
self: ptr RelayRequest, node: ptr WakuNode self: ptr RelayRequest, waku: ptr Waku
): Future[Result[string, string]] {.async.} = ): Future[Result[string, string]] {.async.} =
defer: defer:
destroyShared(self) destroyShared(self)
if node.wakuRelay.isNil(): if waku.node.wakuRelay.isNil():
return err("Operation not supported without Waku Relay enabled.") return err("Operation not supported without Waku Relay enabled.")
case self.operation case self.operation
of SUBSCRIBE: of SUBSCRIBE:
# TO DO: properly perform 'subscribe' # TO DO: properly perform 'subscribe'
discard node.wakuRelay.subscribe($self.pubsubTopic, self.relayEventCallback) discard waku.node.wakuRelay.subscribe($self.pubsubTopic, self.relayEventCallback)
of UNSUBSCRIBE: of UNSUBSCRIBE:
# TODO: properly perform 'unsubscribe' # TODO: properly perform 'unsubscribe'
node.wakuRelay.unsubscribeAll($self.pubsubTopic) waku.node.wakuRelay.unsubscribeAll($self.pubsubTopic)
of PUBLISH: of PUBLISH:
let msg = self.message.toWakuMessage() let msg = self.message.toWakuMessage()
let pubsubTopic = $self.pubsubTopic let pubsubTopic = $self.pubsubTopic
let numPeers = await node.wakuRelay.publish(pubsubTopic, msg) let numPeers = await waku.node.wakuRelay.publish(pubsubTopic, msg)
if numPeers == 0: if numPeers == 0:
return err("Message not sent because no peers found.") return err("Message not sent because no peers found.")
elif numPeers > 0: elif numPeers > 0:

8
library/waku_thread/inter_thread_communication/requests/protocols/store_request.nim
View File

@ -1,7 +1,7 @@
import std/[options, sequtils, strutils] import std/[options, sequtils, strutils]
import chronos, stew/results, stew/shims/net import chronos, stew/results, stew/shims/net
import import
../../../../../waku/node/waku_node, ../../../../../waku/factory/waku,
../../../../../waku/waku_archive/driver/builder, ../../../../../waku/waku_archive/driver/builder,
../../../../../waku/waku_archive/driver, ../../../../../waku/waku_archive/driver,
../../../../../waku/waku_archive/retention_policy/builder, ../../../../../waku/waku_archive/retention_policy/builder,
@ -50,20 +50,20 @@ proc destroyShared(self: ptr StoreQueryRequest) =
deallocShared(self) deallocShared(self)
proc process( proc process(
self: ptr StoreQueryRequest, node: ptr WakuNode self: ptr StoreQueryRequest, waku: ptr Waku
): Future[Result[string, string]] {.async.} = ): Future[Result[string, string]] {.async.} =
defer: defer:
destroyShared(self) destroyShared(self)
proc process*( proc process*(
self: ptr StoreRequest, node: ptr WakuNode self: ptr StoreRequest, waku: ptr Waku
): Future[Result[string, string]] {.async.} = ): Future[Result[string, string]] {.async.} =
defer: defer:
deallocShared(self) deallocShared(self)
case self.operation case self.operation
of REMOTE_QUERY: of REMOTE_QUERY:
return await cast[ptr StoreQueryRequest](self[].storeReq).process(node) return await cast[ptr StoreQueryRequest](self[].storeReq).process(waku)
of LOCAL_QUERY: of LOCAL_QUERY:
discard discard
# cast[ptr StoreQueryRequest](request[].reqContent).process(node) # cast[ptr StoreQueryRequest](request[].reqContent).process(node)

14
library/waku_thread/inter_thread_communication/waku_thread_request.nim
View File

@ -5,7 +5,7 @@
import std/json, stew/results import std/json, stew/results
import chronos import chronos
import import
../../../waku/node/waku_node, ../../../waku/factory/waku,
./requests/node_lifecycle_request, ./requests/node_lifecycle_request,
./requests/peer_manager_request, ./requests/peer_manager_request,
./requests/protocols/relay_request, ./requests/protocols/relay_request,
@ -32,7 +32,7 @@ proc createShared*(
return ret return ret
proc process*( proc process*(
T: type InterThreadRequest, request: ptr InterThreadRequest, node: ptr WakuNode T: type InterThreadRequest, request: ptr InterThreadRequest, waku: ptr Waku
): Future[Result[string, string]] {.async.} = ): Future[Result[string, string]] {.async.} =
## Processes the request and deallocates its memory ## Processes the request and deallocates its memory
defer: defer:
@ -43,15 +43,15 @@ proc process*(
let retFut = let retFut =
case request[].reqType case request[].reqType
of LIFECYCLE: of LIFECYCLE:
cast[ptr NodeLifecycleRequest](request[].reqContent).process(node) cast[ptr NodeLifecycleRequest](request[].reqContent).process(waku)
of PEER_MANAGER: of PEER_MANAGER:
cast[ptr PeerManagementRequest](request[].reqContent).process(node[]) cast[ptr PeerManagementRequest](request[].reqContent).process(waku[])
of RELAY: of RELAY:
cast[ptr RelayRequest](request[].reqContent).process(node) cast[ptr RelayRequest](request[].reqContent).process(waku)
of STORE: of STORE:
cast[ptr StoreRequest](request[].reqContent).process(node) cast[ptr StoreRequest](request[].reqContent).process(waku)
of DEBUG: of DEBUG:
cast[ptr DebugNodeRequest](request[].reqContent).process(node[]) cast[ptr DebugNodeRequest](request[].reqContent).process(waku[])
return await retFut return await retFut

6
library/waku_thread/waku_thread.nim
View File

@ -11,7 +11,7 @@ import
stew/results, stew/results,
stew/shims/net stew/shims/net
import import
../../../waku/node/waku_node, ../../../waku/factory/waku,
../events/[json_message_event, json_base_event], ../events/[json_message_event, json_base_event],
./inter_thread_communication/waku_thread_request, ./inter_thread_communication/waku_thread_request,
./inter_thread_communication/waku_thread_response ./inter_thread_communication/waku_thread_response
@ -48,7 +48,7 @@ proc run(ctx: ptr Context) {.thread.} =
## This is the worker thread body. This thread runs the Waku node ## This is the worker thread body. This thread runs the Waku node
## and attends library user requests (stop, connect_to, etc.) ## and attends library user requests (stop, connect_to, etc.)
var node: WakuNode var waku: Waku
while running.load == true: while running.load == true:
## Trying to get a request from the libwaku main thread ## Trying to get a request from the libwaku main thread
@ -57,7 +57,7 @@ proc run(ctx: ptr Context) {.thread.} =
waitFor ctx.reqSignal.wait() waitFor ctx.reqSignal.wait()
let recvOk = ctx.reqChannel.tryRecv(request) let recvOk = ctx.reqChannel.tryRecv(request)
if recvOk == true: if recvOk == true:
let resultResponse = waitFor InterThreadRequest.process(request, addr node) let resultResponse = waitFor InterThreadRequest.process(request, addr waku)
## Converting a `Result` into a thread-safe transferable response type ## Converting a `Result` into a thread-safe transferable response type
let threadSafeResp = InterThreadResponse.createShared(resultResponse) let threadSafeResp = InterThreadResponse.createShared(resultResponse)

34
tests/wakunode2/test_app.nim
View File

@ -11,51 +11,51 @@ import
libp2p/switch libp2p/switch
import ../testlib/common, ../testlib/wakucore, ../testlib/wakunode import ../testlib/common, ../testlib/wakucore, ../testlib/wakunode
include ../../waku/factory/app include ../../waku/factory/waku
suite "Wakunode2 - App": suite "Wakunode2 - Waku":
test "compilation version should be reported": test "compilation version should be reported":
## Given ## Given
let conf = defaultTestWakuNodeConf() let conf = defaultTestWakuNodeConf()
let wakunode2 = App.init(conf).valueOr: let waku = Waku.init(conf).valueOr:
raiseAssert error raiseAssert error
## When ## When
let version = wakunode2.version let version = waku.version
## Then ## Then
check: check:
version == git_version version == git_version
suite "Wakunode2 - App initialization": suite "Wakunode2 - Waku initialization":
test "peer persistence setup should be successfully mounted": test "peer persistence setup should be successfully mounted":
## Given ## Given
var conf = defaultTestWakuNodeConf() var conf = defaultTestWakuNodeConf()
conf.peerPersistence = true conf.peerPersistence = true
let wakunode2 = App.init(conf).valueOr: let waku = Waku.init(conf).valueOr:
raiseAssert error raiseAssert error
check: check:
not wakunode2.node.peerManager.storage.isNil() not waku.node.peerManager.storage.isNil()
test "node setup is successful with default configuration": test "node setup is successful with default configuration":
## Given ## Given
let conf = defaultTestWakuNodeConf() let conf = defaultTestWakuNodeConf()
## When ## When
var wakunode2 = App.init(conf).valueOr: var waku = Waku.init(conf).valueOr:
raiseAssert error raiseAssert error
wakunode2.startApp().isOkOr: (waitFor startWaku(addr waku)).isOkOr:
raiseAssert error raiseAssert error
wakunode2.metricsServer = waku_metrics.startMetricsServerAndLogging(conf).valueOr: waku.metricsServer = waku_metrics.startMetricsServerAndLogging(conf).valueOr:
raiseAssert error raiseAssert error
## Then ## Then
let node = wakunode2.node let node = waku.node
check: check:
not node.isNil() not node.isNil()
node.wakuArchive.isNil() node.wakuArchive.isNil()
@ -64,7 +64,7 @@ suite "Wakunode2 - App initialization":
not node.rendezvous.isNil() not node.rendezvous.isNil()
## Cleanup ## Cleanup
waitFor wakunode2.stop() waitFor waku.stop()
test "app properly handles dynamic port configuration": test "app properly handles dynamic port configuration":
## Given ## Given
@ -72,21 +72,21 @@ suite "Wakunode2 - App initialization":
conf.tcpPort = Port(0) conf.tcpPort = Port(0)
## When ## When
var wakunode2 = App.init(conf).valueOr: var waku = Waku.init(conf).valueOr:
raiseAssert error raiseAssert error
wakunode2.startApp().isOkOr: (waitFor startWaku(addr waku)).isOkOr:
raiseAssert error raiseAssert error
## Then ## Then
let let
node = wakunode2.node node = waku.node
typedNodeEnr = node.enr.toTypedRecord() typedNodeEnr = node.enr.toTypedRecord()
assert typedNodeEnr.isOk(), $typedNodeEnr.error assert typedNodeEnr.isOk(), $typedNodeEnr.error
check: check:
# App started properly # Waku started properly
not node.isNil() not node.isNil()
node.wakuArchive.isNil() node.wakuArchive.isNil()
node.wakuStore.isNil() node.wakuStore.isNil()
@ -97,4 +97,4 @@ suite "Wakunode2 - App initialization":
typedNodeEnr.get().tcp.get() != 0 typedNodeEnr.get().tcp.get() != 0
## Cleanup ## Cleanup
waitFor wakunode2.stop() waitFor waku.stop()

2
waku/factory/node_factory.nim
View File

@ -337,7 +337,7 @@ proc setupProtocols(
proc startNode*( proc startNode*(
node: WakuNode, conf: WakuNodeConf, dynamicBootstrapNodes: seq[RemotePeerInfo] = @[] node: WakuNode, conf: WakuNodeConf, dynamicBootstrapNodes: seq[RemotePeerInfo] = @[]
): Future[Result[void, string]] {.async.} = ): Future[Result[void, string]] {.async: (raises: []).} =
## Start a configured node and all mounted protocols. ## Start a configured node and all mounted protocols.
## Connect to static nodes and start ## Connect to static nodes and start
## keep-alive, if configured. ## keep-alive, if configured.

111
waku/factory/app.nim → waku/factory/waku.nim
View File

@ -46,37 +46,31 @@ import
../../waku/factory/external_config ../../waku/factory/external_config
logScope: logScope:
topics = "wakunode app" topics = "wakunode waku"
# Git version in git describe format (defined at compile time) # Git version in git describe format (defined at compile time)
const git_version* {.strdefine.} = "n/a" const git_version* {.strdefine.} = "n/a"
type type Waku* = object
App* = object version: string
version: string conf: WakuNodeConf
conf: WakuNodeConf rng: ref HmacDrbgContext
rng: ref HmacDrbgContext key: crypto.PrivateKey
key: crypto.PrivateKey
wakuDiscv5*: WakuDiscoveryV5 wakuDiscv5*: WakuDiscoveryV5
dynamicBootstrapNodes: seq[RemotePeerInfo] dynamicBootstrapNodes: seq[RemotePeerInfo]
node: WakuNode node*: WakuNode
restServer*: WakuRestServerRef restServer*: WakuRestServerRef
metricsServer*: MetricsHttpServerRef metricsServer*: MetricsHttpServerRef
AppResult*[T] = Result[T, string] func version*(waku: Waku): string =
waku.version
func node*(app: App): WakuNode =
app.node
func version*(app: App): string =
app.version
## Initialisation ## Initialisation
proc init*(T: type App, conf: WakuNodeConf): Result[App, string] = proc init*(T: type Waku, conf: WakuNodeConf): Result[Waku, string] =
var confCopy = conf var confCopy = conf
let rng = crypto.newRng() let rng = crypto.newRng()
@ -103,7 +97,7 @@ proc init*(T: type App, conf: WakuNodeConf): Result[App, string] =
error "Failed setting up node", error = nodeRes.error error "Failed setting up node", error = nodeRes.error
return err("Failed setting up node: " & nodeRes.error) return err("Failed setting up node: " & nodeRes.error)
var app = App( var waku = Waku(
version: git_version, version: git_version,
conf: confCopy, conf: confCopy,
rng: rng, rng: rng,
@ -112,11 +106,11 @@ proc init*(T: type App, conf: WakuNodeConf): Result[App, string] =
dynamicBootstrapNodes: dynamicBootstrapNodesRes.get(), dynamicBootstrapNodes: dynamicBootstrapNodesRes.get(),
) )
ok(app) ok(waku)
proc getPorts( proc getPorts(
listenAddrs: seq[MultiAddress] listenAddrs: seq[MultiAddress]
): AppResult[tuple[tcpPort, websocketPort: Option[Port]]] = ): Result[tuple[tcpPort, websocketPort: Option[Port]], string] =
var tcpPort, websocketPort = none(Port) var tcpPort, websocketPort = none(Port)
for a in listenAddrs: for a in listenAddrs:
@ -132,9 +126,9 @@ proc getPorts(
return ok((tcpPort: tcpPort, websocketPort: websocketPort)) return ok((tcpPort: tcpPort, websocketPort: websocketPort))
proc getRunningNetConfig(app: App): AppResult[NetConfig] = proc getRunningNetConfig(waku: ptr Waku): Result[NetConfig, string] =
var conf = app.conf var conf = waku[].conf
let (tcpPort, websocketPort) = getPorts(app.node.switch.peerInfo.listenAddrs).valueOr: let (tcpPort, websocketPort) = getPorts(waku[].node.switch.peerInfo.listenAddrs).valueOr:
return err("Could not retrieve ports " & error) return err("Could not retrieve ports " & error)
if tcpPort.isSome(): if tcpPort.isSome():
@ -149,67 +143,62 @@ proc getRunningNetConfig(app: App): AppResult[NetConfig] =
return ok(netConf) return ok(netConf)
proc updateEnr(app: var App, netConf: NetConfig): AppResult[void] = proc updateEnr(waku: ptr Waku, netConf: NetConfig): Result[void, string] =
let record = enrConfiguration(app.conf, netConf, app.key).valueOr: let record = enrConfiguration(waku[].conf, netConf, waku[].key).valueOr:
return err("ENR setup failed: " & error) return err("ENR setup failed: " & error)
if isClusterMismatched(record, app.conf.clusterId): if isClusterMismatched(record, waku[].conf.clusterId):
return err("cluster id mismatch configured shards") return err("cluster id mismatch configured shards")
app.node.enr = record waku[].node.enr = record
return ok() return ok()
proc updateApp(app: var App): AppResult[void] = proc updateWaku(waku: ptr Waku): Result[void, string] =
if app.conf.tcpPort == Port(0) or app.conf.websocketPort == Port(0): if waku[].conf.tcpPort == Port(0) or waku[].conf.websocketPort == Port(0):
let netConf = getRunningNetConfig(app).valueOr: let netConf = getRunningNetConfig(waku).valueOr:
return err("error calling updateNetConfig: " & $error) return err("error calling updateNetConfig: " & $error)
updateEnr(app, netConf).isOkOr: updateEnr(waku, netConf).isOkOr:
return err("error calling updateEnr: " & $error) return err("error calling updateEnr: " & $error)
app.node.announcedAddresses = netConf.announcedAddresses waku[].node.announcedAddresses = netConf.announcedAddresses
printNodeNetworkInfo(app.node) printNodeNetworkInfo(waku[].node)
return ok() return ok()
proc startApp*(app: var App): AppResult[void] = proc startWaku*(waku: ptr Waku): Future[Result[void, string]] {.async: (raises: []).} =
let nodeRes = catch: (await startNode(waku.node, waku.conf, waku.dynamicBootstrapNodes)).isOkOr:
(waitFor startNode(app.node, app.conf, app.dynamicBootstrapNodes)) return err("error while calling startNode: " & $error)
if nodeRes.isErr():
return err("exception starting node: " & nodeRes.error.msg)
nodeRes.get().isOkOr: # Update waku data that is set dynamically on node start
return err("exception starting node: " & error) updateWaku(waku).isOkOr:
# Update app data that is set dynamically on node start
app.updateApp().isOkOr:
return err("Error in updateApp: " & $error) return err("Error in updateApp: " & $error)
## Discv5 ## Discv5
if app.conf.discv5Discovery: if waku[].conf.discv5Discovery:
app.wakuDiscV5 = waku_discv5.setupDiscoveryV5( waku[].wakuDiscV5 = waku_discv5.setupDiscoveryV5(
app.node.enr, app.node.peerManager, app.node.topicSubscriptionQueue, app.conf, waku.node.enr, waku.node.peerManager, waku.node.topicSubscriptionQueue, waku.conf,
app.dynamicBootstrapNodes, app.rng, app.key, waku.dynamicBootstrapNodes, waku.rng, waku.key,
) )
(waitFor app.wakuDiscV5.start()).isOkOr: (await waku.wakuDiscV5.start()).isOkOr:
return err("failed to start waku discovery v5: " & $error) return err("failed to start waku discovery v5: " & $error)
return ok() return ok()
# App shutdown # Waku shutdown
proc stop*(app: App): Future[void] {.async: (raises: [Exception]).} = proc stop*(waku: Waku): Future[void] {.async: (raises: [Exception]).} =
if not app.restServer.isNil(): if not waku.restServer.isNil():
await app.restServer.stop() await waku.restServer.stop()
if not app.metricsServer.isNil(): if not waku.metricsServer.isNil():
await app.metricsServer.stop() await waku.metricsServer.stop()
if not app.wakuDiscv5.isNil(): if not waku.wakuDiscv5.isNil():
await app.wakuDiscv5.stop() await waku.wakuDiscv5.stop()
if not app.node.isNil(): if not waku.node.isNil():
await app.node.stop() await waku.node.stop()

2
waku/node/waku_node.nim
View File

@ -1063,7 +1063,7 @@ proc mountPeerExchange*(node: WakuNode) {.async, raises: [Defect, LPError].} =
proc fetchPeerExchangePeers*( proc fetchPeerExchangePeers*(
node: Wakunode, amount: uint64 node: Wakunode, amount: uint64
): Future[Result[int, string]] {.async, raises: [Defect].} = ): Future[Result[int, string]] {.async: (raises: []).} =
if node.wakuPeerExchange.isNil(): if node.wakuPeerExchange.isNil():
error "could not get peers from px, waku peer-exchange is nil" error "could not get peers from px, waku peer-exchange is nil"
return err("PeerExchange is not mounted") return err("PeerExchange is not mounted")

9
waku/waku.nim
View File

@ -1,9 +0,0 @@
# Waku
#
# Licenses:
# - MIT ([LICENSE-MIT](../LICENSE-MIT) or http://opensource.org/licenses/MIT)
# - APACHEv2 ([LICENSE-APACHEv2](../LICENSE-APACHEv2) or https://www.apache.org/licenses/LICENSE-2.0)
## An implementation of [Waku v2](https://rfc.vac.dev/spec/10/) in nim.
import waku_node as wakunode2
export wakunode2

17
waku/waku_peer_exchange/protocol.nim
View File

@ -54,7 +54,7 @@ type
proc request*( proc request*(
wpx: WakuPeerExchange, numPeers: uint64, conn: Connection wpx: WakuPeerExchange, numPeers: uint64, conn: Connection
): Future[WakuPeerExchangeResult[PeerExchangeResponse]] {.async, gcsafe.} = ): Future[WakuPeerExchangeResult[PeerExchangeResponse]] {.async: (raises: []).} =
let rpc = PeerExchangeRpc(request: PeerExchangeRequest(numPeers: numPeers)) let rpc = PeerExchangeRpc(request: PeerExchangeRequest(numPeers: numPeers))
var buffer: seq[byte] var buffer: seq[byte]
@ -79,15 +79,18 @@ proc request*(
proc request*( proc request*(
wpx: WakuPeerExchange, numPeers: uint64, peer: RemotePeerInfo wpx: WakuPeerExchange, numPeers: uint64, peer: RemotePeerInfo
): Future[WakuPeerExchangeResult[PeerExchangeResponse]] {.async, gcsafe.} = ): Future[WakuPeerExchangeResult[PeerExchangeResponse]] {.async: (raises: []).} =
let connOpt = await wpx.peerManager.dialPeer(peer, WakuPeerExchangeCodec) try:
if connOpt.isNone(): let connOpt = await wpx.peerManager.dialPeer(peer, WakuPeerExchangeCodec)
return err(dialFailure) if connOpt.isNone():
return await wpx.request(numPeers, connOpt.get()) return err(dialFailure)
return await wpx.request(numPeers, connOpt.get())
except CatchableError:
return err("exception dialing peer: " & getCurrentExceptionMsg())
proc request*( proc request*(
wpx: WakuPeerExchange, numPeers: uint64 wpx: WakuPeerExchange, numPeers: uint64
): Future[WakuPeerExchangeResult[PeerExchangeResponse]] {.async, gcsafe.} = ): Future[WakuPeerExchangeResult[PeerExchangeResponse]] {.async: (raises: []).} =
let peerOpt = wpx.peerManager.selectPeer(WakuPeerExchangeCodec) let peerOpt = wpx.peerManager.selectPeer(WakuPeerExchangeCodec)
if peerOpt.isNone(): if peerOpt.isNone():
waku_px_errors.inc(labelValues = [peerNotFoundFailure]) waku_px_errors.inc(labelValues = [peerNotFoundFailure])