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

chore(deps)!: bump nim-ffi to 0.3.0-rc.1 (per-listener event ABI) (#4070)

Browse Source
This commit is contained in:
Gabriel Cruz 2026-07-31 13:50:59 -03:00 committed by GitHub
parent 230417e7b6
commit 4809b991df
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
31 changed files with 347 additions and 134 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
examples/cbindings/waku_example.c
View File

@ -312,7 +312,15 @@ int main(int argc, char **argv)
printf("Bind addr: %s:%u\n", cfgNode.host, cfgNode.port);
printf("Waku Relay enabled: %s\n", cfgNode.relay == 1 ? "YES" : "NO");
logosdelivery_set_event_callback(ctx, on_event_received, userData);
static const char *kEventNames[] = {
"onMessageSent", "onMessageError",
"onMessagePropagated", "onMessageReceived",
"onConnectionStatusChange", "onTopicHealthChange",
"onConnectionChange", "onReceivedMessage",
"onChannelMessageReceived", "onChannelMessageSent",
"onChannelMessageError"};
for (size_t i = 0; i < sizeof(kEventNames) / sizeof(kEventNames[0]); i++)
logosdelivery_add_event_listener(ctx, kEventNames[i], on_event_received, userData);
logosdelivery_start_node(ctx, event_handler, userData);
waitForCallback();

12
examples/cpp/waku.cpp
View File

@ -307,10 +307,14 @@ int main(int argc, char **argv)
std::cout << "Custom pubsub topic: " << pubsubTopic << std::endl;
logosdelivery_set_event_callback(ctx,
cify([&](const char *msg, size_t len)
{ event_handler(msg, len); }),
nullptr);
auto onEvent = cify([&](const char *msg, size_t len)
{ event_handler(msg, len); });
for (const char *eventName :
{"onMessageSent", "onMessageError", "onMessagePropagated",
"onMessageReceived", "onConnectionStatusChange", "onTopicHealthChange",
"onConnectionChange", "onReceivedMessage", "onChannelMessageReceived",
"onChannelMessageSent", "onChannelMessageError"})
logosdelivery_add_event_listener(ctx, eventName, onEvent, nullptr);
WAKU_CALL(logosdelivery_start_node(ctx,
cify([&](const char *msg, size_t len)

9
examples/golang/waku.go
View File

@ -112,7 +112,14 @@ package main
// This technique is needed because cgo only allows to export Go functions and not methods.
logosdelivery_set_event_callback(wakuCtx, (FFICallBack) globalEventCallback, wakuCtx);
static const char* eventNames[] = {
"onMessageSent", "onMessageError", "onMessagePropagated",
"onMessageReceived", "onConnectionStatusChange", "onTopicHealthChange",
"onConnectionChange", "onReceivedMessage", "onChannelMessageReceived",
"onChannelMessageSent", "onChannelMessageError"};
for (size_t i = 0; i < sizeof(eventNames) / sizeof(eventNames[0]); i++) {
logosdelivery_add_event_listener(wakuCtx, eventNames[i], (FFICallBack) globalEventCallback, wakuCtx);
}
}
static void cGoWakuContentTopic(void* wakuCtx,

12
examples/ios/WakuExample/WakuNode.swift
View File

@ -344,8 +344,16 @@ actor WakuActor {
ctx = createResult.ctx
// Set event callback
logosdelivery_set_event_callback(ctx, WakuActor.eventCallback, nil)
// Register per-event listeners
let eventNames = [
"onMessageSent", "onMessageError", "onMessagePropagated",
"onMessageReceived", "onConnectionStatusChange", "onTopicHealthChange",
"onConnectionChange", "onReceivedMessage", "onChannelMessageReceived",
"onChannelMessageSent", "onChannelMessageError",
]
for name in eventNames {
_ = logosdelivery_add_event_listener(ctx, name, WakuActor.eventCallback, nil)
}
// Start node
let startResult = await callWakuSync { userData in

11
examples/mobile/android/app/src/main/jni/waku_ffi.c
View File

@ -321,5 +321,14 @@ void Java_com_mobile_WakuModule_wakuSetEventCallback(JNIEnv *env, jobject thiz,
cb_env *c = (cb_env *)malloc(sizeof(cb_env));
c->wakuPtr = wakuPtr;
c->env = env;
logosdelivery_set_event_callback((void *)wakuPtr, wk_callback, (void *)c);
static const char *kEventNames[] = {
"onMessageSent", "onMessageError",
"onMessagePropagated", "onMessageReceived",
"onConnectionStatusChange", "onTopicHealthChange",
"onConnectionChange", "onReceivedMessage",
"onChannelMessageReceived", "onChannelMessageSent",
"onChannelMessageError"};
for (size_t i = 0; i < sizeof(kEventNames) / sizeof(kEventNames[0]); i++)
logosdelivery_add_event_listener((void *)wakuPtr, kEventNames[i], wk_callback,
(void *)c);
}

10
examples/nodejs/waku_addon.c
View File

@ -284,7 +284,15 @@ static napi_value WakuSetEventCallback(napi_env env, napi_callback_info info) {
// Inside 'event_handler', the event will be dispatched to the NodeJs
// if there is a proper napi_function (ref_event_callback) being set.
logosdelivery_set_event_callback(event_handler, userData);
static const char *kEventNames[] = {
"onMessageSent", "onMessageError",
"onMessagePropagated", "onMessageReceived",
"onConnectionStatusChange", "onTopicHealthChange",
"onConnectionChange", "onReceivedMessage",
"onChannelMessageReceived", "onChannelMessageSent",
"onChannelMessageError"};
for (size_t i = 0; i < sizeof(kEventNames) / sizeof(kEventNames[0]); i++)
logosdelivery_add_event_listener(ctx, kEventNames[i], event_handler, userData);
return NULL;
}

13
examples/python/waku.py
View File

@ -112,8 +112,17 @@ print("Waku Relay enabled: {}".format(args.relay))
# Set the event callback
callback = callback_type(handle_event) # This line is important so that the callback is not gc'ed
libwaku.logosdelivery_set_event_callback.argtypes = [callback_type, ctypes.c_void_p]
libwaku.logosdelivery_set_event_callback(callback, ctypes.c_void_p(0))
libwaku.logosdelivery_add_event_listener.argtypes = [ctypes.c_void_p,
ctypes.c_char_p,
callback_type,
ctypes.c_void_p]
libwaku.logosdelivery_add_event_listener.restype = ctypes.c_uint64
for event_name in [b"onMessageSent", b"onMessageError", b"onMessagePropagated",
b"onMessageReceived", b"onConnectionStatusChange",
b"onTopicHealthChange", b"onConnectionChange", b"onReceivedMessage",
b"onChannelMessageReceived", b"onChannelMessageSent",
b"onChannelMessageError"]:
libwaku.logosdelivery_add_event_listener(ctx, event_name, callback, ctypes.c_void_p(0))
# Start the node
libwaku.logosdelivery_start_node.argtypes = [ctypes.c_void_p,

8
examples/qt/waku_handler.h
View File

@ -27,7 +27,13 @@ public:
void initialize(const QString& jsonConfig, WakuCallBack event_handler, void* userData) {
ctx = logosdelivery_create_node(jsonConfig.toUtf8().constData(), WakuCallBack(event_handler), userData);
logosdelivery_set_event_callback(ctx, on_event_received, userData);
for (const char *eventName :
{"onMessageSent", "onMessageError", "onMessagePropagated",
"onMessageReceived", "onConnectionStatusChange", "onTopicHealthChange",
"onConnectionChange", "onReceivedMessage", "onChannelMessageReceived",
"onChannelMessageSent", "onChannelMessageError"}) {
logosdelivery_add_event_listener(ctx, eventName, on_event_received, userData);
}
qDebug() << "Waku context initialized, ready to start.";
}

11
library/MESSAGE_EVENTS.md
View File

@ -2,7 +2,7 @@
## Overview
The liblogosdelivery library emits three types of message delivery events that clients can listen to by registering an event callback using `logosdelivery_set_event_callback()`.
The liblogosdelivery library emits three types of message delivery events that clients can listen to by registering a per-event callback with `logosdelivery_add_event_listener()`. The events are delivered under the wire names `onMessageSent`, `onMessagePropagated` and `onMessageError` (the JSON `eventType` inside each payload is `message_sent` / `message_propagated` / `message_error`).
## Event Types
@ -85,9 +85,14 @@ void event_callback(int ret, const char *msg, size_t len, void *userData) {
### 2. Register the Callback
Register the callback once per event name you want to receive. Each call returns a
listener id you can later pass to `logosdelivery_remove_event_listener(ctx, id)`.
```c
void *ctx = logosdelivery_create_node(config, callback, userData);
logosdelivery_set_event_callback(ctx, event_callback, NULL);
logosdelivery_add_event_listener(ctx, "onMessageSent", event_callback, NULL);
logosdelivery_add_event_listener(ctx, "onMessagePropagated", event_callback, NULL);
logosdelivery_add_event_listener(ctx, "onMessageError", event_callback, NULL);
```
### 3. Start the Node
@ -114,7 +119,7 @@ For a failed message send:
## Important Notes
1. **Thread Safety**: The event callback is invoked from the FFI worker thread. Ensure your callback is thread-safe if it accesses shared state.
1. **Thread Safety**: The event callback is invoked from a dedicated event thread (separate from the FFI worker thread). Ensure your callback is thread-safe if it accesses shared state.
2. **Non-Blocking**: Keep the callback fast and non-blocking. Do not perform long-running operations in the callback.

31
library/README.md
View File

@ -154,18 +154,41 @@ Note: The `payload` field should be base64-encoded.
### Events
#### `logosdelivery_set_event_callback`
Sets a callback that will be invoked whenever an event occurs (e.g., message received).
Events are delivered through a per-event listener registry: register one callback
per event name you care about. A registration returns a listener id you can later
pass to remove it.
#### `logosdelivery_add_event_listener`
Registers `callback` for the named event and returns a non-zero listener id (0 on
an invalid context).
```c
void logosdelivery_set_event_callback(
uint64_t logosdelivery_add_event_listener(
void *ctx,
const char *eventName,
FFICallBack callback,
void *userData
);
```
**Important:** The callback should be fast, non-blocking, and thread-safe.
Event names: `onMessageSent`, `onMessageError`, `onMessagePropagated`,
`onMessageReceived`, `onConnectionStatusChange`, `onTopicHealthChange`,
`onConnectionChange`, `onReceivedMessage`, `onChannelMessageReceived`,
`onChannelMessageSent`, `onChannelMessageError`.
#### `logosdelivery_remove_event_listener`
Removes a previously registered listener. Returns `0` on success, `1` if the
listener id was not found or the context is invalid.
```c
int logosdelivery_remove_event_listener(
void *ctx,
uint64_t listenerId
);
```
**Important:** Callbacks run on a dedicated event thread and should be fast,
non-blocking, and thread-safe.
## Building

8
library/channels_api/channel_api.nim
View File

@ -14,7 +14,7 @@ proc logosdelivery_channel_create(
channelIdStr: cstring,
contentTopicStr: cstring,
senderIdStr: cstring,
) {.ffi.} =
) {.ffiRaw.} =
requireInitializedNode(ctx, "ChannelCreate"):
return err(errMsg)
@ -35,7 +35,7 @@ proc logosdelivery_channel_exists(
callback: FFICallBack,
userData: pointer,
channelIdStr: cstring,
) {.ffi.} =
) {.ffiRaw.} =
## Returns `"true"` or `"false"`; a missing channel is not an error.
requireInitializedNode(ctx, "ChannelExists"):
return err(errMsg)
@ -51,7 +51,7 @@ proc logosdelivery_channel_send(
userData: pointer,
channelIdStr: cstring,
messageJson: cstring,
) {.ffi.} =
) {.ffiRaw.} =
## `messageJson` carries `{ "payload": <base64>, "ephemeral": <bool> }`.
requireInitializedNode(ctx, "ChannelSend"):
return err(errMsg)
@ -87,7 +87,7 @@ proc logosdelivery_channel_close(
callback: FFICallBack,
userData: pointer,
channelIdStr: cstring,
) {.ffi.} =
) {.ffiRaw.} =
requireInitializedNode(ctx, "ChannelClose"):
return err(errMsg)

41
library/declare_lib.nim
View File

@ -1,12 +1,30 @@
import ffi
import std/locks
import results
import logos_delivery
declareLibrary("logosdelivery")
declareLibrary("logosdelivery", LogosDelivery)
var eventCallbackLock: Lock
initLock(eventCallbackLock)
template checkParams*(
ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer
) =
## Re-implements the `checkParams` helper dropped from nim-ffi in 0.3.0.
if not ctx.isNil():
ctx[].userData = userData
if callback.isNil():
return RET_MISSING_CALLBACK
template emitEvent*(eventName: string, body: untyped) =
## Enqueues `body`'s payload for nim-ffi's event thread to fan out to listeners.
## Callers are `{.async: (raises: []).}` broker listeners, and the payload
## builders infer an `Exception` effect, so nothing narrower than `Exception`
## compiles here. That also swallows `Defect`, which is why the handler only
## logs: a defect raised while rendering one event must not take the node down,
## and it cannot be re-raised without breaking the `raises: []` contract.
try:
dispatchFFIEvent(eventName):
body
except Exception as e:
chronicles.error "failed to emit FFI event", event = eventName, err = e.msg
template requireInitializedNode*(
ctx: ptr FFIContext[LogosDelivery], opName: string, onError: untyped
@ -35,18 +53,3 @@ template requireChannels*(
ctx.myLib[].ensureChannels().isOkOr:
let errMsg {.inject.} = opName & " failed: " & error
onError
proc logosdelivery_set_event_callback(
ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer
) {.dynlib, exportc, cdecl.} =
if isNil(ctx):
echo "error: invalid context in logosdelivery_set_event_callback"
return
# prevent race conditions that might happen due incorrect usage.
eventCallbackLock.acquire()
defer:
eventCallbackLock.release()
ctx[].eventCallback = cast[pointer](callback)
ctx[].eventUserData = userData

8
library/examples/logosdelivery_example.c
View File

@ -155,9 +155,11 @@ int main() {
return 1;
}
printf("\n2. Setting up event callback...\n");
logosdelivery_set_event_callback(ctx, event_callback, NULL);
printf("Event callback registered for message events\n");
printf("\n2. Setting up event listeners...\n");
logosdelivery_add_event_listener(ctx, "onMessageSent", event_callback, NULL);
logosdelivery_add_event_listener(ctx, "onMessagePropagated", event_callback, NULL);
logosdelivery_add_event_listener(ctx, "onMessageError", event_callback, NULL);
printf("Event listeners registered for message events\n");
printf("\n3. Starting node...\n");
logosdelivery_start_node(ctx, simple_callback, (void *)"start_node");

12
library/kernel_api/debug_node_api.nim
View File

@ -4,14 +4,14 @@ import logos_delivery, library/declare_lib
proc waku_version(
ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer
) {.ffi.} =
) {.ffiRaw.} =
let v = (await ctx.myLib[].waku.version()).valueOr:
return err(error)
return ok(v)
proc waku_listen_addresses(
ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer
) {.ffi.} =
) {.ffiRaw.} =
## returns a comma-separated string of the listen addresses
let addrs = (await ctx.myLib[].waku.listenAddresses()).valueOr:
return err(error)
@ -19,28 +19,28 @@ proc waku_listen_addresses(
proc waku_get_my_enr(
ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer
) {.ffi.} =
) {.ffiRaw.} =
let enrUri = (await ctx.myLib[].waku.myEnr()).valueOr:
return err(error)
return ok(enrUri)
proc waku_get_my_peerid(
ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer
) {.ffi.} =
) {.ffiRaw.} =
let peerId = (await ctx.myLib[].waku.myPeerId()).valueOr:
return err(error)
return ok(peerId)
proc waku_get_metrics(
ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer
) {.ffi.} =
) {.ffiRaw.} =
let m = (await ctx.myLib[].waku.metrics()).valueOr:
return err(error)
return ok(m)
proc waku_is_online(
ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer
) {.ffi.} =
) {.ffiRaw.} =
let online = (await ctx.myLib[].waku.isOnline()).valueOr:
return err(error)
return ok($online)

10
library/kernel_api/discovery_api.nim
View File

@ -7,7 +7,7 @@ proc waku_discv5_update_bootnodes(
callback: FFICallBack,
userData: pointer,
bootnodes: cstring,
) {.ffi.} =
) {.ffiRaw.} =
## Updates the bootnode list used for discovering new peers via DiscoveryV5
## bootnodes - JSON array containing the bootnode ENRs i.e. `["enr:...", "enr:..."]`
(await ctx.myLib[].waku.discv5UpdateBootnodes($bootnodes)).isOkOr:
@ -22,7 +22,7 @@ proc waku_dns_discovery(
enrTreeUrl: cstring,
nameDnsServer: cstring,
timeoutMs: cint,
) {.ffi.} =
) {.ffiRaw.} =
let nodes = (
await ctx.myLib[].waku.dnsDiscovery($enrTreeUrl, $nameDnsServer, int(timeoutMs))
).valueOr:
@ -33,7 +33,7 @@ proc waku_dns_discovery(
proc waku_start_discv5(
ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer
) {.ffi.} =
) {.ffiRaw.} =
(await ctx.myLib[].waku.startDiscv5()).isOkOr:
error "START_DISCV5 failed", error = error
return err(error)
@ -41,7 +41,7 @@ proc waku_start_discv5(
proc waku_stop_discv5(
ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer
) {.ffi.} =
) {.ffiRaw.} =
(await ctx.myLib[].waku.stopDiscv5()).isOkOr:
error "STOP_DISCV5 failed", error = error
return err(error)
@ -52,7 +52,7 @@ proc waku_peer_exchange_request(
callback: FFICallBack,
userData: pointer,
numPeers: uint64,
) {.ffi.} =
) {.ffiRaw.} =
let numValidPeers = (await ctx.myLib[].waku.peerExchangeRequest(numPeers)).valueOr:
error "waku_peer_exchange_request failed", error = error
return err(error)

