mirror of
https://github.com/logos-messaging/logos-delivery.git
synced 2026-07-26 06:23:14 +00:00
Wire KernelInterface onto Waku; non-API broker conversion
Squash of the kernel-wiring work (d86f0651..535fe2c9): - Don't exercise FFI of Brokers - Fix tests and examples to compile; verify product unchanged - bump nim-brokers to v3.1.3 - Events dropListeners changed to async - WIP + finalize KernelInterface <-> Waku wiring
This commit is contained in:
NagyZoltanPeter
parent
8456245714
commit
6bd8873094
1
Makefile
1
Makefile
@ -452,7 +452,6 @@ endif
|
||||
|
||||
# New single-root Logos Delivery library (BrokerFfiApi + all wrappers), built from
|
||||
# ./logos_delivery.nim. Single cross-platform task (no $(BUILD_COMMAND) suffix).
|
||||
# Set SRCGEN=1 to also dump the broker-generated sources (-d:brokerDebug).
|
||||
onelogosdelivery: | build-deps librln
|
||||
$(NIMBLE) --verbose onelogosdelivery logos_delivery.nimble
|
||||
|
||||
|
||||
@ -1,7 +1,6 @@
|
||||
#include <stdio.h>
|
||||
#include <stdlib.h>
|
||||
#include <string.h>
|
||||
#include <argp.h>
|
||||
#include <signal.h>
|
||||
#include <unistd.h>
|
||||
#include <fcntl.h>
|
||||
@ -61,50 +60,69 @@ struct ConfigNode
|
||||
};
|
||||
|
||||
// Arguments parsing
|
||||
static char doc[] = "\nC example that shows how to use the waku library.";
|
||||
static char args_doc[] = "";
|
||||
|
||||
static struct argp_option options[] = {
|
||||
{"host", 'h', "HOST", 0, "IP to listen for for LibP2P traffic. (default: \"0.0.0.0\")"},
|
||||
{"port", 'p', "PORT", 0, "TCP listening port. (default: \"60000\")"},
|
||||
{"key", 'k', "KEY", 0, "P2P node private key as 64 char hex string."},
|
||||
{"relay", 'r', "RELAY", 0, "Enable relay protocol: 1 or 0. (default: 1)"},
|
||||
{"peers", 'a', "PEERS", 0, "Comma-separated list of peer-multiaddress to connect\
|
||||
to. (default: \"\") e.g. \"/ip4/127.0.0.1/tcp/60001/p2p/16Uiu2HAmVFXtAfSj4EiR7mL2KvL4EE2wztuQgUSBoj2Jx2KeXFLN\""},
|
||||
{0}};
|
||||
|
||||
static error_t parse_opt(int key, char *arg, struct argp_state *state)
|
||||
//
|
||||
// Platform-independent hand-rolled parser. Avoids glibc-only <argp.h>, which is
|
||||
// not available on macOS/BSD/Windows. Supports both "--opt value" and
|
||||
// "--opt=value" forms, plus the original short flags ("-h value" etc.).
|
||||
static void print_usage(const char *prog)
|
||||
{
|
||||
printf("\nC example that shows how to use the waku library.\n\n");
|
||||
printf("Usage: %s [options]\n\n", prog);
|
||||
printf(" -h, --host HOST IP to listen for LibP2P traffic. (default: \"0.0.0.0\")\n");
|
||||
printf(" -p, --port PORT TCP listening port. (default: 60000)\n");
|
||||
printf(" -k, --key KEY P2P node private key as 64 char hex string.\n");
|
||||
printf(" -r, --relay RELAY Enable relay protocol: 1 or 0. (default: 1)\n");
|
||||
printf(" -a, --peers PEERS Comma-separated list of peer-multiaddresses to\n");
|
||||
printf(" connect to. (default: \"\")\n");
|
||||
printf(" --help Show this help and exit.\n");
|
||||
}
|
||||
|
||||
struct ConfigNode *cfgNode = (ConfigNode *)state->input;
|
||||
switch (key)
|
||||
// Matches argv[i] against a long ("--name") or short ("-c") option. On match,
|
||||
// returns the option's value: an inline "--name=value", or the following argv
|
||||
// entry (consuming it by advancing *i). Returns nullptr if no match.
|
||||
static const char *match_opt(
|
||||
char **argv, int argc, int *i, const char *longName, char shortName)
|
||||
{
|
||||
const char *a = argv[*i];
|
||||
|
||||
size_t longLen = strlen(longName);
|
||||
if (strncmp(a, "--", 2) == 0 && strncmp(a + 2, longName, longLen) == 0)
|
||||
{
|
||||
case 'h':
|
||||
snprintf(cfgNode->host, 128, "%s", arg);
|
||||
break;
|
||||
case 'p':
|
||||
cfgNode->port = atoi(arg);
|
||||
break;
|
||||
case 'k':
|
||||
snprintf(cfgNode->key, 128, "%s", arg);
|
||||
break;
|
||||
case 'r':
|
||||
cfgNode->relay = atoi(arg);
|
||||
break;
|
||||
case 'a':
|
||||
snprintf(cfgNode->peers, 2048, "%s", arg);
|
||||
break;
|
||||
case ARGP_KEY_ARG:
|
||||
if (state->arg_num >= 1) /* Too many arguments. */
|
||||
argp_usage(state);
|
||||
break;
|
||||
case ARGP_KEY_END:
|
||||
break;
|
||||
default:
|
||||
return ARGP_ERR_UNKNOWN;
|
||||
if (a[2 + longLen] == '=')
|
||||
return a + 2 + longLen + 1; // --name=value
|
||||
if (a[2 + longLen] == '\0' && *i + 1 < argc)
|
||||
return argv[++(*i)]; // --name value
|
||||
}
|
||||
|
||||
return 0;
|
||||
if (a[0] == '-' && a[1] == shortName && a[2] == '\0' && *i + 1 < argc)
|
||||
return argv[++(*i)]; // -c value
|
||||
|
||||
return nullptr;
|
||||
}
|
||||
|
||||
// Returns true on success, false on unknown/invalid arguments.
|
||||
static bool parse_args(int argc, char **argv, struct ConfigNode *cfgNode)
|
||||
{
|
||||
for (int i = 1; i < argc; ++i)
|
||||
{
|
||||
const char *val;
|
||||
if (strcmp(argv[i], "--help") == 0)
|
||||
return false;
|
||||
else if ((val = match_opt(argv, argc, &i, "host", 'h')))
|
||||
snprintf(cfgNode->host, 128, "%s", val);
|
||||
else if ((val = match_opt(argv, argc, &i, "port", 'p')))
|
||||
cfgNode->port = atoi(val);
|
||||
else if ((val = match_opt(argv, argc, &i, "key", 'k')))
|
||||
snprintf(cfgNode->key, 128, "%s", val);
|
||||
else if ((val = match_opt(argv, argc, &i, "relay", 'r')))
|
||||
cfgNode->relay = atoi(val);
|
||||
else if ((val = match_opt(argv, argc, &i, "peers", 'a')))
|
||||
snprintf(cfgNode->peers, 2048, "%s", val);
|
||||
else
|
||||
return false; // unknown argument
|
||||
}
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
void event_handler(const char *msg, size_t len)
|
||||
@ -129,8 +147,6 @@ auto cify(F &&f)
|
||||
};
|
||||
}
|
||||
|
||||
static struct argp argp = {options, parse_opt, args_doc, doc, 0, 0, 0};
|
||||
|
||||
// Beginning of UI program logic
|
||||
|
||||
enum PROGRAM_STATE
|
||||
@ -254,8 +270,9 @@ int main(int argc, char **argv)
|
||||
cfgNode.port = 60000;
|
||||
cfgNode.relay = 1;
|
||||
|
||||
if (argp_parse(&argp, argc, argv, 0, 0, &cfgNode) == ARGP_ERR_UNKNOWN)
|
||||
if (!parse_args(argc, argv, &cfgNode))
|
||||
{
|
||||
print_usage(argv[0]);
|
||||
show_help_and_exit();
|
||||
}
|
||||
|
||||
|
||||
@ -1,6 +1,7 @@
|
||||
import std/json
|
||||
import chronos, chronicles, results, ffi
|
||||
import
|
||||
logos_delivery/api/messaging_client_interface,
|
||||
logos_delivery/waku/factory/waku,
|
||||
logos_delivery/waku/node/waku_node,
|
||||
logos_delivery/waku/api/[api, types],
|
||||
|
||||
@ -15,21 +15,3 @@ export logos_delivery_interface
|
||||
import logos_delivery/logos_delivery
|
||||
|
||||
import brokers/api_library # registerBrokerLibrary
|
||||
|
||||
# `git_version` is exported as a `{.strdefine.}` by several modules in the graph
|
||||
# (waku.nim, waku_node.nim, nim-ffi), so it's ambiguous unqualified. Pin to
|
||||
# waku's and expose an unambiguous local const for registerBrokerLibrary; the
|
||||
# build injects `-d:git_version="$(git describe …)"`.
|
||||
const ldGitVersion = waku.git_version
|
||||
|
||||
registerBrokerLibrary:
|
||||
name:
|
||||
"logosdelivery"
|
||||
version:
|
||||
ldGitVersion
|
||||
mainClass:
|
||||
LogosDeliveryInterface
|
||||
initializeRequest:
|
||||
StartAsClient
|
||||
shutdownRequest:
|
||||
Shutdown
|
||||
|
||||
@ -69,7 +69,7 @@ requires "https://github.com/logos-messaging/nim-ffi#v0.1.3"
|
||||
|
||||
requires "https://github.com/logos-messaging/nim-sds.git#b12f5ee07c5b764303b51fb948b32a4ade1de3b5"
|
||||
|
||||
requires "https://github.com/NagyZoltanPeter/nim-brokers.git#cf5ee65cc20211068d7191de7e5e177c0dc212fa"
|
||||
requires "https://github.com/NagyZoltanPeter/nim-brokers.git#3eab670389aac073cd9d820a592d54898f2973dc"
|
||||
|
||||
requires "https://github.com/vacp2p/nim-lsquic.git#v0.5.1"
|
||||
requires "https://github.com/vacp2p/nim-jwt.git#057ec95eb5af0eea9c49bfe9025b3312c95dc5f2"
|
||||
@ -530,27 +530,11 @@ task onelogosdelivery,
|
||||
|
||||
# Full --out path keeps the shared lib in outdir (a bare basename lands in cwd).
|
||||
exec "nim c" &
|
||||
" -d:BrokerFfiApi" &
|
||||
" -d:BrokerFfiApiGenPy -d:BrokerFfiApiGenRust -d:BrokerFfiApiGenGo" &
|
||||
" --threads:on --app:lib --opt:speed --mm:refc" &
|
||||
" -d:metrics -d:discv5_protocol_id=d5waku" &
|
||||
" --nimMainPrefix:logosdelivery" & " --out:" & outdir & "/liblogosdelivery" & libExt &
|
||||
srcGenFlag & " " & getMyCPU() & getNimParams() & " ./logos_delivery.nim"
|
||||
|
||||
# The generated FFI wrappers land next to the source (repo root): brokers v3.1.2's
|
||||
# `-d:BrokerFfiApiOutDir` is referenced but never declared `{.strdefine.}`, so it
|
||||
# cannot redirect them. Move them into outdir as a post-build step to keep the
|
||||
# repo root clean.
|
||||
for f in [
|
||||
"logosdelivery.h", "logosdelivery.hpp", "logosdelivery.py", "logosdelivery.cddl",
|
||||
"logosdeliveryConfig.cmake", "logosdeliveryConfigVersion.cmake",
|
||||
]:
|
||||
if fileExists(f):
|
||||
mvFile(f, outdir & "/" & f)
|
||||
for d in ["logosdelivery_rs", "logosdelivery_go"]:
|
||||
if dirExists(d):
|
||||
mvDir(d, outdir & "/" & d)
|
||||
|
||||
### Formatting tasks
|
||||
|
||||
task nphchanges, "Run nph on .nim/.nims/.nimble files changed on this branch/PR":
|
||||
|
||||
@ -23,7 +23,7 @@ import
|
||||
|
||||
export types
|
||||
|
||||
BrokerInterface(API, KernelInterface):
|
||||
BrokerInterface(KernelInterface):
|
||||
EventBroker:
|
||||
type ReceivedMessage = object
|
||||
## Inbound relay/filter message delivery (replaces set_event_callback).
|
||||
|
||||
@ -15,7 +15,7 @@ import ./reliable_channel_manager_interface as ireliablechannelmanager_iface
|
||||
|
||||
export ikernel_iface, imessagingclient_iface, ireliablechannelmanager_iface
|
||||
|
||||
BrokerInterface(API, LogosDeliveryInterface):
|
||||
BrokerInterface(LogosDeliveryInterface):
|
||||
EventBroker:
|
||||
type ConnectionStatusChangeEvent* = object
|
||||
connectionStatus*: ConnectionStatus
|
||||
|
||||
@ -9,7 +9,7 @@ import brokers/broker_interface
|
||||
import logos_delivery/api/types
|
||||
export types
|
||||
|
||||
BrokerInterface(API, MessagingClientInterface):
|
||||
BrokerInterface(MessagingClientInterface):
|
||||
EventBroker:
|
||||
# Event emitted when a message is sent to the network
|
||||
type MessageSentEvent* = object
|
||||
|
||||
@ -8,7 +8,7 @@ import brokers/broker_interface
|
||||
import logos_delivery/api/types
|
||||
export types
|
||||
|
||||
BrokerInterface(API, ReliableChannelManagerInterface):
|
||||
BrokerInterface(ReliableChannelManagerInterface):
|
||||
EventBroker:
|
||||
type ChannelMessageReceivedEvent* = object
|
||||
channelId*: ChannelId
|
||||
|
||||
@ -140,14 +140,14 @@ proc tryFinalizeChannelReq(
|
||||
self.channelReqs.del(channelReqId)
|
||||
|
||||
if state.failedCount > 0:
|
||||
await ChannelMessageErrorEvent.emit(
|
||||
ChannelMessageErrorEvent.emit(
|
||||
self.brokerCtx,
|
||||
channelId = self.channelId,
|
||||
requestId = channelReqId,
|
||||
error = "one or more segments failed",
|
||||
)
|
||||
else:
|
||||
await ChannelMessageSentEvent.emit(
|
||||
ChannelMessageSentEvent.emit(
|
||||
self.brokerCtx, channelId = self.channelId, requestId = channelReqId
|
||||
)
|
||||
|
||||
@ -240,7 +240,7 @@ proc onReadyToSend(
|
||||
let encRes = await Encrypt.request(m)
|
||||
let encrypted = encRes.valueOr:
|
||||
### TODO: Emitting of events from another layer is not completly ok to do so.
|
||||
await mci.MessageErrorEvent.emit(
|
||||
mci.MessageErrorEvent.emit(
|
||||
self.brokerCtx,
|
||||
requestId = channelReqId,
|
||||
messageHash = "",
|
||||
@ -272,7 +272,7 @@ proc onReadyToSend(
|
||||
|
||||
let messagingReqId = sendRes.valueOr:
|
||||
### TODO: Emitting of events from another layer is not completly ok to do so.
|
||||
await mci.MessageErrorEvent.emit(
|
||||
mci.MessageErrorEvent.emit(
|
||||
self.brokerCtx,
|
||||
requestId = channelReqId,
|
||||
messageHash = "",
|
||||
@ -382,7 +382,7 @@ proc onMessageReceived(
|
||||
let decRes = await Decrypt.request(payload)
|
||||
let plaintext = decRes.valueOr:
|
||||
### TODO: Emitting of events from another layer is not completly ok to do so.
|
||||
await mci.MessageErrorEvent.emit(
|
||||
mci.MessageErrorEvent.emit(
|
||||
self.brokerCtx,
|
||||
requestId = RequestId(""),
|
||||
messageHash = messageHash,
|
||||
@ -397,7 +397,7 @@ proc onMessageReceived(
|
||||
## marker/contentTopic filter already ran, so surface it as an error event
|
||||
## rather than dropping it silently.
|
||||
let deliverable = (await self.sdsHandler.handleIncoming(plaintextBytes)).valueOr:
|
||||
await mci.MessageErrorEvent.emit(
|
||||
mci.MessageErrorEvent.emit(
|
||||
self.brokerCtx,
|
||||
requestId = RequestId(""),
|
||||
messageHash = messageHash,
|
||||
|
||||
@ -27,7 +27,6 @@ import
|
||||
import tools/confutils/[cli_args, config_option_meta] # WakuNodeConf (+ .load)
|
||||
|
||||
type LogosDelivery* = ref object of LogosDeliveryInterface
|
||||
kernelI: KernelInterface ## lazily-built sub-interface caches (keep them reachable)
|
||||
waku: Waku ## the owned node facade (built in initializeRequest)
|
||||
messagingClient: MessagingClient
|
||||
reliableChannelManager: ReliableChannelManager
|
||||
@ -42,25 +41,27 @@ proc createNode(conf: WakuNodeConf): Future[Result[Waku, string]] {.async.} =
|
||||
return err("Failed to handle the configuration: " & error)
|
||||
|
||||
## We are not defining app callbacks at node creation
|
||||
let wakuRes = (await Waku.new(wakuConf)).valueOr:
|
||||
let wakuRes = (await Waku.createUnderContext(globalBrokerContext(), wakuConf)).valueOr:
|
||||
error "waku initialization failed", error = error
|
||||
return err("Failed setting up Waku: " & $error)
|
||||
|
||||
return ok(wakuRes)
|
||||
|
||||
proc initMessagingClient(self: LogosDelivery): Result[MessagingClient, string] =
|
||||
let subCtx = newInstanceCtx(self.brokerCtx)
|
||||
return ok(
|
||||
MessagingClient.createUnderContext(
|
||||
subCtx, self.waku.conf.p2pReliability, self.waku.node
|
||||
globalBrokerContext(), self.waku.conf.p2pReliability, self.waku.node
|
||||
)
|
||||
)
|
||||
|
||||
proc initReliableChannelManager(
|
||||
self: LogosDelivery
|
||||
): Result[ReliableChannelManager, string] =
|
||||
let subCtx = newInstanceCtx(self.brokerCtx)
|
||||
return ok(ReliableChannelManager.createUnderContext(subCtx, self.messagingClient))
|
||||
return ok(
|
||||
ReliableChannelManager.createUnderContext(
|
||||
globalBrokerContext(), self.messagingClient
|
||||
)
|
||||
)
|
||||
|
||||
BrokerImplement LogosDelivery of LogosDeliveryInterface:
|
||||
method kernel(
|
||||
@ -69,7 +70,7 @@ BrokerImplement LogosDelivery of LogosDeliveryInterface:
|
||||
if self.waku.isNil():
|
||||
return err("not initialized; call startAsClient first")
|
||||
|
||||
return err("kernel not yet implemented")
|
||||
return ok(KernelInterface(self.waku))
|
||||
|
||||
method messaging(
|
||||
self: LogosDelivery
|
||||
@ -201,15 +202,8 @@ BrokerImplement LogosDelivery of LogosDeliveryInterface:
|
||||
let asString = pretty(jsonNode)
|
||||
return ok(pretty(jsonNode))
|
||||
|
||||
# FFI factory registration
|
||||
proc setupProviders(ctx: BrokerContext): Result[void, string] =
|
||||
## Called by registerBrokerLibrary on the processing thread: construct the
|
||||
## main facade impl adopting the FFI context, wiring its providers under `ctx`.
|
||||
discard LogosDelivery.createUnderContext(ctx)
|
||||
ok()
|
||||
|
||||
# DI factory registration: non ffi - nim lib users needs it.
|
||||
LogosDeliveryInterface.provideFactory(
|
||||
proc(): Result[LogosDeliveryInterface, string] {.gcsafe.} =
|
||||
ok(LogosDeliveryInterface(LogosDelivery.createUnderContext(NewBrokerContext())))
|
||||
ok(LogosDeliveryInterface(LogosDelivery.createUnderContext(globalBrokerContext())))
|
||||
)
|
||||
|
||||
@ -101,7 +101,7 @@ proc processIncomingMessage(
|
||||
|
||||
let rxMsg = RecvMessage(msgHash: msgHash, rxTime: message.timestamp)
|
||||
self.recentReceivedMsgs.add(rxMsg)
|
||||
await MessageReceivedEvent.emit(self.brokerCtx, msgHash.to0xHex(), message)
|
||||
MessageReceivedEvent.emit(self.brokerCtx, msgHash.to0xHex(), message)
|
||||
return true
|
||||
|
||||
proc checkStore*(self: RecvService) {.async.} =
|
||||
|
||||
@ -178,7 +178,7 @@ proc reportTaskResult(self: SendService, task: DeliveryTask) {.async.} =
|
||||
if not task.propagateEventEmitted:
|
||||
info "Message successfully propagated",
|
||||
requestId = task.requestId, msgHash = task.msgHash.to0xHex()
|
||||
await MessagePropagatedEvent.emit(
|
||||
MessagePropagatedEvent.emit(
|
||||
self.brokerCtx, task.requestId, task.msgHash.to0xHex()
|
||||
)
|
||||
task.propagateEventEmitted = true
|
||||
@ -186,14 +186,14 @@ proc reportTaskResult(self: SendService, task: DeliveryTask) {.async.} =
|
||||
of DeliveryState.SuccessfullyValidated:
|
||||
info "Message successfully sent",
|
||||
requestId = task.requestId, msgHash = task.msgHash.to0xHex()
|
||||
await MessageSentEvent.emit(self.brokerCtx, task.requestId, task.msgHash.to0xHex())
|
||||
MessageSentEvent.emit(self.brokerCtx, task.requestId, task.msgHash.to0xHex())
|
||||
return
|
||||
of DeliveryState.FailedToDeliver:
|
||||
error "Failed to send message",
|
||||
requestId = task.requestId,
|
||||
msgHash = task.msgHash.to0xHex(),
|
||||
error = task.errorDesc
|
||||
await MessageErrorEvent.emit(
|
||||
MessageErrorEvent.emit(
|
||||
self.brokerCtx, task.requestId, task.msgHash.to0xHex(), task.errorDesc
|
||||
)
|
||||
return
|
||||
@ -208,7 +208,7 @@ proc reportTaskResult(self: SendService, task: DeliveryTask) {.async.} =
|
||||
error = "Message too old",
|
||||
age = task.messageAge()
|
||||
task.state = DeliveryState.FailedToDeliver
|
||||
await MessageErrorEvent.emit(
|
||||
MessageErrorEvent.emit(
|
||||
self.brokerCtx,
|
||||
task.requestId,
|
||||
task.msgHash.to0xHex(),
|
||||
|
||||
@ -5,6 +5,7 @@ import chronicles, chronos, libp2p/peerid, results
|
||||
|
||||
import logos_delivery/waku/factory/waku
|
||||
import logos_delivery/messaging/messaging_client
|
||||
import logos_delivery/channels/reliable_channel_manager
|
||||
import logos_delivery/api/messaging_client_interface
|
||||
# brings the interface `send` method into scope: the impl's `method send` in the
|
||||
# BrokerImplement block is not exported, so the call below dispatches through the
|
||||
@ -32,6 +33,39 @@ proc createNode*(conf: WakuNodeConf): Future[Result[Waku, string]] {.async.} =
|
||||
|
||||
return ok(wakuRes)
|
||||
|
||||
# TODO workaround for legacy use. It will be removed as soon all usage goes through LogosDelivery.
|
||||
proc mountMessagingClient*(w: Waku): Result[void, string] =
|
||||
## Construct and attach a `MessagingClient` to the node, wiring its brokers
|
||||
## under the node's own `brokerCtx` so emitted events (MessageSent/Error/
|
||||
## Propagated) reach listeners registered on that same context.
|
||||
if w.isNil() or w.node.isNil():
|
||||
return err("Waku node is not initialized")
|
||||
if not w.messagingClient.isNil():
|
||||
return ok()
|
||||
|
||||
w.messagingClient = MessagingClient.createUnderContext(
|
||||
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.
|
||||
proc mountReliableChannelManager*(w: Waku): Result[void, string] =
|
||||
## Construct and attach a `ReliableChannelManager` to the node, wiring its
|
||||
## brokers under the node's own `brokerCtx` (matching `mountMessagingClient`)
|
||||
## so channel events reach listeners on that same context. Requires the
|
||||
## messaging client to be mounted first.
|
||||
if w.isNil() or w.node.isNil():
|
||||
return err("Waku node is not initialized")
|
||||
if w.messagingClient.isNil():
|
||||
return err("messaging client must be mounted before reliable channel manager")
|
||||
if not w.reliableChannelManager.isNil():
|
||||
return ok()
|
||||
|
||||
w.reliableChannelManager = ReliableChannelManager.createUnderContext(
|
||||
w.brokerCtx, MessagingClientInterface(w.messagingClient)
|
||||
)
|
||||
return ok()
|
||||
|
||||
proc checkApiAvailability(w: Waku): Result[void, string] =
|
||||
if w.isNil():
|
||||
return err("Waku node is not initialized")
|
||||
|
||||
@ -1,5 +1,9 @@
|
||||
import brokers/event_broker
|
||||
import logos_delivery/waku/[api/types, waku_core/message, waku_core/topics]
|
||||
|
||||
# TODO: This is a temporary solution to avoid extensive code changes at import sites,
|
||||
# due to the move of API facing events for interface level.
|
||||
# Final solution all sites shall utilize interface rather then direct event imports and emit.
|
||||
from logos_delivery/api/messaging_client_interface import
|
||||
MessageSentEvent, MessageErrorEvent, MessagePropagatedEvent, MessageReceivedEvent
|
||||
export
|
||||
|
||||
@ -2,10 +2,12 @@ import logos_delivery/waku/compat/option_valueor
|
||||
{.push raises: [].}
|
||||
|
||||
import
|
||||
std/[options, sequtils, strformat],
|
||||
std/[options, sequtils, strformat, net, strutils],
|
||||
results,
|
||||
chronicles,
|
||||
chronos,
|
||||
secp256k1,
|
||||
libp2p/protocols/ping,
|
||||
libp2p/protocols/connectivity/relay/relay,
|
||||
libp2p/protocols/connectivity/relay/client,
|
||||
libp2p/wire,
|
||||
@ -20,6 +22,7 @@ import
|
||||
metrics,
|
||||
metrics/chronos_httpserver,
|
||||
brokers/broker_context,
|
||||
brokers/broker_implement,
|
||||
logos_delivery/waku/[
|
||||
waku_core,
|
||||
waku_node,
|
||||
@ -48,7 +51,11 @@ import
|
||||
factory/internal_config,
|
||||
factory/app_callbacks,
|
||||
persistency/persistency,
|
||||
waku_lightpush_legacy/client,
|
||||
waku_store/client as waku_store_client,
|
||||
factory/validator_signed,
|
||||
],
|
||||
logos_delivery/api/kernel_interface,
|
||||
logos_delivery/channels/reliable_channel_manager,
|
||||
logos_delivery/messaging/messaging_client,
|
||||
./waku_conf,
|
||||
@ -60,7 +67,7 @@ logScope:
|
||||
# Git version in git describe format (defined at compile time)
|
||||
const git_version* {.strdefine.} = "n/a"
|
||||
|
||||
type Waku* = ref object
|
||||
type Waku* = ref object of KernelInterface
|
||||
stateInfo*: WakuStateInfo
|
||||
conf*: WakuConf
|
||||
rng*: crypto.Rng
|
||||
@ -74,8 +81,8 @@ type Waku* = ref object
|
||||
|
||||
node*: WakuNode
|
||||
|
||||
# TODO: remove this indirection. Now kept for legacy.
|
||||
healthMonitor*: NodeHealthMonitor
|
||||
|
||||
messagingClient*: MessagingClient
|
||||
|
||||
reliableChannelManager*: ReliableChannelManager
|
||||
@ -83,8 +90,7 @@ type Waku* = ref object
|
||||
restServer*: WakuRestServerRef
|
||||
metricsServer*: MetricsHttpServerRef
|
||||
appCallbacks*: AppCallbacks
|
||||
|
||||
brokerCtx*: BrokerContext
|
||||
## `brokerCtx` is inherited from the `KernelInterface` broker base.
|
||||
|
||||
proc setupSwitchServices(
|
||||
waku: Waku, conf: WakuConf, circuitRelay: Relay, rng: crypto.Rng
|
||||
@ -188,65 +194,6 @@ proc setupAppCallbacks(
|
||||
|
||||
return ok()
|
||||
|
||||
proc new*(
|
||||
T: type Waku, wakuConf: WakuConf, appCallbacks: AppCallbacks = nil
|
||||
): Future[Result[Waku, string]] {.async.} =
|
||||
let rng = crypto.newRng()
|
||||
let brokerCtx = globalBrokerContext()
|
||||
|
||||
logging.setupLog(wakuConf.logLevel, wakuConf.logFormat)
|
||||
|
||||
?wakuConf.validate()
|
||||
wakuConf.logConf()
|
||||
|
||||
let relay = newCircuitRelay(wakuConf.circuitRelayClient)
|
||||
|
||||
let node = (await setupNode(wakuConf, rng, relay)).valueOr:
|
||||
error "Failed setting up node", error = $error
|
||||
return err("Failed setting up node: " & $error)
|
||||
|
||||
let healthMonitor = NodeHealthMonitor.new(node, wakuConf.dnsAddrsNameServers)
|
||||
|
||||
let restServer: WakuRestServerRef =
|
||||
if wakuConf.restServerConf.isSome():
|
||||
let restServer = startRestServerEssentials(
|
||||
healthMonitor, wakuConf.restServerConf.get(), wakuConf.portsShift
|
||||
).valueOr:
|
||||
error "Starting essential REST server failed", error = $error
|
||||
return err("Failed to start essential REST server in Waku.new: " & $error)
|
||||
|
||||
restServer
|
||||
else:
|
||||
nil
|
||||
|
||||
if not restServer.isNil():
|
||||
let boundRestPort = restServer.httpServer.address.port
|
||||
node.ports.rest = boundRestPort.uint16
|
||||
wakuConf.restServerConf.get().port = boundRestPort
|
||||
|
||||
# Set the extMultiAddrsOnly flag so the node knows not to replace explicit addresses
|
||||
node.extMultiAddrsOnly = wakuConf.endpointConf.extMultiAddrsOnly
|
||||
|
||||
node.setupAppCallbacks(wakuConf, appCallbacks, healthMonitor).isOkOr:
|
||||
error "Failed setting up app callbacks", error = error
|
||||
return err("Failed setting up app callbacks: " & $error)
|
||||
|
||||
var waku = Waku(
|
||||
stateInfo: WakuStateInfo.init(node),
|
||||
conf: wakuConf,
|
||||
rng: rng,
|
||||
key: wakuConf.nodeKey,
|
||||
node: node,
|
||||
healthMonitor: healthMonitor,
|
||||
appCallbacks: appCallbacks,
|
||||
restServer: restServer,
|
||||
brokerCtx: brokerCtx,
|
||||
)
|
||||
|
||||
waku.setupSwitchServices(wakuConf, relay, rng)
|
||||
|
||||
ok(waku)
|
||||
|
||||
proc getPorts(
|
||||
listenAddrs: seq[MultiAddress]
|
||||
): Result[tuple[tcpPort, websocketPort, quicPort: Option[Port]], string] =
|
||||
@ -384,6 +331,7 @@ proc startDnsDiscoveryRetryLoop(waku: Waku): Future[void] {.async.} =
|
||||
error "failed to connect to dynamic bootstrap nodes: " & getCurrentExceptionMsg()
|
||||
return
|
||||
|
||||
# Notice this interface to be used only from LogosDelivery, hence not in the interface level.
|
||||
proc start*(waku: Waku): Future[Result[void, string]] {.async: (raises: []).} =
|
||||
if waku.node.started:
|
||||
warn "start: waku node already started"
|
||||
@ -546,6 +494,7 @@ proc start*(waku: Waku): Future[Result[void, string]] {.async: (raises: []).} =
|
||||
|
||||
return ok()
|
||||
|
||||
# Notice this interface to be used only from LogosDelivery, hence not in the interface level.
|
||||
proc stop*(waku: Waku): Future[Result[void, string]] {.async: (raises: []).} =
|
||||
if not waku.node.started:
|
||||
warn "stop: attempting to stop node that isn't running"
|
||||
@ -587,12 +536,499 @@ proc stop*(waku: Waku): Future[Result[void, string]] {.async: (raises: []).} =
|
||||
|
||||
return ok()
|
||||
|
||||
proc isModeCoreAvailable*(waku: Waku): bool =
|
||||
return not waku.node.wakuRelay.isNil()
|
||||
|
||||
proc isModeEdgeAvailable*(waku: Waku): bool =
|
||||
return
|
||||
waku.node.wakuRelay.isNil() and not waku.node.wakuStoreClient.isNil() and
|
||||
not waku.node.wakuFilterClient.isNil() and not waku.node.wakuLightPushClient.isNil()
|
||||
|
||||
{.pop.}
|
||||
# end of `{.push raises: [].}` — kernel impl methods may propagate
|
||||
# CatchableError (the BrokerImplement provider wrappers catch them).
|
||||
|
||||
const FilterOpTimeout = 5.seconds
|
||||
|
||||
BrokerImplement Waku of KernelInterface:
|
||||
## `new` is the BARE constructor (no ctx, no providers). Legacy callers keep
|
||||
## using `Waku.new(...)` unchanged — it is re-emitted verbatim by the macro
|
||||
## and returns a `globalBrokerContext`-bound node exactly as before, with the
|
||||
## kernel request-broker providers left unwired. `Waku.create(...)` /
|
||||
## `Waku.createUnderContext(...)` are additionally generated (async `Result`
|
||||
## shape) to wire the kernel under a fresh per-instance ctx when needed.
|
||||
proc new*(
|
||||
T: type Waku, wakuConf: WakuConf, appCallbacks: AppCallbacks = nil
|
||||
): Future[Result[Waku, string]] {.async.} =
|
||||
let rng = crypto.newRng()
|
||||
let brokerCtx = globalBrokerContext()
|
||||
|
||||
logging.setupLog(wakuConf.logLevel, wakuConf.logFormat)
|
||||
|
||||
?wakuConf.validate()
|
||||
wakuConf.logConf()
|
||||
|
||||
let relay = newCircuitRelay(wakuConf.circuitRelayClient)
|
||||
|
||||
let node = (await setupNode(wakuConf, rng, relay)).valueOr:
|
||||
error "Failed setting up node", error = $error
|
||||
return err("Failed setting up node: " & $error)
|
||||
|
||||
let healthMonitor = NodeHealthMonitor.new(node, wakuConf.dnsAddrsNameServers)
|
||||
|
||||
let restServer: WakuRestServerRef =
|
||||
if wakuConf.restServerConf.isSome():
|
||||
let restServer = startRestServerEssentials(
|
||||
healthMonitor, wakuConf.restServerConf.get(), wakuConf.portsShift
|
||||
).valueOr:
|
||||
error "Starting essential REST server failed", error = $error
|
||||
return err("Failed to start essential REST server in Waku.new: " & $error)
|
||||
|
||||
restServer
|
||||
else:
|
||||
nil
|
||||
|
||||
if not restServer.isNil():
|
||||
let boundRestPort = restServer.httpServer.address.port
|
||||
node.ports.rest = boundRestPort.uint16
|
||||
wakuConf.restServerConf.get().port = boundRestPort
|
||||
|
||||
# Set the extMultiAddrsOnly flag so the node knows not to replace explicit addresses
|
||||
node.extMultiAddrsOnly = wakuConf.endpointConf.extMultiAddrsOnly
|
||||
|
||||
node.setupAppCallbacks(wakuConf, appCallbacks, healthMonitor).isOkOr:
|
||||
error "Failed setting up app callbacks", error = error
|
||||
return err("Failed setting up app callbacks: " & $error)
|
||||
|
||||
var waku = Waku(
|
||||
stateInfo: WakuStateInfo.init(node),
|
||||
conf: wakuConf,
|
||||
rng: rng,
|
||||
key: wakuConf.nodeKey,
|
||||
node: node,
|
||||
healthMonitor: healthMonitor,
|
||||
appCallbacks: appCallbacks,
|
||||
restServer: restServer,
|
||||
brokerCtx: brokerCtx,
|
||||
)
|
||||
|
||||
waku.setupSwitchServices(wakuConf, relay, rng)
|
||||
|
||||
ok(waku)
|
||||
|
||||
# --- topic construction ---
|
||||
method buildContentTopic(
|
||||
self: Waku, appName: string, appVersion: uint32, name: string, encoding: string
|
||||
): Future[Result[ContentTopic, string]] {.async.} =
|
||||
try:
|
||||
return ok(ContentTopic(fmt"/{appName}/{appVersion}/{name}/{encoding}"))
|
||||
except CatchableError as e:
|
||||
return err(e.msg)
|
||||
|
||||
method buildPubsubTopic(
|
||||
self: Waku, topicName: string
|
||||
): Future[Result[PubsubTopic, string]] {.async.} =
|
||||
try:
|
||||
return ok(PubsubTopic(fmt"/waku/2/{topicName}"))
|
||||
except CatchableError as e:
|
||||
return err(e.msg)
|
||||
|
||||
method defaultPubsubTopic(self: Waku): Future[Result[PubsubTopic, string]] {.async.} =
|
||||
return ok(DefaultPubsubTopic)
|
||||
|
||||
# --- relay ---
|
||||
method relayPublish(
|
||||
self: Waku, pubsubTopic: PubsubTopic, message: WakuMessage, timeoutMs: uint32
|
||||
): Future[Result[int, string]] {.async.} =
|
||||
try:
|
||||
if self.node.wakuRelay.isNil():
|
||||
return err("relayPublish: WakuRelay not mounted")
|
||||
|
||||
let numPeers = (await self.node.wakuRelay.publish(pubsubTopic, message)).valueOr:
|
||||
return err($error)
|
||||
|
||||
return ok(numPeers)
|
||||
except CatchableError as e:
|
||||
return err(e.msg)
|
||||
|
||||
method relaySubscribe(
|
||||
self: Waku, pubsubTopic: PubsubTopic
|
||||
): Future[Result[bool, string]] {.async.} =
|
||||
try:
|
||||
if self.node.wakuRelay.isNil():
|
||||
return err("relaySubscribe: WakuRelay not mounted")
|
||||
|
||||
let handler = proc(topic: PubsubTopic, msg: WakuMessage) {.async.} =
|
||||
## Bridge inbound relay traffic to the `ReceivedMessage` kernel event
|
||||
## (replaces libwaku's set_event_callback message path).
|
||||
ReceivedMessage.emit(
|
||||
self.brokerCtx, ReceivedMessage(pubsubTopic: topic, message: msg)
|
||||
)
|
||||
|
||||
self.node.subscribe(
|
||||
(kind: SubscriptionKind.PubsubSub, topic: pubsubTopic),
|
||||
WakuRelayHandler(handler),
|
||||
).isOkOr:
|
||||
return err($error)
|
||||
|
||||
return ok(true)
|
||||
except CatchableError as e:
|
||||
return err(e.msg)
|
||||
|
||||
method relayUnsubscribe(
|
||||
self: Waku, pubsubTopic: PubsubTopic
|
||||
): Future[Result[bool, string]] {.async.} =
|
||||
try:
|
||||
if self.node.wakuRelay.isNil():
|
||||
return err("relayUnsubscribe: WakuRelay not mounted")
|
||||
|
||||
self.node.unsubscribe((kind: SubscriptionKind.PubsubSub, topic: pubsubTopic)).isOkOr:
|
||||
return err($error)
|
||||
|
||||
return ok(true)
|
||||
except CatchableError as e:
|
||||
return err(e.msg)
|
||||
|
||||
method relayAddProtectedShard(
|
||||
self: Waku, clusterId: uint16, shardId: uint16, publicKey: string
|
||||
): Future[Result[bool, string]] {.async.} =
|
||||
try:
|
||||
if self.node.wakuRelay.isNil():
|
||||
return err("relayAddProtectedShard: WakuRelay not mounted")
|
||||
|
||||
let pubKey = SkPublicKey.fromHex(publicKey).valueOr:
|
||||
return err("relayAddProtectedShard: invalid public key: " & $error)
|
||||
|
||||
let protectedShard = ProtectedShard(shard: shardId, key: pubKey)
|
||||
self.node.wakuRelay.addSignedShardsValidator(@[protectedShard], clusterId)
|
||||
return ok(true)
|
||||
except CatchableError as e:
|
||||
return err(e.msg)
|
||||
|
||||
method relayConnectedPeers(
|
||||
self: Waku, pubsubTopic: PubsubTopic
|
||||
): Future[Result[seq[string], string]] {.async.} =
|
||||
try:
|
||||
if self.node.wakuRelay.isNil():
|
||||
return err("relayConnectedPeers: WakuRelay not mounted")
|
||||
|
||||
let connPeers = self.node.wakuRelay.getConnectedPeers(pubsubTopic).valueOr:
|
||||
return err($error)
|
||||
|
||||
return ok(connPeers.mapIt($it))
|
||||
except CatchableError as e:
|
||||
return err(e.msg)
|
||||
|
||||
method relayPeersInMesh(
|
||||
self: Waku, pubsubTopic: PubsubTopic
|
||||
): Future[Result[seq[string], string]] {.async.} =
|
||||
try:
|
||||
if self.node.wakuRelay.isNil():
|
||||
return err("relayPeersInMesh: WakuRelay not mounted")
|
||||
|
||||
let meshPeers = self.node.wakuRelay.getPeersInMesh(pubsubTopic).valueOr:
|
||||
return err($error)
|
||||
|
||||
return ok(meshPeers.mapIt($it))
|
||||
except CatchableError as e:
|
||||
return err(e.msg)
|
||||
|
||||
# --- filter ---
|
||||
method filterSubscribe(
|
||||
self: Waku,
|
||||
pubsubTopic: Option[PubsubTopic],
|
||||
contentTopics: seq[ContentTopic],
|
||||
peer: string,
|
||||
): Future[Result[bool, string]] {.async.} =
|
||||
try:
|
||||
if self.node.wakuFilterClient.isNil():
|
||||
return err("wakuFilterClient is not mounted")
|
||||
|
||||
let subFut = self.node.filterSubscribe(pubsubTopic, contentTopics, peer)
|
||||
if not await subFut.withTimeout(FilterOpTimeout):
|
||||
return err("filter subscription timed out")
|
||||
subFut.read().isOkOr:
|
||||
return err($error)
|
||||
|
||||
return ok(true)
|
||||
except CatchableError as e:
|
||||
return err(e.msg)
|
||||
|
||||
method filterUnsubscribe(
|
||||
self: Waku,
|
||||
pubsubTopic: Option[PubsubTopic],
|
||||
contentTopics: seq[ContentTopic],
|
||||
peer: string,
|
||||
): Future[Result[bool, string]] {.async.} =
|
||||
try:
|
||||
if self.node.wakuFilterClient.isNil():
|
||||
return err("wakuFilterClient is not mounted")
|
||||
|
||||
let unsubFut = self.node.filterUnsubscribe(pubsubTopic, contentTopics, peer)
|
||||
if not await unsubFut.withTimeout(FilterOpTimeout):
|
||||
return err("filter un-subscription timed out")
|
||||
unsubFut.read().isOkOr:
|
||||
return err($error)
|
||||
|
||||
return ok(true)
|
||||
except CatchableError as e:
|
||||
return err(e.msg)
|
||||
|
||||
method filterUnsubscribeAll(
|
||||
self: Waku, peer: string
|
||||
): Future[Result[bool, string]] {.async.} =
|
||||
try:
|
||||
if self.node.wakuFilterClient.isNil():
|
||||
return err("wakuFilterClient is not mounted")
|
||||
|
||||
let unsubFut = self.node.filterUnsubscribeAll(peer)
|
||||
if not await unsubFut.withTimeout(FilterOpTimeout):
|
||||
return err("filter un-subscription all timed out")
|
||||
unsubFut.read().isOkOr:
|
||||
return err($error)
|
||||
|
||||
return ok(true)
|
||||
except CatchableError as e:
|
||||
return err(e.msg)
|
||||
|
||||
# --- lightpush ---
|
||||
method lightpushPublish(
|
||||
self: Waku, pubsubTopic: PubsubTopic, message: WakuMessage, peer: string
|
||||
): Future[Result[string, string]] {.async.} =
|
||||
try:
|
||||
if self.node.wakuLegacyLightpushClient.isNil():
|
||||
return err("wakuLegacyLightpushClient is not mounted")
|
||||
|
||||
let remotePeer = parsePeerInfo(peer).valueOr:
|
||||
return err("lightpushPublish failed to parse peer addr: " & $error)
|
||||
|
||||
let msgHashHex = (
|
||||
await self.node.wakuLegacyLightpushClient.publish(
|
||||
pubsubTopic, message, remotePeer
|
||||
)
|
||||
).valueOr:
|
||||
return err($error)
|
||||
|
||||
return ok(msgHashHex)
|
||||
except CatchableError as e:
|
||||
return err(e.msg)
|
||||
|
||||
# --- store ---
|
||||
method storeQuery(
|
||||
self: Waku, request: StoreQueryRequest, peer: string, timeoutMs: int
|
||||
): Future[Result[StoreQueryResponse, string]] {.async.} =
|
||||
try:
|
||||
if self.node.wakuStoreClient.isNil():
|
||||
return err("wakuStoreClient is not mounted")
|
||||
|
||||
let remotePeer = parsePeerInfo(peer).valueOr:
|
||||
return err("storeQuery failed to parse peer addr: " & $error)
|
||||
|
||||
let queryFut = self.node.wakuStoreClient.query(request, remotePeer)
|
||||
if not await queryFut.withTimeout(timeoutMs.milliseconds):
|
||||
return err("storeQuery timed out")
|
||||
|
||||
let queryResponse = queryFut.read().valueOr:
|
||||
return err("storeQuery failed: " & $error)
|
||||
|
||||
return ok(queryResponse)
|
||||
except CatchableError as e:
|
||||
return err(e.msg)
|
||||
|
||||
# --- peer management ---
|
||||
method connect(
|
||||
self: Waku, peers: seq[string], timeoutMs: uint32
|
||||
): Future[Result[bool, string]] {.async.} =
|
||||
try:
|
||||
await self.node.connectToNodes(peers.mapIt(strip(it)), source = "static")
|
||||
return ok(true)
|
||||
except CatchableError as e:
|
||||
return err(e.msg)
|
||||
|
||||
method disconnectPeerById(
|
||||
self: Waku, peerId: string
|
||||
): Future[Result[bool, string]] {.async.} =
|
||||
try:
|
||||
let pId = PeerId.init(peerId).valueOr:
|
||||
return err($error)
|
||||
await self.node.peerManager.disconnectNode(pId)
|
||||
return ok(true)
|
||||
except CatchableError as e:
|
||||
return err(e.msg)
|
||||
|
||||
method disconnectAllPeers(self: Waku): Future[Result[bool, string]] {.async.} =
|
||||
try:
|
||||
await self.node.peerManager.disconnectAllPeers()
|
||||
return ok(true)
|
||||
except CatchableError as e:
|
||||
return err(e.msg)
|
||||
|
||||
method dialPeer(
|
||||
self: Waku, peerAddr: string, protocol: string, timeoutMs: int
|
||||
): Future[Result[bool, string]] {.async.} =
|
||||
try:
|
||||
let remotePeerInfo = parsePeerInfo(peerAddr).valueOr:
|
||||
return err($error)
|
||||
let conn = await self.node.peerManager.dialPeer(remotePeerInfo, protocol)
|
||||
if conn.isNone():
|
||||
return err("failed dialing peer")
|
||||
return ok(true)
|
||||
except CatchableError as e:
|
||||
return err(e.msg)
|
||||
|
||||
method dialPeerById(
|
||||
self: Waku, peerId: string, protocol: string, timeoutMs: int
|
||||
): Future[Result[bool, string]] {.async.} =
|
||||
try:
|
||||
let pId = PeerId.init(peerId).valueOr:
|
||||
return err($error)
|
||||
let conn = await self.node.peerManager.dialPeer(pId, protocol)
|
||||
if conn.isNone():
|
||||
return err("failed dialing peer")
|
||||
return ok(true)
|
||||
except CatchableError as e:
|
||||
return err(e.msg)
|
||||
|
||||
method peerIdsFromPeerstore(
|
||||
self: Waku
|
||||
): Future[Result[seq[string], string]] {.async.} =
|
||||
try:
|
||||
return ok(self.node.peerManager.switch.peerStore.peers().mapIt($it.peerId))
|
||||
except CatchableError as e:
|
||||
return err(e.msg)
|
||||
|
||||
method connectedPeersInfo(self: Waku): Future[Result[seq[string], string]] {.async.} =
|
||||
try:
|
||||
return ok(
|
||||
self.node.peerManager.switch.peerStore
|
||||
.peers()
|
||||
.filterIt(it.connectedness == Connected)
|
||||
.mapIt($it.peerId)
|
||||
)
|
||||
except CatchableError as e:
|
||||
return err(e.msg)
|
||||
|
||||
method connectedPeers(self: Waku): Future[Result[seq[string], string]] {.async.} =
|
||||
try:
|
||||
let (inPeerIds, outPeerIds) = self.node.peerManager.connectedPeers()
|
||||
return ok(concat(inPeerIds, outPeerIds).mapIt($it))
|
||||
except CatchableError as e:
|
||||
return err(e.msg)
|
||||
|
||||
method peerIdsByProtocol(
|
||||
self: Waku, protocol: string
|
||||
): Future[Result[seq[string], string]] {.async.} =
|
||||
try:
|
||||
return ok(
|
||||
self.node.peerManager.switch.peerStore
|
||||
.peers(protocol)
|
||||
.filterIt(it.connectedness == Connected)
|
||||
.mapIt($it.peerId)
|
||||
)
|
||||
except CatchableError as e:
|
||||
return err(e.msg)
|
||||
|
||||
# --- discovery ---
|
||||
method dnsDiscovery(
|
||||
self: Waku, enrTreeUrl: string, nameServer: string, timeoutMs: int
|
||||
): Future[Result[seq[string], string]] {.async.} =
|
||||
try:
|
||||
let dnsNameServers = @[parseIpAddress(nameServer)]
|
||||
let discoveredPeers = (
|
||||
await retrieveDynamicBootstrapNodes(enrTreeUrl, dnsNameServers)
|
||||
).valueOr:
|
||||
return err("failed discovering peers from DNS: " & $error)
|
||||
|
||||
var multiAddresses = newSeq[string]()
|
||||
for discPeer in discoveredPeers:
|
||||
for address in discPeer.addrs:
|
||||
multiAddresses.add($address & "/p2p/" & $discPeer)
|
||||
|
||||
return ok(multiAddresses)
|
||||
except CatchableError as e:
|
||||
return err(e.msg)
|
||||
|
||||
method discv5UpdateBootnodes(
|
||||
self: Waku, bootnodes: seq[string]
|
||||
): Future[Result[bool, string]] {.async.} =
|
||||
try:
|
||||
if self.wakuDiscv5.isNil():
|
||||
return err("discv5 not started")
|
||||
let jsonArray = "[" & bootnodes.mapIt("\"" & it & "\"").join(",") & "]"
|
||||
self.wakuDiscv5.updateBootstrapRecords(jsonArray).isOkOr:
|
||||
return err("error in discv5UpdateBootnodes: " & $error)
|
||||
return ok(true)
|
||||
except CatchableError as e:
|
||||
return err(e.msg)
|
||||
|
||||
method startDiscv5(self: Waku): Future[Result[bool, string]] {.async.} =
|
||||
try:
|
||||
if self.wakuDiscv5.isNil():
|
||||
return err("discv5 not started")
|
||||
(await self.wakuDiscv5.start()).isOkOr:
|
||||
return err("error starting discv5: " & $error)
|
||||
return ok(true)
|
||||
except CatchableError as e:
|
||||
return err(e.msg)
|
||||
|
||||
method stopDiscv5(self: Waku): Future[Result[bool, string]] {.async.} =
|
||||
try:
|
||||
if self.wakuDiscv5.isNil():
|
||||
return err("discv5 not started")
|
||||
await self.wakuDiscv5.stop()
|
||||
return ok(true)
|
||||
except CatchableError as e:
|
||||
return err(e.msg)
|
||||
|
||||
method peerExchangeRequest(
|
||||
self: Waku, numPeers: uint64
|
||||
): Future[Result[int, string]] {.async.} =
|
||||
try:
|
||||
let numPeersRecv = (await self.node.fetchPeerExchangePeers(numPeers)).valueOr:
|
||||
return err("failed peer exchange: " & $error)
|
||||
return ok(numPeersRecv)
|
||||
except CatchableError as e:
|
||||
return err(e.msg)
|
||||
|
||||
# --- debug / info ---
|
||||
method version(self: Waku): Future[Result[string, string]] {.async.} =
|
||||
return ok(WakuNodeVersionString)
|
||||
|
||||
method listenAddresses(self: Waku): Future[Result[seq[string], string]] {.async.} =
|
||||
try:
|
||||
return ok(self.node.info().listenAddresses)
|
||||
except CatchableError as e:
|
||||
return err(e.msg)
|
||||
|
||||
method myEnr(self: Waku): Future[Result[string, string]] {.async.} =
|
||||
try:
|
||||
return ok(self.node.enr.toURI())
|
||||
except CatchableError as e:
|
||||
return err(e.msg)
|
||||
|
||||
method myPeerId(self: Waku): Future[Result[string, string]] {.async.} =
|
||||
try:
|
||||
return ok($self.node.peerId())
|
||||
except CatchableError as e:
|
||||
return err(e.msg)
|
||||
|
||||
method metrics(self: Waku): Future[Result[string, string]] {.async.} =
|
||||
{.gcsafe.}:
|
||||
try:
|
||||
return ok(defaultRegistry.toText())
|
||||
except CatchableError as e:
|
||||
return err(e.msg)
|
||||
|
||||
method isOnline(self: Waku): Future[Result[bool, string]] {.async.} =
|
||||
return ok(self.healthMonitor.onlineMonitor.amIOnline())
|
||||
|
||||
method pingPeer(
|
||||
self: Waku, peerAddr: string, timeoutMs: int
|
||||
): Future[Result[int64, string]] {.async.} =
|
||||
try:
|
||||
let peerInfo = parsePeerInfo(peerAddr).valueOr:
|
||||
return err("pingPeer failed to parse peer addr: " & $error)
|
||||
|
||||
let conn = await self.node.switch.dial(peerInfo.peerId, peerInfo.addrs, PingCodec)
|
||||
defer:
|
||||
await conn.close()
|
||||
let pingRTT = await self.node.libp2pPing.ping(conn)
|
||||
|
||||
if pingRTT == 0.nanos:
|
||||
return err("could not ping peer: rtt-0")
|
||||
|
||||
return ok(pingRTT.nanos)
|
||||
except CatchableError as e:
|
||||
return err(e.msg)
|
||||
|
||||
@ -162,13 +162,13 @@ proc registerProviders(backend: KvBackend, ctx: BrokerContext): Result[void, str
|
||||
|
||||
return ok()
|
||||
|
||||
proc clearProviders(ctx: BrokerContext) =
|
||||
proc clearProviders(ctx: BrokerContext) {.async.} =
|
||||
KvGet.clearProvider(ctx)
|
||||
KvExists.clearProvider(ctx)
|
||||
KvScan.clearProvider(ctx)
|
||||
KvCount.clearProvider(ctx)
|
||||
KvDelete.clearProvider(ctx)
|
||||
PersistEvent.dropAllListeners(ctx)
|
||||
await PersistEvent.dropAllListeners(ctx)
|
||||
|
||||
# ── thread proc ─────────────────────────────────────────────────────────
|
||||
|
||||
@ -217,7 +217,7 @@ proc storageThreadMain(arg: ptr StorageThreadArg) {.thread.} =
|
||||
except CatchableError as e:
|
||||
error "storage thread loop crashed", err = e.msg
|
||||
|
||||
clearProviders(arg.ctx)
|
||||
waitFor clearProviders(arg.ctx)
|
||||
backend.close()
|
||||
|
||||
# ── lifecycle ───────────────────────────────────────────────────────────
|
||||
|
||||
@ -272,7 +272,7 @@ proc hasJob*(p: Persistency, jobId: string): bool {.inline.} =
|
||||
proc persist*(t: Job, ops: seq[TxOp]): Future[void] {.async.} =
|
||||
## Emit a batched persist event. The handler treats >1 ops as a single
|
||||
## BEGIN IMMEDIATE/COMMIT transaction (see backend_sqlite.applyOps).
|
||||
await PersistEvent.emit(t.context, PersistEvent(ops: ops))
|
||||
PersistEvent.emit(t.context, PersistEvent(ops: ops))
|
||||
|
||||
proc persist*(t: Job, op: TxOp): Future[void] {.async.} =
|
||||
await persist(t, @[op])
|
||||
|
||||
@ -328,8 +328,8 @@
|
||||
}
|
||||
},
|
||||
"brokers": {
|
||||
"version": "#cf5ee65cc20211068d7191de7e5e177c0dc212fa",
|
||||
"vcsRevision": "cf5ee65cc20211068d7191de7e5e177c0dc212fa",
|
||||
"version": "#3eab670389aac073cd9d820a592d54898f2973dc",
|
||||
"vcsRevision": "3eab670389aac073cd9d820a592d54898f2973dc",
|
||||
"url": "https://github.com/NagyZoltanPeter/nim-brokers.git",
|
||||
"downloadMethod": "git",
|
||||
"dependencies": [
|
||||
@ -341,7 +341,7 @@
|
||||
"cbor_serialization"
|
||||
],
|
||||
"checksums": {
|
||||
"sha1": "9bb46550b5bb9bde6722697d678ea236e2143350"
|
||||
"sha1": "36903ba45a27621c204ab55dfb4609f0a63b05de"
|
||||
}
|
||||
},
|
||||
"stint": {
|
||||
|
||||
@ -31,28 +31,28 @@ proc waitForConnectionStatus(
|
||||
) {.async.} =
|
||||
var future = newFuture[void]("waitForConnectionStatus")
|
||||
|
||||
let handler: EventConnectionStatusChangeListenerProc = proc(
|
||||
e: ConnectionStatusChangeEvent
|
||||
let handler: health_events.ConnectionStatusChangeEventListenerProc = proc(
|
||||
e: health_events.ConnectionStatusChangeEvent
|
||||
) {.async: (raises: []), gcsafe.} =
|
||||
if not future.finished:
|
||||
if e.connectionStatus == expected:
|
||||
future.complete()
|
||||
|
||||
let handle = ConnectionStatusChangeEvent.listen(brokerCtx, handler).valueOr:
|
||||
let handle = health_events.ConnectionStatusChangeEvent.listen(brokerCtx, handler).valueOr:
|
||||
raiseAssert error
|
||||
|
||||
try:
|
||||
if not await future.withTimeout(TestTimeout):
|
||||
raiseAssert "Timeout waiting for status: " & $expected
|
||||
finally:
|
||||
await ConnectionStatusChangeEvent.dropListener(brokerCtx, handle)
|
||||
await health_events.ConnectionStatusChangeEvent.dropListener(brokerCtx, handle)
|
||||
|
||||
proc waitForShardHealthy(
|
||||
brokerCtx: BrokerContext
|
||||
): Future[ShardTopicHealthChangeEvent] {.async.} =
|
||||
var future = newFuture[ShardTopicHealthChangeEvent]("waitForShardHealthy")
|
||||
|
||||
let handler: EventShardTopicHealthChangeListenerProc = proc(
|
||||
let handler: ShardTopicHealthChangeEventListenerProc = proc(
|
||||
e: ShardTopicHealthChangeEvent
|
||||
) {.async: (raises: []), gcsafe.} =
|
||||
if not future.finished:
|
||||
|
||||
@ -85,7 +85,7 @@ proc waitForConnectionStatus(
|
||||
proc createApiNodeConf(numShards: uint16 = 1): WakuNodeConf =
|
||||
var conf = defaultWakuNodeConf().valueOr:
|
||||
raiseAssert error
|
||||
conf.mode = cli_args.WakuMode.Core
|
||||
conf.mode = some(cli_args.WakuMode.Core)
|
||||
conf.listenAddress = parseIpAddress("0.0.0.0")
|
||||
conf.tcpPort = Port(0)
|
||||
conf.discv5UdpPort = Port(0)
|
||||
|
||||
@ -1,6 +1,6 @@
|
||||
{.used.}
|
||||
|
||||
import std/strutils
|
||||
import std/[strutils, options]
|
||||
import chronos, testutils/unittests, stew/byteutils, libp2p/[switch, peerinfo]
|
||||
import brokers/broker_context
|
||||
import ../testlib/[common, wakucore, wakunode, testasync]
|
||||
@ -122,7 +122,7 @@ proc validate(
|
||||
proc createApiNodeConf(mode: cli_args.WakuMode = cli_args.WakuMode.Core): WakuNodeConf =
|
||||
var conf = defaultWakuNodeConf().valueOr:
|
||||
raiseAssert error
|
||||
conf.mode = mode
|
||||
conf.mode = some(mode)
|
||||
conf.listenAddress = parseIpAddress("0.0.0.0")
|
||||
conf.tcpPort = Port(0)
|
||||
conf.discv5UdpPort = Port(0)
|
||||
|
||||
@ -73,7 +73,7 @@ proc createApiNodeConf(
|
||||
): WakuNodeConf =
|
||||
var conf = defaultWakuNodeConf().valueOr:
|
||||
raiseAssert error
|
||||
conf.mode = mode
|
||||
conf.mode = some(mode)
|
||||
conf.listenAddress = parseIpAddress("0.0.0.0")
|
||||
conf.tcpPort = Port(0)
|
||||
conf.discv5UdpPort = Port(0)
|
||||
|
||||
@ -23,12 +23,16 @@ import logos_delivery/waku/persistency/sds_persistency
|
||||
import sds
|
||||
import snapshot_codec
|
||||
|
||||
import logos_delivery/api/messaging_client_interface
|
||||
# for the `Send` request broker — send-side tests mock its provider to
|
||||
# intercept `messagingClient.send` without touching the network.
|
||||
|
||||
const TestTimeout = chronos.seconds(15)
|
||||
|
||||
proc createApiNodeConf(): WakuNodeConf =
|
||||
var conf = defaultWakuNodeConf().valueOr:
|
||||
raiseAssert error
|
||||
conf.mode = cli_args.WakuMode.Core
|
||||
conf.mode = some(cli_args.WakuMode.Core)
|
||||
conf.listenAddress = parseIpAddress("0.0.0.0")
|
||||
conf.tcpPort = Port(0)
|
||||
conf.discv5UdpPort = Port(0)
|
||||
@ -68,9 +72,11 @@ suite "Reliable Channel - ingress":
|
||||
## plaintext anyway, but installing them is the documented setup.
|
||||
setNoopEncryption()
|
||||
|
||||
discard manager
|
||||
.createReliableChannel(channelId, contentTopic, SdsParticipantID("local"))
|
||||
.expect("createReliableChannel")
|
||||
discard (
|
||||
await manager.createReliableChannel(
|
||||
channelId, contentTopic, SdsParticipantID("local")
|
||||
)
|
||||
).expect("createReliableChannel")
|
||||
|
||||
let received = newFuture[seq[byte]]("channel-message-received")
|
||||
discard ChannelMessageReceivedEvent
|
||||
@ -133,9 +139,11 @@ suite "Reliable Channel - ingress":
|
||||
|
||||
setNoopEncryption()
|
||||
|
||||
discard manager
|
||||
.createReliableChannel(channelId, contentTopic, SdsParticipantID("local"))
|
||||
.expect("createReliableChannel")
|
||||
discard (
|
||||
await manager.createReliableChannel(
|
||||
channelId, contentTopic, SdsParticipantID("local")
|
||||
)
|
||||
).expect("createReliableChannel")
|
||||
|
||||
var fired = false
|
||||
discard ChannelMessageReceivedEvent
|
||||
@ -193,17 +201,22 @@ suite "Reliable Channel - send state machine":
|
||||
setNoopEncryption()
|
||||
|
||||
var sendCalls = 0
|
||||
let fakeSend: SendHandler = proc(
|
||||
env: MessageEnvelope
|
||||
): Future[Result[RequestId, string]] {.async: (raises: [CatchableError]), gcsafe.} =
|
||||
sendCalls.inc
|
||||
return ok(fakeMsgReqId)
|
||||
## Mock the messaging egress: override the `Send` provider the mounted
|
||||
## MessagingClient installed, so the channel's `messagingClient.send`
|
||||
## returns a canned RequestId without hitting the network.
|
||||
Send.clearProvider(brokerCtx)
|
||||
Send.setProvider(
|
||||
brokerCtx,
|
||||
proc(env: MessageEnvelope): Future[Result[RequestId, string]] {.async.} =
|
||||
sendCalls.inc
|
||||
return ok(fakeMsgReqId),
|
||||
).expect("set Send provider")
|
||||
|
||||
discard manager
|
||||
.createReliableChannel(
|
||||
channelId, contentTopic, SdsParticipantID("local"), sendHandler = fakeSend
|
||||
discard (
|
||||
await manager.createReliableChannel(
|
||||
channelId, contentTopic, SdsParticipantID("local")
|
||||
)
|
||||
.expect("createReliableChannel")
|
||||
).expect("createReliableChannel")
|
||||
|
||||
let sentFut = newFuture[RequestId]("channel-sent")
|
||||
discard ChannelMessageSentEvent
|
||||
@ -216,7 +229,8 @@ suite "Reliable Channel - send state machine":
|
||||
)
|
||||
.expect("listen ChannelMessageSentEvent")
|
||||
|
||||
let channelReqId = (await manager.send(channelId, "hello".toBytes())).expect("send")
|
||||
let channelReqId =
|
||||
(await manager.sendOnChannel(channelId, "hello".toBytes(), false)).expect("send")
|
||||
|
||||
let dispatchDeadline = Moment.now() + 1.seconds
|
||||
while Moment.now() < dispatchDeadline and sendCalls == 0:
|
||||
@ -260,18 +274,21 @@ suite "Reliable Channel - send state machine":
|
||||
setNoopEncryption()
|
||||
|
||||
var msgReqIds: seq[RequestId]
|
||||
let fakeSend: SendHandler = proc(
|
||||
env: MessageEnvelope
|
||||
): Future[Result[RequestId, string]] {.async: (raises: [CatchableError]), gcsafe.} =
|
||||
let id = RequestId("fake-msg-req-" & $(msgReqIds.len + 1))
|
||||
msgReqIds.add(id)
|
||||
return ok(id)
|
||||
## Mock the messaging egress (see the single-send case above).
|
||||
Send.clearProvider(brokerCtx)
|
||||
Send.setProvider(
|
||||
brokerCtx,
|
||||
proc(env: MessageEnvelope): Future[Result[RequestId, string]] {.async.} =
|
||||
let id = RequestId("fake-msg-req-" & $(msgReqIds.len + 1))
|
||||
msgReqIds.add(id)
|
||||
return ok(id),
|
||||
).expect("set Send provider")
|
||||
|
||||
discard manager
|
||||
.createReliableChannel(
|
||||
channelId, contentTopic, SdsParticipantID("local"), sendHandler = fakeSend
|
||||
discard (
|
||||
await manager.createReliableChannel(
|
||||
channelId, contentTopic, SdsParticipantID("local")
|
||||
)
|
||||
.expect("createReliableChannel")
|
||||
).expect("createReliableChannel")
|
||||
|
||||
let sentFut = newFuture[RequestId]("channel-sent")
|
||||
let erroredFut = newFuture[RequestId]("channel-errored")
|
||||
@ -295,9 +312,9 @@ suite "Reliable Channel - send state machine":
|
||||
.expect("listen ChannelMessageErrorEvent")
|
||||
|
||||
let channelReqId1 =
|
||||
(await manager.send(channelId, "first".toBytes())).expect("send 1")
|
||||
(await manager.sendOnChannel(channelId, "first".toBytes(), false)).expect("send 1")
|
||||
let channelReqId2 =
|
||||
(await manager.send(channelId, "second".toBytes())).expect("send 2")
|
||||
(await manager.sendOnChannel(channelId, "second".toBytes(), false)).expect("send 2")
|
||||
|
||||
let dispatchDeadline = Moment.now() + 1.seconds
|
||||
while Moment.now() < dispatchDeadline and msgReqIds.len < 2:
|
||||
@ -364,29 +381,33 @@ suite "Reliable Channel - send state machine":
|
||||
|
||||
var msgReqIds: seq[RequestId]
|
||||
var sendsReturned = 0
|
||||
let fakeSend: SendHandler = proc(
|
||||
env: MessageEnvelope
|
||||
): Future[Result[RequestId, string]] {.async: (raises: [CatchableError]), gcsafe.} =
|
||||
## Call 2 fires the first segment's terminal event and then
|
||||
## yields, so the listener task runs while the second segment
|
||||
## is still mid-`await` in `onReadyToSend` — the exact race
|
||||
## window the regression test targets.
|
||||
let id = RequestId("race-msg-req-" & $(msgReqIds.len + 1))
|
||||
msgReqIds.add(id)
|
||||
if msgReqIds.len == 2:
|
||||
waku_message_events.MessageSentEvent.emit(
|
||||
brokerCtx,
|
||||
waku_message_events.MessageSentEvent(requestId: msgReqIds[0], messageHash: ""),
|
||||
)
|
||||
await sleepAsync(50.milliseconds)
|
||||
sendsReturned.inc()
|
||||
return ok(id)
|
||||
## Mock the messaging egress. Call 2 fires the first segment's terminal
|
||||
## event and then yields, so the listener task runs while the second
|
||||
## segment is still mid-`await` in `onReadyToSend` — the exact race
|
||||
## window the regression test targets.
|
||||
Send.clearProvider(brokerCtx)
|
||||
Send.setProvider(
|
||||
brokerCtx,
|
||||
proc(env: MessageEnvelope): Future[Result[RequestId, string]] {.async.} =
|
||||
let id = RequestId("race-msg-req-" & $(msgReqIds.len + 1))
|
||||
msgReqIds.add(id)
|
||||
if msgReqIds.len == 2:
|
||||
waku_message_events.MessageSentEvent.emit(
|
||||
brokerCtx,
|
||||
waku_message_events.MessageSentEvent(
|
||||
requestId: msgReqIds[0], messageHash: ""
|
||||
),
|
||||
)
|
||||
await sleepAsync(50.milliseconds)
|
||||
sendsReturned.inc()
|
||||
return ok(id),
|
||||
).expect("set Send provider")
|
||||
|
||||
discard manager
|
||||
.createReliableChannel(
|
||||
channelId, contentTopic, SdsParticipantID("local"), sendHandler = fakeSend
|
||||
discard (
|
||||
await manager.createReliableChannel(
|
||||
channelId, contentTopic, SdsParticipantID("local")
|
||||
)
|
||||
.expect("createReliableChannel")
|
||||
).expect("createReliableChannel")
|
||||
|
||||
var finalisedReqIds: seq[RequestId]
|
||||
let bothFinalised = newFuture[void]("both-finalised")
|
||||
@ -403,7 +424,7 @@ suite "Reliable Channel - send state machine":
|
||||
.expect("listen ChannelMessageSentEvent")
|
||||
|
||||
let channelReqId1 =
|
||||
(await manager.send(channelId, "first".toBytes())).expect("send 1")
|
||||
(await manager.sendOnChannel(channelId, "first".toBytes(), false)).expect("send 1")
|
||||
|
||||
## Drain the first segment fully before queueing the second, so
|
||||
## the rate-limit FIFO between sibling sends isn't itself under
|
||||
@ -414,9 +435,9 @@ suite "Reliable Channel - send state machine":
|
||||
check msgReqIds.len == 1
|
||||
|
||||
let channelReqId2 =
|
||||
(await manager.send(channelId, "second".toBytes())).expect("send 2")
|
||||
(await manager.sendOnChannel(channelId, "second".toBytes(), false)).expect("send 2")
|
||||
|
||||
## Wait until `fakeSend(m2)` has fully returned and yield once
|
||||
## Wait until `sendOnChannel(m2)` has fully returned and yield once
|
||||
## more so `onReadyToSend`'s post-await continuation gets a chance
|
||||
## to register `id2` in `inflightMessagingIds` before we emit its
|
||||
## terminal event.
|
||||
@ -480,7 +501,7 @@ suite "Reliable Channel - SDS persistence":
|
||||
)
|
||||
.expect("createReliableChannel")
|
||||
|
||||
discard (await manager.send(channelId, "persist me".toBytes())).expect("send")
|
||||
discard (await manager.sendOnChannel(channelId, "persist me".toBytes(), false)).expect("send")
|
||||
|
||||
## Same handle the channel layer writes through (`openJob` is idempotent).
|
||||
let job = persistency.openJob("sds").expect("openJob sds")
|
||||
@ -850,7 +871,7 @@ suite "Reliable Channel - SDS protocol semantics":
|
||||
)
|
||||
await sleepAsync(100.milliseconds)
|
||||
|
||||
discard (await manager.send(channelId, "reply".toBytes())).expect("send")
|
||||
discard (await manager.sendOnChannel(channelId, "reply".toBytes(), false)).expect("send")
|
||||
var deadline = Moment.now() + 2.seconds
|
||||
while Moment.now() < deadline and capturedWires.len < 1:
|
||||
await sleepAsync(5.milliseconds)
|
||||
@ -902,7 +923,7 @@ suite "Reliable Channel - SDS protocol semantics":
|
||||
)
|
||||
.expect("createReliableChannel")
|
||||
|
||||
discard (await manager.send(channelId, "needs ack".toBytes())).expect("send")
|
||||
discard (await manager.sendOnChannel(channelId, "needs ack".toBytes(), false)).expect("send")
|
||||
var deadline = Moment.now() + 2.seconds
|
||||
while Moment.now() < deadline and capturedWires.len < 1:
|
||||
await sleepAsync(5.milliseconds)
|
||||
@ -1156,8 +1177,8 @@ suite "Reliable Channel - SDS protocol semantics":
|
||||
.expect("listen ChannelMessageReceivedEvent")
|
||||
|
||||
## Send side: the same payload twice must produce two distinct ids.
|
||||
discard (await manager.send(channelId, appPayload)).expect("send 1")
|
||||
discard (await manager.send(channelId, appPayload)).expect("send 2")
|
||||
discard (await manager.sendOnChannel(channelId, appPayload, false)).expect("send 1")
|
||||
discard (await manager.sendOnChannel(channelId, appPayload, false)).expect("send 2")
|
||||
var deadline = Moment.now() + 2.seconds
|
||||
while Moment.now() < deadline and capturedWires.len < 2:
|
||||
await sleepAsync(5.milliseconds)
|
||||
@ -1197,7 +1218,7 @@ suite "Reliable Channel - SDS protocol semantics":
|
||||
waku.mountReliableChannelManager().expect("mountReliableChannelManager")
|
||||
manager = waku.reliableChannelManager
|
||||
|
||||
check (await manager.send(ChannelId("no-such-channel"), "x".toBytes())).isErr()
|
||||
check (await manager.sendOnChannel(ChannelId("no-such-channel"), "x".toBytes(), false)).isErr()
|
||||
check (await manager.closeChannel(ChannelId("no-such-channel"))).isErr()
|
||||
|
||||
(await waku.stop()).expect("stop")
|
||||
|
||||
@ -156,7 +156,7 @@ suite "Persistency lifecycle":
|
||||
let ev = PersistEvent(
|
||||
ops: @[TxOp(category: "msg", key: k, kind: txPut, payload: payloadBytes("hello"))]
|
||||
)
|
||||
await PersistEvent.emit(t.context, ev)
|
||||
PersistEvent.emit(t.context, ev)
|
||||
let ckOk2 = await t.pollExists("msg", k)
|
||||
check ckOk2
|
||||
|
||||
@ -178,7 +178,7 @@ suite "Persistency lifecycle":
|
||||
check a.context != b.context
|
||||
|
||||
let k = key("shared", 1'i64)
|
||||
await PersistEvent.emit(
|
||||
PersistEvent.emit(
|
||||
a.context,
|
||||
PersistEvent(
|
||||
ops: @[
|
||||
@ -188,7 +188,7 @@ suite "Persistency lifecycle":
|
||||
]
|
||||
),
|
||||
)
|
||||
await PersistEvent.emit(
|
||||
PersistEvent.emit(
|
||||
b.context,
|
||||
PersistEvent(
|
||||
ops: @[
|
||||
@ -255,7 +255,7 @@ suite "Persistency lifecycle":
|
||||
ops.add(
|
||||
TxOp(category: "msg", key: key("c", i), kind: txPut, payload: payloadBytes($i))
|
||||
)
|
||||
await PersistEvent.emit(t.context, PersistEvent(ops: ops))
|
||||
PersistEvent.emit(t.context, PersistEvent(ops: ops))
|
||||
# Wait for the last insert to land.
|
||||
let ckOk5 = await t.pollExists("msg", key("c", 5'i64))
|
||||
check ckOk5
|
||||
@ -285,7 +285,7 @@ suite "Persistency lifecycle":
|
||||
let r1 = aw7.get()
|
||||
check r1.existed == false
|
||||
|
||||
await PersistEvent.emit(
|
||||
PersistEvent.emit(
|
||||
t.context,
|
||||
PersistEvent(
|
||||
ops: @[TxOp(category: "msg", key: k, kind: txPut, payload: payloadBytes("v"))]
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user