Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 17 additions & 12 deletions logos_delivery/logos_delivery.nim
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@
import results, chronos
import std/options # some()/Option for WakuNodeConf.mode
import std/json # newJArray/newJObject/%*/pretty in getAvailableConfigs
import std/strutils

import brokers/broker_context, brokers/broker_interface, brokers/broker_implement
import logos_delivery/api/logos_delivery_interface as logosdelivery_iface
Expand Down Expand Up @@ -48,10 +49,8 @@ proc createNode(conf: WakuNodeConf): Future[Result[Waku, string]] {.async.} =
return ok(wakuRes)

proc initMessagingClient(self: LogosDelivery): Result[MessagingClient, string] =
return ok(
MessagingClient.createUnderContext(
globalBrokerContext(), self.waku.conf.p2pReliability, self.waku.node
)
newMessagingClient(
globalBrokerContext(), self.waku.conf.p2pReliability, self.waku.node
)

proc initReliableChannelManager(
Expand Down Expand Up @@ -147,32 +146,36 @@ BrokerImplement LogosDelivery of LogosDeliveryInterface:
return err("initialize failed: " & e.msg)

method stop(self: LogosDelivery): Future[Result[void, string]] {.async.} =
var errs: seq[string]

if not self.reliableChannelManager.isNil():
try:
await self.reliableChannelManager.stop()
self.reliableChannelManager.close()
self.reliableChannelManager = nil
except CatchableError as e:
return err("ReliableChannelManager stop failed: " & e.msg)
errs.add("ReliableChannelManager stop failed: " & e.msg)
self.reliableChannelManager = nil

if not self.messagingClient.isNil():
try:
await self.messagingClient.stop()
self.messagingClient.close()
self.messagingClient = nil
except CatchableError as e:
return err("MessagingClient stop failed: " & e.msg)
errs.add("MessagingClient stop failed: " & e.msg)
self.messagingClient = nil

if not self.waku.isNil():
try:
(await self.waku.stop()).isOkOr:
return err("Node stop failed: " & $error)
# `Waku` is a plain ref object (not a BrokerImplement) — `stop()` is its
# only teardown; there is no `close()`.
self.waku = nil
(await self.waku.stop()).isOkOr:
errs.add("Node stop failed: " & $error)
except CatchableError as e:
return err("Node stop failed: " & e.msg)
errs.add("Node stop failed: " & e.msg)
self.waku = nil

if errs.len > 0:
return err(errs.join("; "))
return ok()

method shutdown(self: LogosDelivery): Future[Result[void, string]] {.async.} =
Expand All @@ -181,6 +184,8 @@ BrokerImplement LogosDelivery of LogosDeliveryInterface:
method getNodeInfo(
self: LogosDelivery, id: NodeInfoId
): Future[Result[string, string]] {.async.} =
if self.waku.isNil():
return err("not initialized; call startAsNode or startAsClient first")
return ok(self.waku.stateInfo.getNodeInfoItem(id))

method getAvailableConfigs(
Expand Down
23 changes: 16 additions & 7 deletions logos_delivery/messaging/messaging_client.nim
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
import results, chronos
import chronicles
import brokers/broker_implement
import brokers/broker_implement, brokers/broker_context
import logos_delivery/api/messaging_client_interface
import
logos_delivery/waku/api/types,
Expand Down Expand Up @@ -30,12 +30,12 @@ proc stop*(self: MessagingClient) {.async.} =
self.started = false

BrokerImplement MessagingClient of MessagingClientInterface:
proc new*(T: typedesc[MessagingClient], useP2PReliability: bool, node: WakuNode): T =
let sendService = SendService.new(useP2PReliability, node).valueOr:
error "Failed to initialize SendService", error = error
quit(QuitFailure)

let recvService = RecvService.new(node)
proc new*(
T: typedesc[MessagingClient],
node: WakuNode,
sendService: SendService,
recvService: RecvService,
): T =
T(node: node, sendService: sendService, recvService: recvService, started: false)

method subscribe(
Expand Down Expand Up @@ -71,3 +71,12 @@ BrokerImplement MessagingClient of MessagingClientInterface:
asyncSpawn self.sendService.send(deliveryTask)

return ok(requestId)

proc new*(T: typedesc[MessagingClient],
ctx: BrokerContext, useP2PReliability: bool, node: WakuNode
): Result[T, string] =
let sendService = SendService.new(useP2PReliability, node).valueOr:
error "Failed to initialize SendService", error = error
return err("Failed to initialize SendService: " & error)
let recvService = RecvService.new(node)
ok(MessagingClient.createUnderContext(ctx, node, sendService, recvService))
3 changes: 1 addition & 2 deletions logos_delivery/waku/api/api.nim
Original file line number Diff line number Diff line change
Expand Up @@ -44,8 +44,7 @@ proc mountMessagingClient*(w: Waku): Result[void, string] =
if not w.messagingClient.isNil():
return ok()

w.messagingClient =
MessagingClient.createUnderContext(w.brokerCtx, w.conf.p2pReliability, w.node)
w.messagingClient = ?newMessagingClient(w.brokerCtx, w.conf.p2pReliability, w.node)
return ok()

# TODO workaround for legacy use. It will be removed as soon all usage goes through LogosDelivery.
Expand Down
8 changes: 6 additions & 2 deletions tests/node/test_wakunode_health_monitor.nim
Original file line number Diff line number Diff line change
Expand Up @@ -228,7 +228,9 @@ suite "Health Monitor - events":
nodeA.mountMetadata(1, @[0'u16]).expect("Node A failed to mount metadata")
await nodeA.start()

let ds = MessagingClient.new(false, nodeA)
let ds = newMessagingClient(nodeA.brokerCtx, false, nodeA).expect(
"Failed to create MessagingClient"
)
ds.start().expect("Failed to start MessagingClient")

let monitorA = NodeHealthMonitor.new(nodeA)
Expand Down Expand Up @@ -331,7 +333,9 @@ suite "Health Monitor - events":
nodeA.mountMetadata(1, @[0'u16]).expect("Node A failed to mount metadata")
await nodeA.start()

let ds = MessagingClient.new(false, nodeA)
let ds = newMessagingClient(nodeA.brokerCtx, false, nodeA).expect(
"Failed to create MessagingClient"
)
ds.start().expect("Failed to start MessagingClient")
let subMgr = nodeA.subscriptionManager

Expand Down
Loading