18
library/kernel_api/peer_manager_api.nim
View File

@ -8,7 +8,7 @@ type PeerInfo = object
proc waku_get_peerids_from_peerstore(
ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer
) {.ffi.} =
) {.ffiRaw.} =
## returns a comma-separated string of peerIDs
let peerIds = (await ctx.myLib[].waku.peerIdsFromPeerstore()).valueOr:
return err(error)
@ -20,7 +20,7 @@ proc waku_connect(
userData: pointer,
peerMultiAddr: cstring,
timeoutMs: cuint,
) {.ffi.} =
) {.ffiRaw.} =
let peers = ($peerMultiAddr).split(",")
(await ctx.myLib[].waku.connect(peers, uint32(timeoutMs))).isOkOr:
return err(error)
@ -31,7 +31,7 @@ proc waku_disconnect_peer_by_id(
callback: FFICallBack,
userData: pointer,
peerId: cstring,
) {.ffi.} =
) {.ffiRaw.} =
(await ctx.myLib[].waku.disconnectPeerById($peerId)).isOkOr:
error "DISCONNECT_PEER_BY_ID failed", error = error
return err(error)
@ -39,7 +39,7 @@ proc waku_disconnect_peer_by_id(
proc waku_disconnect_all_peers(
ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer
) {.ffi.} =
) {.ffiRaw.} =
(await ctx.myLib[].waku.disconnectAllPeers()).isOkOr:
return err(error)
return ok("")
@ -51,7 +51,7 @@ proc waku_dial_peer(
peerMultiAddr: cstring,
protocol: cstring,
timeoutMs: cuint,
) {.ffi.} =
) {.ffiRaw.} =
(await ctx.myLib[].waku.dialPeer($peerMultiAddr, $protocol, int(timeoutMs))).isOkOr:
error "DIAL_PEER failed", error = error
return err(error)
@ -64,7 +64,7 @@ proc waku_dial_peer_by_id(
peerId: cstring,
protocol: cstring,
timeoutMs: cuint,
) {.ffi.} =
) {.ffiRaw.} =
(await ctx.myLib[].waku.dialPeerById($peerId, $protocol, int(timeoutMs))).isOkOr:
error "DIAL_PEER_BY_ID failed", error = error
return err(error)
@ -72,7 +72,7 @@ proc waku_dial_peer_by_id(
proc waku_get_connected_peers_info(
ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer
) {.ffi.} =
) {.ffiRaw.} =
## returns a JSON string mapping peerIDs to objects with protocols and addresses
let peers = (await ctx.myLib[].waku.connectedPeersInfo()).valueOr:
return err(error)
@ -86,7 +86,7 @@ proc waku_get_connected_peers_info(
proc waku_get_connected_peers(
ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer
) {.ffi.} =
) {.ffiRaw.} =
## returns a comma-separated string of peerIDs
let peerIds = (await ctx.myLib[].waku.connectedPeers()).valueOr:
return err(error)
@ -97,7 +97,7 @@ proc waku_get_peerids_by_protocol(
callback: FFICallBack,
userData: pointer,
protocol: cstring,
) {.ffi.} =
) {.ffiRaw.} =
## returns a comma-separated string of peerIDs that mount the given protocol
let peerIds = (await ctx.myLib[].waku.peerIdsByProtocol($protocol)).valueOr:
return err(error)

2
library/kernel_api/ping_api.nim
View File

@ -7,7 +7,7 @@ proc waku_ping_peer(
userData: pointer,
peerAddr: cstring,
timeoutMs: cuint,
) {.ffi.} =
) {.ffiRaw.} =
let rttNanos = (await ctx.myLib[].waku.pingPeer($peerAddr, int(timeoutMs))).valueOr:
return err(error)
return ok($rttNanos)

8
library/kernel_api/protocols/filter_api.nim
View File

@ -15,10 +15,10 @@ proc waku_filter_subscribe(
userData: pointer,
pubSubTopic: cstring,
contentTopics: cstring,
) {.ffi.} =
) {.ffiRaw.} =
proc onReceivedMessage(ctx: ptr FFIContext[LogosDelivery]): FilterPushHandler =
return proc(pubsubTopic: PubsubTopic, msg: WakuMessage) {.async.} =
callEventCallback(ctx, "onReceivedMessage"):
emitEvent("onReceivedMessage"):
$JsonMessageEvent.new(pubsubTopic, msg)
(
@ -38,7 +38,7 @@ proc waku_filter_unsubscribe(
userData: pointer,
pubSubTopic: cstring,
contentTopics: cstring,
) {.ffi.} =
) {.ffiRaw.} =
(
await ctx.myLib[].waku.filterUnsubscribe(
PubsubTopic($pubSubTopic), ($contentTopics).split(",").mapIt(ContentTopic(it))
@ -50,7 +50,7 @@ proc waku_filter_unsubscribe(
proc waku_filter_unsubscribe_all(
ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer
) {.ffi.} =
) {.ffiRaw.} =
(await ctx.myLib[].waku.filterUnsubscribeAll()).isOkOr:
error "fail filter unsubscribe all", error = error
return err(error)

2
library/kernel_api/protocols/lightpush_api.nim
View File

@ -13,7 +13,7 @@ proc waku_lightpush_publish(
userData: pointer,
pubSubTopic: cstring,
jsonWakuMessage: cstring,
) {.ffi.} =
) {.ffiRaw.} =
var jsonMessage: JsonMessage
try:
let jsonContent = parseJson($jsonWakuMessage)

24
library/kernel_api/protocols/relay_api.nim
View File

@ -13,7 +13,7 @@ proc waku_relay_get_peers_in_mesh(
callback: FFICallBack,
userData: pointer,
pubSubTopic: cstring,
) {.ffi.} =
) {.ffiRaw.} =
let peers = (await ctx.myLib[].waku.relayPeersInMesh(PubsubTopic($pubSubTopic))).valueOr:
error "LIST_MESH_PEERS failed", error = error
return err(error)
@ -25,7 +25,7 @@ proc waku_relay_get_num_peers_in_mesh(
callback: FFICallBack,
userData: pointer,
pubSubTopic: cstring,
) {.ffi.} =
) {.ffiRaw.} =
let n = (await ctx.myLib[].waku.relayNumPeersInMesh(PubsubTopic($pubSubTopic))).valueOr:
error "NUM_MESH_PEERS failed", error = error
return err(error)
@ -36,7 +36,7 @@ proc waku_relay_get_connected_peers(
callback: FFICallBack,
userData: pointer,
pubSubTopic: cstring,
) {.ffi.} =
) {.ffiRaw.} =
## Returns the list of all connected peers to an specific pubsub topic
let peers = (await ctx.myLib[].waku.relayConnectedPeers(PubsubTopic($pubSubTopic))).valueOr:
error "LIST_CONNECTED_PEERS failed", error = error
@ -48,7 +48,7 @@ proc waku_relay_get_num_connected_peers(
callback: FFICallBack,
userData: pointer,
pubSubTopic: cstring,
) {.ffi.} =
) {.ffiRaw.} =
let n = (await ctx.myLib[].waku.relayNumConnectedPeers(PubsubTopic($pubSubTopic))).valueOr:
error "NUM_CONNECTED_PEERS failed", error = error
return err(error)
@ -61,7 +61,7 @@ proc waku_relay_add_protected_shard(
clusterId: cint,
shardId: cint,
publicKey: cstring,
) {.ffi.} =
) {.ffiRaw.} =
## Protects a shard with a public key
(
await ctx.myLib[].waku.relayAddProtectedShard(
@ -76,10 +76,10 @@ proc waku_relay_subscribe(
callback: FFICallBack,
userData: pointer,
pubSubTopic: cstring,
) {.ffi.} =
) {.ffiRaw.} =
proc onReceivedMessage(ctx: ptr FFIContext[LogosDelivery]): WakuRelayHandler =
return proc(pubsubTopic: PubsubTopic, msg: WakuMessage) {.async.} =
callEventCallback(ctx, "onReceivedMessage"):
emitEvent("onReceivedMessage"):
$JsonMessageEvent.new(pubsubTopic, msg)
(
@ -96,7 +96,7 @@ proc waku_relay_unsubscribe(
callback: FFICallBack,
userData: pointer,
pubSubTopic: cstring,
) {.ffi.} =
) {.ffiRaw.} =
(await ctx.myLib[].waku.relayUnsubscribe(PubsubTopic($pubSubTopic))).isOkOr:
error "UNSUBSCRIBE failed", error = error
return err(error)
@ -109,7 +109,7 @@ proc waku_relay_publish(
pubSubTopic: cstring,
jsonWakuMessage: cstring,
timeoutMs: cuint,
) {.ffi.} =
) {.ffiRaw.} =
var jsonMessage: JsonMessage
try:
let jsonContent = parseJson($jsonWakuMessage)
@ -132,7 +132,7 @@ proc waku_relay_publish(
proc waku_default_pubsub_topic(
ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer
) {.ffi.} =
) {.ffiRaw.} =
let topic = (await ctx.myLib[].waku.defaultPubsubTopic()).valueOr:
return err(error)
return ok(string(topic))
@ -145,7 +145,7 @@ proc waku_content_topic(
appVersion: cuint,
contentTopicName: cstring,
encoding: cstring,
) {.ffi.} =
) {.ffiRaw.} =
let topic = (
await ctx.myLib[].waku.buildContentTopic(
$appName, uint32(appVersion), $contentTopicName, $encoding
@ -159,7 +159,7 @@ proc waku_pubsub_topic(
callback: FFICallBack,
userData: pointer,
topicName: cstring,
) {.ffi.} =
) {.ffiRaw.} =
let topic = (await ctx.myLib[].waku.buildPubsubTopic($topicName)).valueOr:
return err(error)
return ok(string(topic))

2
library/kernel_api/protocols/store_api.nim
View File

@ -71,7 +71,7 @@ proc waku_store_query(
jsonQuery: cstring,
peerAddr: cstring,
timeoutMs: cint,
) {.ffi.} =
) {.ffiRaw.} =
let jsonContentRes = catch:
parseJson($jsonQuery)

21
library/liblogosdelivery.h
View File

@ -109,16 +109,25 @@ extern "C"
void *userData,
const char *channelId);
// Channel lifecycle events are delivered through the event callback set via
// logosdelivery_set_event_callback: "onChannelMessageReceived" (payload
// base64-encoded), "onChannelMessageSent", "onChannelMessageError".
// Channel lifecycle events are delivered through a per-event listener,
// registered by name: "onChannelMessageReceived" (payload base64-encoded),
// "onChannelMessageSent", "onChannelMessageError".
// Sets a callback that will be invoked whenever an event occurs.
// It is crucial that the passed callback is fast, non-blocking and potentially thread-safe.
void logosdelivery_set_event_callback(void *ctx,
// Registers a callback for the named event and returns a non-zero listener id
// (0 on an invalid context). Register one listener per event name of interest;
// see the README for the full list of event names.
// The callback runs on a dedicated event thread and must be fast,
// non-blocking and thread-safe.
uint64_t logosdelivery_add_event_listener(void *ctx,
const char *eventName,
FFICallBack callback,
void *userData);
// Removes a previously registered listener. Returns 0 on success, 1 if the
// listener id was not found or the context is invalid.
int logosdelivery_remove_event_listener(void *ctx,
uint64_t listenerId);
// Retrieves the list of available node info IDs.
int logosdelivery_get_available_node_info_ids(void *ctx,
FFICallBack callback,

2
library/liblogosdelivery_kernel.h
View File

@ -36,7 +36,7 @@ extern "C"
FFICallBack callback,
void *userData);
// NOTE: event callbacks are registered via logosdelivery_set_event_callback
// NOTE: event callbacks are registered via logosdelivery_add_event_listener
// (declared above) which the waku_* API shares.
int waku_content_topic(void *ctx,

6
library/logos_delivery_api/debug_api.nim
View File

@ -4,7 +4,7 @@ import tools/confutils/[cli_args, config_option_meta]
proc logosdelivery_get_available_node_info_ids(
ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer
) {.ffi.} =
) {.ffiRaw.} =
## Returns the list of all available node info item ids that
## can be queried with `get_node_info_item`.
requireInitializedNode(ctx, "GetNodeInfoIds"):
@ -17,7 +17,7 @@ proc logosdelivery_get_node_info(
callback: FFICallBack,
userData: pointer,
nodeInfoId: cstring,
) {.ffi.} =
) {.ffiRaw.} =
## Returns the content of the node info item with the given id if it exists.
requireInitializedNode(ctx, "GetNodeInfoItem"):
return err(errMsg)
@ -32,7 +32,7 @@ proc logosdelivery_get_node_info(
proc logosdelivery_get_available_configs(
ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer
) {.ffi.} =
) {.ffiRaw.} =
## Returns information about the accepted config items.
requireInitializedNode(ctx, "GetAvailableConfigs"):
return err(errMsg)

6
library/logos_delivery_api/messaging_api.nim
View File

@ -13,7 +13,7 @@ proc logosdelivery_subscribe(
callback: FFICallBack,
userData: pointer,
contentTopicStr: cstring,
) {.ffi.} =
) {.ffiRaw.} =
requireInitializedNode(ctx, "Subscribe"):
return err(errMsg)
@ -34,7 +34,7 @@ proc logosdelivery_unsubscribe(
callback: FFICallBack,
userData: pointer,
contentTopicStr: cstring,
) {.ffi.} =
) {.ffiRaw.} =
requireInitializedNode(ctx, "Unsubscribe"):
return err(errMsg)
@ -55,7 +55,7 @@ proc logosdelivery_send(
callback: FFICallBack,
userData: pointer,
messageJson: cstring,
) {.ffi.} =
) {.ffiRaw.} =
requireInitializedNode(ctx, "Send"):
return err(errMsg)

36
library/logos_delivery_api/node_api.nim
View File

@ -34,9 +34,13 @@ proc logosdelivery_destroy(
ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer
): cint {.dynlib, exportc, cdecl.} =
initializeLibrary()
if not LogosDeliveryFFIPool.isValidCtx(cast[pointer](ctx)):
return RET_ERR
checkParams(ctx, callback, userData)
ffi.destroyFFIContext(ctx).isOkOr:
# Recycle instead of destroy: under refc a full teardown cannot close the
# context signal fds, so every create/destroy cycle would leak them.
ffi.recycleFFIContext(LogosDeliveryFFIPool, ctx).isOkOr:
let msg = "liblogosdelivery error: " & $error
callback(RET_ERR, unsafeAddr msg[0], cast[csize_t](len(msg)), userData)
return RET_ERR
@ -55,7 +59,7 @@ proc logosdelivery_create_node(
echo "error: missing callback in logosdelivery_create_node"
return nil
var ctx = ffi.createFFIContext[LogosDelivery]().valueOr:
var ctx = ffi.createFFIContext(LogosDeliveryFFIPool).valueOr:
let msg = "Error in createFFIContext: " & $error
callback(RET_ERR, unsafeAddr msg[0], cast[csize_t](len(msg)), userData)
return nil
@ -68,8 +72,8 @@ proc logosdelivery_create_node(
let msg = "error in sendRequestToFFIThread: " & $error
callback(RET_ERR, unsafeAddr msg[0], cast[csize_t](len(msg)), userData)
# free allocated resources as they won't be available
ffi.destroyFFIContext(ctx).isOkOr:
chronicles.error "Error in destroyFFIContext after sendRequestToFFIThread during creation",
ffi.recycleFFIContext(LogosDeliveryFFIPool, ctx).isOkOr:
chronicles.error "Error in recycleFFIContext after sendRequestToFFIThread during creation",
err = $error
return nil
@ -77,7 +81,7 @@ proc logosdelivery_create_node(
proc logosdelivery_start_node(
ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer
) {.ffi.} =
) {.ffiRaw.} =
requireInitializedNode(ctx, "START_NODE"):
return err(errMsg)
@ -85,7 +89,7 @@ proc logosdelivery_start_node(
let sentListener = MessageSentEvent.listen(
ctx.myLib[].waku.brokerCtx,
proc(event: MessageSentEvent) {.async: (raises: []).} =
callEventCallback(ctx, "onMessageSent"):
emitEvent("onMessageSent"):
$newJsonEvent("message_sent", event),
).valueOr:
chronicles.error "MessageSentEvent.listen failed", err = $error
@ -94,7 +98,7 @@ proc logosdelivery_start_node(
let errorListener = MessageErrorEvent.listen(
ctx.myLib[].waku.brokerCtx,
proc(event: MessageErrorEvent) {.async: (raises: []).} =
callEventCallback(ctx, "onMessageError"):
emitEvent("onMessageError"):
$newJsonEvent("message_error", event),
).valueOr:
chronicles.error "MessageErrorEvent.listen failed", err = $error
@ -103,7 +107,7 @@ proc logosdelivery_start_node(
let propagatedListener = MessagePropagatedEvent.listen(
ctx.myLib[].waku.brokerCtx,
proc(event: MessagePropagatedEvent) {.async: (raises: []).} =
callEventCallback(ctx, "onMessagePropagated"):
emitEvent("onMessagePropagated"):
$newJsonEvent("message_propagated", event),
).valueOr:
chronicles.error "MessagePropagatedEvent.listen failed", err = $error
@ -112,7 +116,7 @@ proc logosdelivery_start_node(
let receivedListener = MessageReceivedEvent.listen(
ctx.myLib[].waku.brokerCtx,
proc(event: MessageReceivedEvent) {.async: (raises: []).} =
callEventCallback(ctx, "onMessageReceived"):
emitEvent("onMessageReceived"):
$newJsonEvent("message_received", event),
).valueOr:
chronicles.error "MessageReceivedEvent.listen failed", err = $error
@ -121,7 +125,7 @@ proc logosdelivery_start_node(
let ConnectionStatusChangeListener = EventConnectionStatusChange.listen(
ctx.myLib[].waku.brokerCtx,
proc(event: EventConnectionStatusChange) {.async: (raises: []).} =
callEventCallback(ctx, "onConnectionStatusChange"):
emitEvent("onConnectionStatusChange"):
$newJsonEvent("connection_status_change", event),
).valueOr:
chronicles.error "ConnectionStatusChange.listen failed", err = $error
@ -130,7 +134,7 @@ proc logosdelivery_start_node(
let shardTopicHealthListener = EventShardTopicHealthChange.listen(
ctx.myLib[].waku.brokerCtx,
proc(event: EventShardTopicHealthChange) {.async: (raises: []).} =
callEventCallback(ctx, "onTopicHealthChange"):
emitEvent("onTopicHealthChange"):
$(
%*{
"eventType": "relay_topic_health_change",
@ -145,7 +149,7 @@ proc logosdelivery_start_node(
let peerEventListener = WakuPeerEvent.listen(
ctx.myLib[].waku.brokerCtx,
proc(event: WakuPeerEvent) {.async: (raises: []).} =
callEventCallback(ctx, "onConnectionChange"):
emitEvent("onConnectionChange"):
$(
%*{
"eventType": "connection_change",
@ -160,7 +164,7 @@ proc logosdelivery_start_node(
let channelReceivedListener = ChannelMessageReceivedEvent.listen(
ctx.myLib[].waku.brokerCtx,
proc(event: ChannelMessageReceivedEvent) {.async: (raises: []).} =
callEventCallback(ctx, "onChannelMessageReceived"):
emitEvent("onChannelMessageReceived"):
$(
%*{
"eventType": "channel_message_received",
@ -176,7 +180,7 @@ proc logosdelivery_start_node(
let channelSentListener = ChannelMessageSentEvent.listen(
ctx.myLib[].waku.brokerCtx,
proc(event: ChannelMessageSentEvent) {.async: (raises: []).} =
callEventCallback(ctx, "onChannelMessageSent"):
emitEvent("onChannelMessageSent"):
$newJsonEvent("channel_message_sent", event),
).valueOr:
chronicles.error "ChannelMessageSentEvent.listen failed", err = $error
@ -185,7 +189,7 @@ proc logosdelivery_start_node(
let channelErrorListener = ChannelMessageErrorEvent.listen(
ctx.myLib[].waku.brokerCtx,
proc(event: ChannelMessageErrorEvent) {.async: (raises: []).} =
callEventCallback(ctx, "onChannelMessageError"):
emitEvent("onChannelMessageError"):
$newJsonEvent("channel_message_error", event),
).valueOr:
chronicles.error "ChannelMessageErrorEvent.listen failed", err = $error
@ -199,7 +203,7 @@ proc logosdelivery_start_node(
proc logosdelivery_stop_node(
ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer
) {.ffi.} =
) {.ffiRaw.} =
requireInitializedNode(ctx, "STOP_NODE"):
return err(errMsg)

2
logos_delivery.nimble
View File

@ -61,7 +61,7 @@ requires "nim >= 2.2.4",
# Packages not on nimble (use git URLs)
requires "https://github.com/logos-messaging/nim-ffi#v0.1.3"
requires "https://github.com/logos-messaging/nim-ffi#aad9374354a5e3d98964a9adf80766a12f8f200d" # v0.3.0-rc.1
requires "https://github.com/logos-messaging/nim-sds.git#b12f5ee07c5b764303b51fb948b32a4ade1de3b5"

9
nimble.lock
View File

@ -643,18 +643,19 @@
}
},
"ffi": {
"version": "0.1.3",
"vcsRevision": "06111de155253b34e47ed2aaed1d61d08d62cc1b",
"version": "0.3.0",
"vcsRevision": "aad9374354a5e3d98964a9adf80766a12f8f200d",
"url": "https://github.com/logos-messaging/nim-ffi",
"downloadMethod": "git",
"dependencies": [
"nim",
"chronos",
"chronicles",
"taskpools"
"taskpools",
"cbor_serialization"
],
"checksums": {
"sha1": "6f9d49375ea1dc71add55c72ac80a808f238e5b0"
"sha1": "db5fc50aa4717418e481cb0b1f7ca36f2d76586c"
}
},
"boringssl": {

4
nix/deps.nix
View File

@ -285,8 +285,8 @@
ffi = pkgs.fetchgit {
url = "https://github.com/logos-messaging/nim-ffi";
rev = "06111de155253b34e47ed2aaed1d61d08d62cc1b";
sha256 = "0rb0d2i519amgsp7q0bn6m5465z1vwj4rab89529pyiivh3fgh8j";
rev = "aad9374354a5e3d98964a9adf80766a12f8f200d";
sha256 = "075ax4spvzr7idd5b5sncpkr7b3163qncr8fsxv9d02dix4ailqc";
fetchSubmodules = true;
};

2
nix/submodules.json
View File

@ -56,7 +56,7 @@
{
"path": "vendor/nim-ffi",
"url": "https://github.com/logos-messaging/nim-ffi",
"rev": "06111de155253b34e47ed2aaed1d61d08d62cc1b"
"rev": "aad9374354a5e3d98964a9adf80766a12f8f200d"
}
,
{

131
tests-e2e/vendor/logos-delivery-python-bindings/waku/wrapper.py vendored
View File

@ -29,12 +29,18 @@ int logosdelivery_stop_node(
void *userData
);
void logosdelivery_set_event_callback(
uint64_t logosdelivery_add_event_listener(
void *ctx,
const char *eventName,
FFICallBack callback,
void *userData
);
int logosdelivery_remove_event_listener(
void *ctx,
uint64_t listenerId
);
int logosdelivery_destroy(
void *ctx,
FFICallBack callback,
@ -88,6 +94,67 @@ lib = ffi.dlopen(str(_repo_root / "lib" / "liblogosdelivery.so"))
CallbackType = ffi.callback("void(int, const char*, size_t, void*)")
RET_OK = 0
# Non-terminal progress tick. It fires every ~5s while a request is still in
# flight and is always followed by a terminal RET_OK/RET_ERR, so a caller that
# latched it would fail every call slower than five seconds -- start_node most
# of all, since it boots the node and joins the network.
RET_STALE_WARN = 3
# Since 0.3.0 a listener is registered per event name, so a caller that wants
# every event registers once per name.
EVENT_NAMES = (
"onMessageSent",
"onMessageError",
"onMessagePropagated",
"onMessageReceived",
"onConnectionStatusChange",
"onTopicHealthChange",
"onConnectionChange",
"onReceivedMessage",
"onChannelMessageReceived",
"onChannelMessageSent",
"onChannelMessageError",
)
_CBOR_MAJOR_BYTES = 2
_CBOR_MAJOR_TEXT = 3
# The event thread calls these from outside Python. cffi frees the trampoline
# when the object dies, so a late event would land on freed memory; keep every
# event callback alive for the whole process.
_PINNED_EVENT_CALLBACKS = []
def _decode_cbor_string(raw: bytes) -> bytes:
"""Unwrap a CBOR definite-length text/byte string.
Since 0.3.0 the library encodes every RET_OK reply payload with CBOR. Error
payloads and the RET_STALE_WARN tick stay plain, and so do event payloads.
"""
if not raw:
return b""
header = raw[0]
if header >> 5 not in (_CBOR_MAJOR_BYTES, _CBOR_MAJOR_TEXT):
raise ValueError(f"reply is not a CBOR string: header {header:#04x}")
info = header & 0x1F
if info < 24:
length, offset = info, 1
elif info <= 27:
size = 1 << (info - 24)
length, offset = int.from_bytes(raw[1 : 1 + size], "big"), 1 + size
else:
raise ValueError(f"unsupported CBOR string header: {header:#04x}")
payload = raw[offset : offset + length]
if len(payload) != length:
raise ValueError(f"truncated CBOR string: want {length} bytes, got {len(payload)}")
return payload
def _new_cb_state():
return {
@ -106,10 +173,17 @@ def _wait_cb_raw(
if not ok:
return Err(f"{op_name}: timeout after {timeout_s}s")
if state["ret"] is None:
cb_ret = state["ret"]
if cb_ret is None:
return Err(f"{op_name}: callback ret is None")
return Ok((state["ret"], state["msg"]))
if cb_ret != RET_OK:
return Ok((cb_ret, state["msg"]))
try:
return Ok((cb_ret, _decode_cbor_string(state["msg"])))
except ValueError as e:
return Err(f"{op_name}: {e}")
def _wait_cb_ok(state, op_name: str, timeout_s: float = 20.0) -> Result[int, str]:
@ -125,14 +199,18 @@ def _wait_cb_ok(state, op_name: str, timeout_s: float = 20.0) -> Result[int, str
class NodeWrapper:
def __init__(self, ctx, config_buffer, event_cb_handler):
def __init__(self, ctx, config_buffer, event_cb_handler, listener_ids=()):
self.ctx = ctx
self._config_buffer = config_buffer
self._event_cb_handler = event_cb_handler
self._listener_ids = tuple(listener_ids)
@staticmethod
def _make_waiting_cb(state):
def c_cb(ret, char_p, length, userData):
if int(ret) == RET_STALE_WARN:
return
msg = ffi.buffer(char_p, length)[:] if char_p != ffi.NULL else b""
if not state["done"].is_set():
@ -148,7 +226,9 @@ class NodeWrapper:
msg = ffi.buffer(char_p, length)[:] if char_p != ffi.NULL else b""
py_callback(int(ret), msg)
return CallbackType(c_cb)
handler = CallbackType(c_cb)
_PINNED_EVENT_CALLBACKS.append(handler)
return handler
@classmethod
def create_node(
@ -173,20 +253,30 @@ class NodeWrapper:
if ctx == ffi.NULL:
return Err("create_node: ctx is NULL")
node = cls(ctx, config_buffer, None)
wait_result = _wait_cb_ok(state, "create_node", timeout_s)
if wait_result.is_err():
node.destroy()
return Err(wait_result.err())
event_cb_handler = None
if event_cb is not None:
event_cb_handler = cls._make_event_cb(event_cb)
lib.logosdelivery_set_event_callback(
if event_cb is None:
return Ok(node)
node._event_cb_handler = cls._make_event_cb(event_cb)
for event_name in EVENT_NAMES:
listener_id = lib.logosdelivery_add_event_listener(
ctx,
event_cb_handler,
event_name.encode("utf-8"),
node._event_cb_handler,
ffi.NULL,
)
if listener_id == 0:
node.destroy()
return Err(f"create_node: add_event_listener({event_name}) failed")
node._listener_ids += (listener_id,)
return Ok(cls(ctx, config_buffer, event_cb_handler))
return Ok(node)
@classmethod
def create_and_start(
@ -208,6 +298,9 @@ class NodeWrapper:
start_result = node.start_node(timeout_s=timeout_s)
if start_result.is_err():
# The caller drops the node here, so tear it down before its
# callbacks outlive the wrapper that owns them.
node.destroy(timeout_s=timeout_s)
return Err(start_result.err())
return Ok(node)
@ -233,6 +326,15 @@ class NodeWrapper:
return _wait_cb_ok(state, "stop_node", timeout_s)
def destroy(self, *, timeout_s: float = 20.0) -> Result[int, str]:
if self.ctx == ffi.NULL:
return Ok(RET_OK)
# Drop the listeners first so the event thread cannot reach the Python
# callback once the context is gone.
for listener_id in self._listener_ids:
lib.logosdelivery_remove_event_listener(self.ctx, listener_id)
self._listener_ids = ()
state = _new_cb_state()
cb = self._make_waiting_cb(state)
@ -249,10 +351,15 @@ class NodeWrapper:
def stop_and_destroy(self, *, timeout_s: float = 20.0) -> Result[int, str]:
stop_result = self.stop_node(timeout_s=timeout_s)
# Destroy even when the stop fails: a node that keeps its context alive
# also keeps calling back into a wrapper the caller is about to drop.
destroy_result = self.destroy(timeout_s=timeout_s)
if stop_result.is_err():
return Err(stop_result.err())
return self.destroy(timeout_s=timeout_s)
return destroy_result
def subscribe_content_topic(self, content_topic: str, *, timeout_s: float = 20.0) -> Result[int, str]:
state = _new_cb_state()