From 4a85db1b6ab28f072e6690b5cd4e2ec59e50d837 Mon Sep 17 00:00:00 2001 From: Gabriel Cruz <8129788+gmelodie@users.noreply.github.com> Date: Thu, 6 Aug 2026 23:53:38 -0300 Subject: [PATCH] feat(ffi)!: migrate liblogosdelivery to the nim-ffi 0.3.0 typed C ABI (#4082) --- .gitignore | 3 + library/MESSAGE_EVENTS.md | 16 +- library/README.md | 206 ++++++++---- library/channels_api/channel_api.nim | 72 ++--- library/declare_lib.nim | 39 +-- library/examples/logosdelivery_example.c | 76 +++-- library/kernel_api/debug_node_api.nim | 36 +-- library/kernel_api/discovery_api.nim | 45 +-- library/kernel_api/peer_manager_api.nim | 76 ++--- library/kernel_api/ping_api.nim | 10 +- library/kernel_api/protocols/filter_api.nim | 36 +-- .../kernel_api/protocols/lightpush_api.nim | 18 +- library/kernel_api/protocols/relay_api.nim | 129 +++----- library/kernel_api/protocols/store_api.nim | 13 +- library/liblogosdelivery.h | 159 ++-------- library/liblogosdelivery.nim | 3 + library/liblogosdelivery_kernel.h | 241 +------------- library/logos_delivery_api/debug_api.nim | 44 +-- library/logos_delivery_api/messaging_api.nim | 49 +-- library/logos_delivery_api/node_api.nim | 186 +++++------ logos_delivery.nimble | 17 +- nimble.lock | 4 +- nix/deps.nix | 4 +- nix/submodules.json | 2 +- tests-e2e/src/node/wrapper_helpers.py | 4 +- tests-e2e/src/node/wrappers_manager.py | 19 +- .../waku/wrapper.py | 300 +++++++----------- 27 files changed, 646 insertions(+), 1161 deletions(-) diff --git a/.gitignore b/.gitignore index bfd8f269f..6dce81c7f 100644 --- a/.gitignore +++ b/.gitignore @@ -91,3 +91,6 @@ nimbledeps # Python bytecode from tests/simulator __pycache__/ *.pyc + +# Emitted by genBindings() during the liblogosdelivery build. +library/generated/ diff --git a/library/MESSAGE_EVENTS.md b/library/MESSAGE_EVENTS.md index 1c2640fda..42174e2d1 100644 --- a/library/MESSAGE_EVENTS.md +++ b/library/MESSAGE_EVENTS.md @@ -86,13 +86,17 @@ 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)`. +listener id you can later pass to `logosdelivery_remove_event_listener(rawCtx, id)`. + +The event API takes the raw context, which is the `ptr` field of the +`LogosDeliveryCtx` that `logosdelivery_ctx_create` hands to its callback. ```c -void *ctx = logosdelivery_create_node(config, callback, userData); -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); +// ctx comes from the logosdelivery_ctx_create callback; see the README. +void *rawCtx = ctx->ptr; +logosdelivery_add_event_listener(rawCtx, "onMessageSent", event_callback, NULL); +logosdelivery_add_event_listener(rawCtx, "onMessagePropagated", event_callback, NULL); +logosdelivery_add_event_listener(rawCtx, "onMessageError", event_callback, NULL); ``` ### 3. Start the Node @@ -100,7 +104,7 @@ logosdelivery_add_event_listener(ctx, "onMessageError", event_callback, NULL); Once the node is started, events will be delivered to your callback: ```c -logosdelivery_start_node(ctx, callback, userData); +logosdelivery_ctx_start_node(ctx, on_reply, userData); ``` ## Event Flow diff --git a/library/README.md b/library/README.md index 84e48bc2b..82e1a77c8 100644 --- a/library/README.md +++ b/library/README.md @@ -6,27 +6,68 @@ A C FFI library providing a simplified interface to Logos Messaging functionalit This library wraps the high-level API functions from `waku/api/api.nim` and exposes them via a C FFI interface, making them accessible from C, C++, and other languages that support C FFI. +The call surface is generated by nim-ffi from the `{.ffi.}` annotations in +`library/*.nim`. `make liblogosdelivery` writes it to +`library/generated/logosdelivery.h` on every build, so it can never drift from +the Nim signatures. It is a build artifact and is not checked in: build the +library before you compile anything against it. + +Include `library/liblogosdelivery.h`, which pulls in the generated header and +adds the event-listener ABI. + +Every entry point takes the context handle (`void *ctx`) first, except the +constructor. The rest of the signature depends on the call: + +- No-argument calls (`start_node`, `stop_node`, `get_available_configs`, + `get_available_node_info_ids`) take a raw `LogosDeliveryScalarRawFn`: + `(void *ctx, LogosDeliveryScalarRawFn cb, void *userData)`. +- Argument-taking calls (`subscribe`, `unsubscribe`, `send`, `get_node_info`) + take a per-call `LogosDeliveryReplyFn` and pass their arguments last, in + a request struct: + `(void *ctx, LogosDeliveryReplyFn onReply, void *userData, const Req *req)`. + +The generator emits one reply typedef per call (e.g. `LogosDeliverySubscribeReplyFn`), +all with the same shape: + +```c +typedef void (*LogosDeliveryScalarRawFn)(int callerRet, char *msg, size_t len, void *userData); +typedef void (*LogosDeliverySubscribeReplyFn)(int errCode, const char *reply, const char *errMsg, void *userData); +``` + +`reply`, `errMsg` and `msg` are borrowed: copy them if you need them after the +callback returns. + ## API Functions ### Node Lifecycle #### `logosdelivery_create_node` -Creates a new instance of the node from the given configuration JSON. +Creates a node from the given configuration JSON. ```c +typedef struct { const char *configJson; } CreateNodeCtorReq; + +typedef void (*LogosDeliveryCreateRawFn)( + int errCode, + const char *ctxAddr, // context address as decimal text, on success + const char *errMsg, + void *userData +); + void *logosdelivery_create_node( - const char *configJson, - FFICallBack callback, + const CreateNodeCtorReq *req, + LogosDeliveryCreateRawFn onCreated, void *userData ); ``` **Parameters:** -- `configJson`: JSON string containing node configuration -- `callback`: Callback function to receive the result +- `req->configJson`: JSON string containing node configuration +- `onCreated`: Callback that receives the terminal result - `userData`: User data passed to the callback -**Returns:** Pointer to the context needed by other API functions, or NULL on error. +**Returns:** the context handle, or `NULL` on failure. Creation is asynchronous: +wait for `onCreated` before you make any other call. **Example configuration JSON:** ```json @@ -63,33 +104,22 @@ Available presets: Starts the node. ```c -int logosdelivery_start_node( - void *ctx, - FFICallBack callback, - void *userData -); +int logosdelivery_start_node(void *ctx, LogosDeliveryScalarRawFn callback, void *userData); ``` #### `logosdelivery_stop_node` Stops the node. ```c -int logosdelivery_stop_node( - void *ctx, - FFICallBack callback, - void *userData -); +int logosdelivery_stop_node(void *ctx, LogosDeliveryScalarRawFn callback, void *userData); ``` #### `logosdelivery_destroy` -Destroys a node instance and frees resources. +Destroys a node instance and frees resources. This call is synchronous; do not +use `ctx` afterwards. ```c -int logosdelivery_destroy( - void *ctx, - FFICallBack callback, - void *userData -); +int logosdelivery_destroy(void *ctx); ``` ### Messaging @@ -98,29 +128,33 @@ int logosdelivery_destroy( Subscribe to a content topic to receive messages. ```c +typedef struct { const char *contentTopicStr; } SubscribeReq; + int logosdelivery_subscribe( void *ctx, - FFICallBack callback, + LogosDeliverySubscribeReplyFn onReply, void *userData, - const char *contentTopic + const SubscribeReq *req ); ``` **Parameters:** -- `ctx`: Context pointer from `logosdelivery_create_node` -- `callback`: Callback function to receive the result +- `ctx`: Context handle returned by `logosdelivery_create_node` +- `req->contentTopicStr`: Content topic string (e.g., "/myapp/1/chat/proto") +- `onReply`: Callback function to receive the result - `userData`: User data passed to the callback -- `contentTopic`: Content topic string (e.g., "/myapp/1/chat/proto") #### `logosdelivery_unsubscribe` Unsubscribe from a content topic. ```c +typedef struct { const char *contentTopicStr; } UnsubscribeReq; + int logosdelivery_unsubscribe( void *ctx, - FFICallBack callback, + LogosDeliveryUnsubscribeReplyFn onReply, void *userData, - const char *contentTopic + const UnsubscribeReq *req ); ``` @@ -128,16 +162,18 @@ int logosdelivery_unsubscribe( Send a message. ```c +typedef struct { const char *messageJson; } SendReq; + int logosdelivery_send( void *ctx, - FFICallBack callback, + LogosDeliverySendReplyFn onReply, void *userData, - const char *messageJson + const SendReq *req ); ``` **Parameters:** -- `messageJson`: JSON string containing the message +- `req->messageJson`: JSON string containing the message **Example message JSON:** ```json @@ -206,15 +242,43 @@ make liblogosdeliveryDynamic # Build dynamic library All functions that return `int` use the following return codes: -- `RET_OK` (0): Success -- `RET_ERR` (1): Error -- `RET_MISSING_CALLBACK` (2): Missing callback function +- `NIMFFI_RET_OK` / `RET_OK` (0): Success +- `NIMFFI_RET_ERR` / `RET_ERR` (1): Error +- `NIMFFI_RET_MISSING_CALLBACK` / `RET_MISSING_CALLBACK` (2): Missing callback function +- `NIMFFI_RET_STALE_WARN` (3): Non-terminal progress tick, always followed by a + terminal code. Ignore it unless you want progress. -## Callback Function +## Callback Functions -All API functions use the following callback signature: +Results come back through one of four callback shapes. The generated names carry +the library prefix (`LogosDelivery`); the reply typedef is emitted once per call. ```c +// Argument-taking calls: one typedef per call, all this shape. +typedef void (*LogosDeliverySubscribeReplyFn)( + int errCode, + const char *reply, + const char *errMsg, + void *userData +); + +// No-argument calls (start/stop/get_available_*). +typedef void (*LogosDeliveryScalarRawFn)( + int callerRet, + char *msg, + size_t len, + void *userData +); + +// Constructor. +typedef void (*LogosDeliveryCreateRawFn)( + int errCode, + const char *ctxAddr, + const char *errMsg, + void *userData +); + +// Event listeners (declared by liblogosdelivery.h, not the generated header). typedef void (*FFICallBack)( int callerRet, const char *msg, @@ -223,44 +287,68 @@ typedef void (*FFICallBack)( ); ``` -**Parameters:** -- `callerRet`: Return code (RET_OK, RET_ERR, etc.) -- `msg`: Response message (may be empty for success) -- `len`: Length of the message -- `userData`: User data passed in the original call +- Reply typedefs (`LogosDeliveryReplyFn`): `reply` is the result on success + (NUL-terminated, may be empty); `errMsg` is the message on failure. +- `LogosDeliveryScalarRawFn` and `FFICallBack`: `msg` holds `len` bytes and is + not NUL-terminated. +- `LogosDeliveryCreateRawFn`: `ctxAddr` is the context address as decimal text on + success. + +All of these strings are borrowed and valid only for the duration of the call. +Copy them if you need them afterwards. ## Example Usage ```c #include "liblogosdelivery.h" #include +#include -void callback(int ret, const char *msg, size_t len, void *userData) { +static volatile int created = -1; +static void *node = NULL; + +// The argument-taking calls share this reply shape. +void on_reply(int ret, const char *reply, const char *errMsg, void *userData) { if (ret == RET_OK) { - printf("Success: %.*s\n", (int)len, msg); + printf("Success: %s\n", reply ? reply : ""); } else { - printf("Error: %.*s\n", (int)len, msg); + printf("Error: %s\n", errMsg ? errMsg : "unknown error"); } } +// The no-argument calls (start/stop) take the raw callback. +void on_scalar(int ret, char *msg, size_t len, void *userData) { + if (ret == RET_STALE_WARN) return; // progress tick, ignore + printf("%.*s\n", (int)len, msg ? msg : ""); +} + +void on_created(int ret, const char *ctxAddr, const char *errMsg, void *userData) { + created = (ret == RET_OK); +} + int main() { const char *config = "{" - "\"logLevel\": \"INFO\"," "\"mode\": \"Core\"," "\"preset\": \"logos.dev\"" "}"; - // Create node - void *ctx = logosdelivery_create_node(config, callback, NULL); - if (ctx == NULL) { + // Create the node. The return value is the context handle; wait for + // on_created before making any other call. + CreateNodeCtorReq createReq = { .configJson = config }; + node = logosdelivery_create_node(&createReq, on_created, NULL); + for (int i = 0; i < 100 && created == -1; i++) { + usleep(100000); + } + if (created != 1 || node == NULL) { return 1; } // Start node - logosdelivery_start_node(ctx, callback, NULL); + logosdelivery_start_node(node, on_scalar, NULL); // Subscribe to a topic - logosdelivery_subscribe(ctx, callback, NULL, "/myapp/1/chat/proto"); + SubscribeReq subReq = { .contentTopicStr = "/myapp/1/chat/proto" }; + logosdelivery_subscribe(node, on_reply, NULL, &subReq); // Send a message const char *msg = "{" @@ -268,11 +356,12 @@ int main() { "\"payload\": \"SGVsbG8gV29ybGQ=\"," "\"ephemeral\": false" "}"; - logosdelivery_send(ctx, callback, NULL, msg); + SendReq sendReq = { .messageJson = msg }; + logosdelivery_send(node, on_reply, NULL, &sendReq); - // Clean up - logosdelivery_stop_node(ctx, callback, NULL); - logosdelivery_destroy(ctx, callback, NULL); + // Clean up. logosdelivery_destroy is synchronous. + logosdelivery_stop_node(node, on_scalar, NULL); + logosdelivery_destroy(node); return 0; } @@ -282,11 +371,12 @@ int main() { The library is structured as follows: -- `liblogosdelivery.h`: C header file with function declarations +- `liblogosdelivery.h`: Public C header; includes the generated header and adds the event ABI +- `generated/logosdelivery.h`: Generated call surface, emitted by `make liblogosdelivery` (not checked in) - `liblogosdelivery.nim`: Main library entry point - `declare_lib.nim`: Library declaration and initialization -- `lmapi/node_api.nim`: Node lifecycle API implementation -- `lmapi/messaging_api.nim`: Subscribe/send API implementation +- `logos_delivery_api/node_api.nim`: Node lifecycle API implementation +- `logos_delivery_api/messaging_api.nim`: Subscribe/send API implementation The library uses the nim-ffi framework for FFI infrastructure, which handles: - Thread-safe request processing diff --git a/library/channels_api/channel_api.nim b/library/channels_api/channel_api.nim index 67058b0af..7f423d477 100644 --- a/library/channels_api/channel_api.nim +++ b/library/channels_api/channel_api.nim @@ -8,60 +8,42 @@ import ../declare_lib proc logosdelivery_channel_create( - ctx: ptr FFIContext[LogosDelivery], - callback: FFICallBack, - userData: pointer, - channelIdStr: cstring, - contentTopicStr: cstring, - senderIdStr: cstring, -) {.ffiRaw.} = - requireInitializedNode(ctx, "ChannelCreate"): + self: LogosDelivery, + channelIdStr: string, + contentTopicStr: string, + senderIdStr: string, +): Future[Result[string, string]] {.ffi.} = + requireChannels(self, "ChannelCreate"): return err(errMsg) - requireChannels(ctx, "ChannelCreate"): - return err(errMsg) - - let id = ctx.myLib[].reliableChannelManager.createReliableChannel( - ChannelId($channelIdStr), - ContentTopic($contentTopicStr), - SdsParticipantID($senderIdStr), + let id = self.reliableChannelManager.createReliableChannel( + ChannelId(channelIdStr), + ContentTopic(contentTopicStr), + SdsParticipantID(senderIdStr), ).valueOr: return err("ChannelCreate failed: " & $error) return ok(string(id)) proc logosdelivery_channel_exists( - ctx: ptr FFIContext[LogosDelivery], - callback: FFICallBack, - userData: pointer, - channelIdStr: cstring, -) {.ffiRaw.} = + self: LogosDelivery, channelIdStr: string +): Future[Result[string, string]] {.ffi.} = ## Returns `"true"` or `"false"`; a missing channel is not an error. - requireInitializedNode(ctx, "ChannelExists"): + requireChannels(self, "ChannelExists"): return err(errMsg) - requireChannels(ctx, "ChannelExists"): - return err(errMsg) - - return ok($ctx.myLib[].reliableChannelManager.channelExists(ChannelId($channelIdStr))) + return ok($self.reliableChannelManager.channelExists(ChannelId(channelIdStr))) proc logosdelivery_channel_send( - ctx: ptr FFIContext[LogosDelivery], - callback: FFICallBack, - userData: pointer, - channelIdStr: cstring, - messageJson: cstring, -) {.ffiRaw.} = + self: LogosDelivery, channelIdStr: string, messageJson: string +): Future[Result[string, string]] {.ffi.} = ## `messageJson` carries `{ "payload": , "ephemeral": }`. - requireInitializedNode(ctx, "ChannelSend"): - return err(errMsg) - - requireChannels(ctx, "ChannelSend"): + requireChannels(self, "ChannelSend"): return err(errMsg) var jsonNode: JsonNode try: - jsonNode = parseJson($messageJson) + jsonNode = parseJson(messageJson) except Exception as e: return err("Failed to parse channel message JSON: " & e.msg) @@ -74,27 +56,19 @@ proc logosdelivery_channel_send( let ephemeral = jsonNode.getOrDefault("ephemeral").getBool(false) let requestId = ( - await ctx.myLib[].reliableChannelManager.send( - ChannelId($channelIdStr), payload, ephemeral - ) + await self.reliableChannelManager.send(ChannelId(channelIdStr), payload, ephemeral) ).valueOr: return err("ChannelSend failed: " & $error) return ok($requestId) proc logosdelivery_channel_close( - ctx: ptr FFIContext[LogosDelivery], - callback: FFICallBack, - userData: pointer, - channelIdStr: cstring, -) {.ffiRaw.} = - requireInitializedNode(ctx, "ChannelClose"): + self: LogosDelivery, channelIdStr: string +): Future[Result[string, string]] {.ffi.} = + requireChannels(self, "ChannelClose"): return err(errMsg) - requireChannels(ctx, "ChannelClose"): - return err(errMsg) - - (await ctx.myLib[].reliableChannelManager.closeChannel(ChannelId($channelIdStr))).isOkOr: + (await self.reliableChannelManager.closeChannel(ChannelId(channelIdStr))).isOkOr: return err("ChannelClose failed: " & $error) return ok("") diff --git a/library/declare_lib.nim b/library/declare_lib.nim index 515b75f80..d2b72b3c4 100644 --- a/library/declare_lib.nim +++ b/library/declare_lib.nim @@ -2,16 +2,7 @@ import ffi import results import logos_delivery -declareLibrary("logosdelivery", LogosDelivery) - -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 +declareLibrary("logosdelivery", LogosDelivery, defaultABIFormat = "c") template emitEvent*(eventName: string, body: untyped) = ## Enqueues `body`'s payload for nim-ffi's event thread to fan out to listeners. @@ -26,30 +17,14 @@ template emitEvent*(eventName: string, body: untyped) = 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 -) = - if isNil(ctx): - let errMsg {.inject.} = opName & " failed: invalid context" - onError - elif isNil(ctx.myLib) or isNil(ctx.myLib[]): - let errMsg {.inject.} = opName & " failed: node is not initialized" - onError - -template requireMessaging*( - ctx: ptr FFIContext[LogosDelivery], opName: string, onError: untyped -) = - ## Use after `requireInitializedNode`. Fails if the node has no messaging client - ## (a kernel-only / fleet node). - ctx.myLib[].ensureMessaging().isOkOr: +template requireMessaging*(self: LogosDelivery, opName: string, onError: untyped) = + ## Fails if the node has no messaging client (a kernel-only / fleet node). + self.ensureMessaging().isOkOr: let errMsg {.inject.} = opName & " failed: " & error onError -template requireChannels*( - ctx: ptr FFIContext[LogosDelivery], opName: string, onError: untyped -) = - ## Use after `requireInitializedNode`. Fails if the node has no reliable channel - ## manager (a kernel-only / fleet node). - ctx.myLib[].ensureChannels().isOkOr: +template requireChannels*(self: LogosDelivery, opName: string, onError: untyped) = + ## Fails if the node has no reliable channel manager (a kernel-only / fleet node). + self.ensureChannels().isOkOr: let errMsg {.inject.} = opName & " failed: " & error onError diff --git a/library/examples/logosdelivery_example.c b/library/examples/logosdelivery_example.c index 99f9a24ab..d52de1144 100644 --- a/library/examples/logosdelivery_example.c +++ b/library/examples/logosdelivery_example.c @@ -107,20 +107,35 @@ void event_callback(int ret, const char *msg, size_t len, void *userData) { free(eventJson); } -// Simple callback that prints results -void simple_callback(int ret, const char *msg, size_t len, void *userData) { - const char *operation = (const char *)userData; - - if (operation != NULL && strcmp(operation, "create_node") == 0) { - create_node_ok = (ret == RET_OK) ? 1 : 0; +// Constructor callback (LogosDeliveryCreateRawFn): reports the terminal result +// of create_node. `ctxAddr` is the context address as text on success. +void on_created(int ret, const char *ctxAddr, const char *errMsg, void *userData) { + create_node_ok = (ret == RET_OK) ? 1 : 0; + if (ret != RET_OK) { + printf("[create_node] Error: %s\n", errMsg ? errMsg : "unknown error"); } +} +// Reply callback for the argument-taking calls (subscribe, unsubscribe, send, +// get_node_info). `reply` is the result on success, `errMsg` on failure. +void on_reply(int ret, const char *reply, const char *errMsg, void *userData) { + const char *operation = (const char *)userData; if (ret == RET_OK) { - if (len > 0) { - printf("[%s] Success: %.*s\n", operation, (int)len, msg); - } else { - printf("[%s] Success\n", operation); - } + printf("[%s] Success: %s\n", operation, reply ? reply : ""); + } else { + printf("[%s] Error: %s\n", operation, errMsg ? errMsg : "unknown error"); + } +} + +// Raw callback for the no-argument calls (start_node, stop_node, +// get_available_*). `msg` is `len` bytes and not NUL-terminated. +void on_scalar(int ret, char *msg, size_t len, void *userData) { + const char *operation = (const char *)userData; + if (ret == RET_STALE_WARN) { + return; // non-terminal progress tick + } + if (ret == RET_OK) { + printf("[%s] Success: %.*s\n", operation, (int)len, msg); } else { printf("[%s] Error: %.*s\n", operation, (int)len, msg); } @@ -140,7 +155,8 @@ int main() { "}"; printf("1. Creating node...\n"); - void *ctx = logosdelivery_create_node(config, simple_callback, (void *)"create_node"); + CreateNodeCtorReq createReq = { .configJson = config }; + void *ctx = logosdelivery_create_node(&createReq, on_created, NULL); if (ctx == NULL) { printf("Failed to create node\n"); return 1; @@ -151,7 +167,7 @@ int main() { if (create_node_ok != 1) { printf("Create node failed, stopping example early.\n"); - logosdelivery_destroy(ctx, simple_callback, (void *)"destroy"); + logosdelivery_destroy(ctx); return 1; } @@ -162,33 +178,35 @@ int main() { printf("Event listeners registered for message events\n"); printf("\n3. Starting node...\n"); - logosdelivery_start_node(ctx, simple_callback, (void *)"start_node"); + logosdelivery_start_node(ctx, on_scalar, (void *)"start_node"); // Wait for node to start sleep(5); printf("\n4. Subscribing to content topic...\n"); const char *contentTopic = "/example/1/chat/proto"; - logosdelivery_subscribe(ctx, simple_callback, (void *)"subscribe", contentTopic); + SubscribeReq subscribeReq = { .contentTopicStr = contentTopic }; + logosdelivery_subscribe(ctx, on_reply, (void *)"subscribe", &subscribeReq); // Wait for subscription sleep(1); - printf("\n5. Retrieving all possibl node info ids...\n"); - logosdelivery_get_available_node_info_ids(ctx, simple_callback, (void *)"get_available_node_info_ids"); + printf("\n5. Retrieving all possible node info ids...\n"); + logosdelivery_get_available_node_info_ids(ctx, on_scalar, (void *)"get_available_node_info_ids"); printf("\nRetrieving node info for a specific invalid ID...\n"); - logosdelivery_get_node_info(ctx, simple_callback, (void *)"get_node_info", "WrongNodeInfoId"); + GetNodeInfoReq nodeInfoReq = { .nodeInfoId = "WrongNodeInfoId" }; + logosdelivery_get_node_info(ctx, on_reply, (void *)"get_node_info", &nodeInfoReq); printf("\nRetrieving several node info for specific correct IDs...\n"); - logosdelivery_get_node_info(ctx, simple_callback, (void *)"get_node_info", "Version"); - // logosdelivery_get_node_info(ctx, simple_callback, (void *)"get_node_info", "Metrics"); - logosdelivery_get_node_info(ctx, simple_callback, (void *)"get_node_info", "MyMultiaddresses"); - logosdelivery_get_node_info(ctx, simple_callback, (void *)"get_node_info", "MyENR"); - logosdelivery_get_node_info(ctx, simple_callback, (void *)"get_node_info", "MyPeerId"); + const char *nodeInfoIds[] = {"Version", "MyMultiaddresses", "MyENR", "MyPeerId"}; + for (size_t i = 0; i < sizeof(nodeInfoIds) / sizeof(nodeInfoIds[0]); i++) { + GetNodeInfoReq req = { .nodeInfoId = nodeInfoIds[i] }; + logosdelivery_get_node_info(ctx, on_reply, (void *)"get_node_info", &req); + } printf("\nRetrieving available configs...\n"); - logosdelivery_get_available_configs(ctx, simple_callback, (void *)"get_available_configs"); + logosdelivery_get_available_configs(ctx, on_scalar, (void *)"get_available_configs"); printf("\n6. Sending a message...\n"); printf("Watch for message events (sent, propagated, or error):\n"); @@ -198,7 +216,8 @@ int main() { "\"payload\": \"SGVsbG8sIExvZ29zIE1lc3NhZ2luZyE=\"," "\"ephemeral\": false" "}"; - logosdelivery_send(ctx, simple_callback, (void *)"send", message); + SendReq sendReq = { .messageJson = message }; + logosdelivery_send(ctx, on_reply, (void *)"send", &sendReq); // Poll for terminal message events (sent, error, or received) with timeout printf("Waiting for message delivery events...\n"); @@ -214,17 +233,18 @@ int main() { } printf("\n7. Unsubscribing from content topic...\n"); - logosdelivery_unsubscribe(ctx, simple_callback, (void *)"unsubscribe", contentTopic); + UnsubscribeReq unsubscribeReq = { .contentTopicStr = contentTopic }; + logosdelivery_unsubscribe(ctx, on_reply, (void *)"unsubscribe", &unsubscribeReq); sleep(1); printf("\n8. Stopping node...\n"); - logosdelivery_stop_node(ctx, simple_callback, (void *)"stop_node"); + logosdelivery_stop_node(ctx, on_scalar, (void *)"stop_node"); sleep(1); printf("\n9. Destroying context...\n"); - logosdelivery_destroy(ctx, simple_callback, (void *)"destroy"); + logosdelivery_destroy(ctx); printf("\n=== Example completed ===\n"); return 0; diff --git a/library/kernel_api/debug_node_api.nim b/library/kernel_api/debug_node_api.nim index 7c406a5b6..60fa84b49 100644 --- a/library/kernel_api/debug_node_api.nim +++ b/library/kernel_api/debug_node_api.nim @@ -2,45 +2,35 @@ import std/strutils import chronos, results, ffi import logos_delivery, library/declare_lib -proc waku_version( - ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer -) {.ffiRaw.} = - let v = (await ctx.myLib[].waku.version()).valueOr: +proc waku_version(self: LogosDelivery): Future[Result[string, string]] {.ffi.} = + let v = (await self.waku.version()).valueOr: return err(error) return ok(v) proc waku_listen_addresses( - ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer -) {.ffiRaw.} = + self: LogosDelivery +): Future[Result[string, string]] {.ffi.} = ## returns a comma-separated string of the listen addresses - let addrs = (await ctx.myLib[].waku.listenAddresses()).valueOr: + let addrs = (await self.waku.listenAddresses()).valueOr: return err(error) return ok(addrs.join(",")) -proc waku_get_my_enr( - ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer -) {.ffiRaw.} = - let enrUri = (await ctx.myLib[].waku.myEnr()).valueOr: +proc waku_get_my_enr(self: LogosDelivery): Future[Result[string, string]] {.ffi.} = + let enrUri = (await self.waku.myEnr()).valueOr: return err(error) return ok(enrUri) -proc waku_get_my_peerid( - ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer -) {.ffiRaw.} = - let peerId = (await ctx.myLib[].waku.myPeerId()).valueOr: +proc waku_get_my_peerid(self: LogosDelivery): Future[Result[string, string]] {.ffi.} = + let peerId = (await self.waku.myPeerId()).valueOr: return err(error) return ok(peerId) -proc waku_get_metrics( - ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer -) {.ffiRaw.} = - let m = (await ctx.myLib[].waku.metrics()).valueOr: +proc waku_get_metrics(self: LogosDelivery): Future[Result[string, string]] {.ffi.} = + let m = (await self.waku.metrics()).valueOr: return err(error) return ok(m) -proc waku_is_online( - ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer -) {.ffiRaw.} = - let online = (await ctx.myLib[].waku.isOnline()).valueOr: +proc waku_is_online(self: LogosDelivery): Future[Result[string, string]] {.ffi.} = + let online = (await self.waku.isOnline()).valueOr: return err(error) return ok($online) diff --git a/library/kernel_api/discovery_api.nim b/library/kernel_api/discovery_api.nim index 234ffe717..6c1c2ad1a 100644 --- a/library/kernel_api/discovery_api.nim +++ b/library/kernel_api/discovery_api.nim @@ -3,57 +3,40 @@ import chronos, chronicles, results, ffi import logos_delivery, library/declare_lib proc waku_discv5_update_bootnodes( - ctx: ptr FFIContext[LogosDelivery], - callback: FFICallBack, - userData: pointer, - bootnodes: cstring, -) {.ffiRaw.} = + self: LogosDelivery, bootnodes: string +): Future[Result[string, string]] {.ffi.} = ## 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: + (await self.waku.discv5UpdateBootnodes(bootnodes)).isOkOr: error "UPDATE_DISCV5_BOOTSTRAP_NODES failed", error = error return err(error) return ok("discovery request processed correctly") proc waku_dns_discovery( - ctx: ptr FFIContext[LogosDelivery], - callback: FFICallBack, - userData: pointer, - enrTreeUrl: cstring, - nameDnsServer: cstring, - timeoutMs: cint, -) {.ffiRaw.} = - let nodes = ( - await ctx.myLib[].waku.dnsDiscovery($enrTreeUrl, $nameDnsServer, int(timeoutMs)) - ).valueOr: + self: LogosDelivery, enrTreeUrl: string, nameDnsServer: string, timeoutMs: int32 +): Future[Result[string, string]] {.ffi.} = + ## returns a comma-separated string of bootstrap nodes' multiaddresses + let nodes = (await self.waku.dnsDiscovery(enrTreeUrl, nameDnsServer, int(timeoutMs))).valueOr: error "GET_BOOTSTRAP_NODES failed", error = error return err(error) - ## returns a comma-separated string of bootstrap nodes' multiaddresses return ok(nodes.join(",")) -proc waku_start_discv5( - ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer -) {.ffiRaw.} = - (await ctx.myLib[].waku.startDiscv5()).isOkOr: +proc waku_start_discv5(self: LogosDelivery): Future[Result[string, string]] {.ffi.} = + (await self.waku.startDiscv5()).isOkOr: error "START_DISCV5 failed", error = error return err(error) return ok("discv5 started correctly") -proc waku_stop_discv5( - ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer -) {.ffiRaw.} = - (await ctx.myLib[].waku.stopDiscv5()).isOkOr: +proc waku_stop_discv5(self: LogosDelivery): Future[Result[string, string]] {.ffi.} = + (await self.waku.stopDiscv5()).isOkOr: error "STOP_DISCV5 failed", error = error return err(error) return ok("discv5 stopped correctly") proc waku_peer_exchange_request( - ctx: ptr FFIContext[LogosDelivery], - callback: FFICallBack, - userData: pointer, - numPeers: uint64, -) {.ffiRaw.} = - let numValidPeers = (await ctx.myLib[].waku.peerExchangeRequest(numPeers)).valueOr: + self: LogosDelivery, numPeers: uint64 +): Future[Result[string, string]] {.ffi.} = + let numValidPeers = (await self.waku.peerExchangeRequest(numPeers)).valueOr: error "waku_peer_exchange_request failed", error = error return err(error) return ok($numValidPeers) diff --git a/library/kernel_api/peer_manager_api.nim b/library/kernel_api/peer_manager_api.nim index 154a396ce..809ad22b9 100644 --- a/library/kernel_api/peer_manager_api.nim +++ b/library/kernel_api/peer_manager_api.nim @@ -7,74 +7,57 @@ type PeerInfo = object addresses: seq[string] proc waku_get_peerids_from_peerstore( - ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer -) {.ffiRaw.} = + self: LogosDelivery +): Future[Result[string, string]] {.ffi.} = ## returns a comma-separated string of peerIDs - let peerIds = (await ctx.myLib[].waku.peerIdsFromPeerstore()).valueOr: + let peerIds = (await self.waku.peerIdsFromPeerstore()).valueOr: return err(error) return ok(peerIds.join(",")) proc waku_connect( - ctx: ptr FFIContext[LogosDelivery], - callback: FFICallBack, - userData: pointer, - peerMultiAddr: cstring, - timeoutMs: cuint, -) {.ffiRaw.} = - let peers = ($peerMultiAddr).split(",") - (await ctx.myLib[].waku.connect(peers, uint32(timeoutMs))).isOkOr: + self: LogosDelivery, peerMultiAddr: string, timeoutMs: uint32 +): Future[Result[string, string]] {.ffi.} = + let peers = peerMultiAddr.split(",") + (await self.waku.connect(peers, timeoutMs)).isOkOr: return err(error) return ok("") proc waku_disconnect_peer_by_id( - ctx: ptr FFIContext[LogosDelivery], - callback: FFICallBack, - userData: pointer, - peerId: cstring, -) {.ffiRaw.} = - (await ctx.myLib[].waku.disconnectPeerById($peerId)).isOkOr: + self: LogosDelivery, peerId: string +): Future[Result[string, string]] {.ffi.} = + (await self.waku.disconnectPeerById(peerId)).isOkOr: error "DISCONNECT_PEER_BY_ID failed", error = error return err(error) return ok("") proc waku_disconnect_all_peers( - ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer -) {.ffiRaw.} = - (await ctx.myLib[].waku.disconnectAllPeers()).isOkOr: + self: LogosDelivery +): Future[Result[string, string]] {.ffi.} = + (await self.waku.disconnectAllPeers()).isOkOr: return err(error) return ok("") proc waku_dial_peer( - ctx: ptr FFIContext[LogosDelivery], - callback: FFICallBack, - userData: pointer, - peerMultiAddr: cstring, - protocol: cstring, - timeoutMs: cuint, -) {.ffiRaw.} = - (await ctx.myLib[].waku.dialPeer($peerMultiAddr, $protocol, int(timeoutMs))).isOkOr: + self: LogosDelivery, peerMultiAddr: string, protocol: string, timeoutMs: uint32 +): Future[Result[string, string]] {.ffi.} = + (await self.waku.dialPeer(peerMultiAddr, protocol, int(timeoutMs))).isOkOr: error "DIAL_PEER failed", error = error return err(error) return ok("") proc waku_dial_peer_by_id( - ctx: ptr FFIContext[LogosDelivery], - callback: FFICallBack, - userData: pointer, - peerId: cstring, - protocol: cstring, - timeoutMs: cuint, -) {.ffiRaw.} = - (await ctx.myLib[].waku.dialPeerById($peerId, $protocol, int(timeoutMs))).isOkOr: + self: LogosDelivery, peerId: string, protocol: string, timeoutMs: uint32 +): Future[Result[string, string]] {.ffi.} = + (await self.waku.dialPeerById(peerId, protocol, int(timeoutMs))).isOkOr: error "DIAL_PEER_BY_ID failed", error = error return err(error) return ok("") proc waku_get_connected_peers_info( - ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer -) {.ffiRaw.} = + self: LogosDelivery +): Future[Result[string, string]] {.ffi.} = ## returns a JSON string mapping peerIDs to objects with protocols and addresses - let peers = (await ctx.myLib[].waku.connectedPeersInfo()).valueOr: + let peers = (await self.waku.connectedPeersInfo()).valueOr: return err(error) var peersMap = initTable[string, PeerInfo]() @@ -85,20 +68,17 @@ proc waku_get_connected_peers_info( return ok($(%*peersMap)) proc waku_get_connected_peers( - ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer -) {.ffiRaw.} = + self: LogosDelivery +): Future[Result[string, string]] {.ffi.} = ## returns a comma-separated string of peerIDs - let peerIds = (await ctx.myLib[].waku.connectedPeers()).valueOr: + let peerIds = (await self.waku.connectedPeers()).valueOr: return err(error) return ok(peerIds.join(",")) proc waku_get_peerids_by_protocol( - ctx: ptr FFIContext[LogosDelivery], - callback: FFICallBack, - userData: pointer, - protocol: cstring, -) {.ffiRaw.} = + self: LogosDelivery, protocol: string +): Future[Result[string, string]] {.ffi.} = ## returns a comma-separated string of peerIDs that mount the given protocol - let peerIds = (await ctx.myLib[].waku.peerIdsByProtocol($protocol)).valueOr: + let peerIds = (await self.waku.peerIdsByProtocol(protocol)).valueOr: return err(error) return ok(peerIds.join(",")) diff --git a/library/kernel_api/ping_api.nim b/library/kernel_api/ping_api.nim index 3435cd3aa..855a8f8c5 100644 --- a/library/kernel_api/ping_api.nim +++ b/library/kernel_api/ping_api.nim @@ -2,12 +2,8 @@ import chronos, results, ffi import logos_delivery, library/declare_lib proc waku_ping_peer( - ctx: ptr FFIContext[LogosDelivery], - callback: FFICallBack, - userData: pointer, - peerAddr: cstring, - timeoutMs: cuint, -) {.ffiRaw.} = - let rttNanos = (await ctx.myLib[].waku.pingPeer($peerAddr, int(timeoutMs))).valueOr: + self: LogosDelivery, peerAddr: string, timeoutMs: uint32 +): Future[Result[string, string]] {.ffi.} = + let rttNanos = (await self.waku.pingPeer(peerAddr, int(timeoutMs))).valueOr: return err(error) return ok($rttNanos) diff --git a/library/kernel_api/protocols/filter_api.nim b/library/kernel_api/protocols/filter_api.nim index ea106ab36..d38c043b9 100644 --- a/library/kernel_api/protocols/filter_api.nim +++ b/library/kernel_api/protocols/filter_api.nim @@ -10,22 +10,18 @@ import library/declare_lib proc waku_filter_subscribe( - ctx: ptr FFIContext[LogosDelivery], - callback: FFICallBack, - userData: pointer, - pubSubTopic: cstring, - contentTopics: cstring, -) {.ffiRaw.} = - proc onReceivedMessage(ctx: ptr FFIContext[LogosDelivery]): FilterPushHandler = + self: LogosDelivery, pubSubTopic: string, contentTopics: string +): Future[Result[string, string]] {.ffi.} = + proc onReceivedMessage(): FilterPushHandler = return proc(pubsubTopic: PubsubTopic, msg: WakuMessage) {.async.} = emitEvent("onReceivedMessage"): $JsonMessageEvent.new(pubsubTopic, msg) ( - await ctx.myLib[].waku.filterSubscribe( - PubsubTopic($pubSubTopic), - ($contentTopics).split(",").mapIt(ContentTopic(it)), - FilterPushHandler(onReceivedMessage(ctx)), + await self.waku.filterSubscribe( + PubsubTopic(pubSubTopic), + contentTopics.split(",").mapIt(ContentTopic(it)), + FilterPushHandler(onReceivedMessage()), ) ).isOkOr: error "fail filter subscribe", error = error @@ -33,15 +29,11 @@ proc waku_filter_subscribe( return ok("") proc waku_filter_unsubscribe( - ctx: ptr FFIContext[LogosDelivery], - callback: FFICallBack, - userData: pointer, - pubSubTopic: cstring, - contentTopics: cstring, -) {.ffiRaw.} = + self: LogosDelivery, pubSubTopic: string, contentTopics: string +): Future[Result[string, string]] {.ffi.} = ( - await ctx.myLib[].waku.filterUnsubscribe( - PubsubTopic($pubSubTopic), ($contentTopics).split(",").mapIt(ContentTopic(it)) + await self.waku.filterUnsubscribe( + PubsubTopic(pubSubTopic), contentTopics.split(",").mapIt(ContentTopic(it)) ) ).isOkOr: error "fail filter unsubscribe", error = error @@ -49,9 +41,9 @@ proc waku_filter_unsubscribe( return ok("") proc waku_filter_unsubscribe_all( - ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer -) {.ffiRaw.} = - (await ctx.myLib[].waku.filterUnsubscribeAll()).isOkOr: + self: LogosDelivery +): Future[Result[string, string]] {.ffi.} = + (await self.waku.filterUnsubscribeAll()).isOkOr: error "fail filter unsubscribe all", error = error return err(error) return ok("") diff --git a/library/kernel_api/protocols/lightpush_api.nim b/library/kernel_api/protocols/lightpush_api.nim index c2fe09abe..fd3c749fa 100644 --- a/library/kernel_api/protocols/lightpush_api.nim +++ b/library/kernel_api/protocols/lightpush_api.nim @@ -8,26 +8,20 @@ import library/declare_lib proc waku_lightpush_publish( - ctx: ptr FFIContext[LogosDelivery], - callback: FFICallBack, - userData: pointer, - pubSubTopic: cstring, - jsonWakuMessage: cstring, -) {.ffiRaw.} = + self: LogosDelivery, pubSubTopic: string, jsonWakuMessage: string +): Future[Result[string, string]] {.ffi.} = var jsonMessage: JsonMessage try: - let jsonContent = parseJson($jsonWakuMessage) + let jsonContent = parseJson(jsonWakuMessage) jsonMessage = JsonMessage.fromJsonNode(jsonContent).valueOr: raise newException(JsonParsingError, $error) - except JsonParsingError as exc: - return err(fmt"Error parsing json message: {exc.msg}") + except JsonParsingError as e: + return err(fmt"Error parsing json message: {e.msg}") let msg = json_message_event.toWakuMessage(jsonMessage).valueOr: return err("Problem building the WakuMessage: " & $error) - let msgHashHex = ( - await ctx.myLib[].waku.lightpushPublish(PubsubTopic($pubSubTopic), msg) - ).valueOr: + let msgHashHex = (await self.waku.lightpushPublish(PubsubTopic(pubSubTopic), msg)).valueOr: error "PUBLISH failed", error = error return err(error) diff --git a/library/kernel_api/protocols/relay_api.nim b/library/kernel_api/protocols/relay_api.nim index f01652820..5c9055767 100644 --- a/library/kernel_api/protocols/relay_api.nim +++ b/library/kernel_api/protocols/relay_api.nim @@ -9,82 +9,58 @@ import library/declare_lib proc waku_relay_get_peers_in_mesh( - ctx: ptr FFIContext[LogosDelivery], - callback: FFICallBack, - userData: pointer, - pubSubTopic: cstring, -) {.ffiRaw.} = - let peers = (await ctx.myLib[].waku.relayPeersInMesh(PubsubTopic($pubSubTopic))).valueOr: + self: LogosDelivery, pubSubTopic: string +): Future[Result[string, string]] {.ffi.} = + ## returns a comma-separated string of peerIDs + let peers = (await self.waku.relayPeersInMesh(PubsubTopic(pubSubTopic))).valueOr: error "LIST_MESH_PEERS failed", error = error return err(error) - ## returns a comma-separated string of peerIDs return ok(peers.join(",")) proc waku_relay_get_num_peers_in_mesh( - ctx: ptr FFIContext[LogosDelivery], - callback: FFICallBack, - userData: pointer, - pubSubTopic: cstring, -) {.ffiRaw.} = - let n = (await ctx.myLib[].waku.relayNumPeersInMesh(PubsubTopic($pubSubTopic))).valueOr: + self: LogosDelivery, pubSubTopic: string +): Future[Result[string, string]] {.ffi.} = + let n = (await self.waku.relayNumPeersInMesh(PubsubTopic(pubSubTopic))).valueOr: error "NUM_MESH_PEERS failed", error = error return err(error) return ok($n) proc waku_relay_get_connected_peers( - ctx: ptr FFIContext[LogosDelivery], - callback: FFICallBack, - userData: pointer, - pubSubTopic: cstring, -) {.ffiRaw.} = + self: LogosDelivery, pubSubTopic: string +): Future[Result[string, string]] {.ffi.} = ## Returns the list of all connected peers to an specific pubsub topic - let peers = (await ctx.myLib[].waku.relayConnectedPeers(PubsubTopic($pubSubTopic))).valueOr: + let peers = (await self.waku.relayConnectedPeers(PubsubTopic(pubSubTopic))).valueOr: error "LIST_CONNECTED_PEERS failed", error = error return err(error) return ok(peers.join(",")) proc waku_relay_get_num_connected_peers( - ctx: ptr FFIContext[LogosDelivery], - callback: FFICallBack, - userData: pointer, - pubSubTopic: cstring, -) {.ffiRaw.} = - let n = (await ctx.myLib[].waku.relayNumConnectedPeers(PubsubTopic($pubSubTopic))).valueOr: + self: LogosDelivery, pubSubTopic: string +): Future[Result[string, string]] {.ffi.} = + let n = (await self.waku.relayNumConnectedPeers(PubsubTopic(pubSubTopic))).valueOr: error "NUM_CONNECTED_PEERS failed", error = error return err(error) return ok($n) proc waku_relay_add_protected_shard( - ctx: ptr FFIContext[LogosDelivery], - callback: FFICallBack, - userData: pointer, - clusterId: cint, - shardId: cint, - publicKey: cstring, -) {.ffiRaw.} = + self: LogosDelivery, clusterId: uint16, shardId: uint16, publicKey: string +): Future[Result[string, string]] {.ffi.} = ## Protects a shard with a public key - ( - await ctx.myLib[].waku.relayAddProtectedShard( - uint16(clusterId), uint16(shardId), $publicKey - ) - ).isOkOr: + (await self.waku.relayAddProtectedShard(clusterId, shardId, publicKey)).isOkOr: return err(error) return ok("") proc waku_relay_subscribe( - ctx: ptr FFIContext[LogosDelivery], - callback: FFICallBack, - userData: pointer, - pubSubTopic: cstring, -) {.ffiRaw.} = - proc onReceivedMessage(ctx: ptr FFIContext[LogosDelivery]): WakuRelayHandler = + self: LogosDelivery, pubSubTopic: string +): Future[Result[string, string]] {.ffi.} = + proc onReceivedMessage(): WakuRelayHandler = return proc(pubsubTopic: PubsubTopic, msg: WakuMessage) {.async.} = emitEvent("onReceivedMessage"): $JsonMessageEvent.new(pubsubTopic, msg) ( - await ctx.myLib[].waku.relaySubscribe( - PubsubTopic($pubSubTopic), WakuRelayHandler(onReceivedMessage(ctx)) + await self.waku.relaySubscribe( + PubsubTopic(pubSubTopic), WakuRelayHandler(onReceivedMessage()) ) ).isOkOr: error "SUBSCRIBE failed", error = error @@ -92,74 +68,55 @@ proc waku_relay_subscribe( return ok("") proc waku_relay_unsubscribe( - ctx: ptr FFIContext[LogosDelivery], - callback: FFICallBack, - userData: pointer, - pubSubTopic: cstring, -) {.ffiRaw.} = - (await ctx.myLib[].waku.relayUnsubscribe(PubsubTopic($pubSubTopic))).isOkOr: + self: LogosDelivery, pubSubTopic: string +): Future[Result[string, string]] {.ffi.} = + (await self.waku.relayUnsubscribe(PubsubTopic(pubSubTopic))).isOkOr: error "UNSUBSCRIBE failed", error = error return err(error) return ok("") proc waku_relay_publish( - ctx: ptr FFIContext[LogosDelivery], - callback: FFICallBack, - userData: pointer, - pubSubTopic: cstring, - jsonWakuMessage: cstring, - timeoutMs: cuint, -) {.ffiRaw.} = + self: LogosDelivery, pubSubTopic: string, jsonWakuMessage: string, timeoutMs: uint32 +): Future[Result[string, string]] {.ffi.} = var jsonMessage: JsonMessage try: - let jsonContent = parseJson($jsonWakuMessage) + let jsonContent = parseJson(jsonWakuMessage) jsonMessage = JsonMessage.fromJsonNode(jsonContent).valueOr: raise newException(JsonParsingError, $error) - except JsonParsingError as exc: - return err("Error parsing json message: " & exc.msg) + except JsonParsingError as e: + return err("Error parsing json message: " & e.msg) let msg = json_message_event.toWakuMessage(jsonMessage).valueOr: return err("Problem building the WakuMessage: " & $error) - let msgHash = ( - await ctx.myLib[].waku.relayPublish( - PubsubTopic($pubSubTopic), msg, uint32(timeoutMs) - ) - ).valueOr: + let msgHash = (await self.waku.relayPublish(PubsubTopic(pubSubTopic), msg, timeoutMs)).valueOr: error "PUBLISH failed", error = error return err(error) return ok(msgHash) proc waku_default_pubsub_topic( - ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer -) {.ffiRaw.} = - let topic = (await ctx.myLib[].waku.defaultPubsubTopic()).valueOr: + self: LogosDelivery +): Future[Result[string, string]] {.ffi.} = + let topic = (await self.waku.defaultPubsubTopic()).valueOr: return err(error) return ok(string(topic)) proc waku_content_topic( - ctx: ptr FFIContext[LogosDelivery], - callback: FFICallBack, - userData: pointer, - appName: cstring, - appVersion: cuint, - contentTopicName: cstring, - encoding: cstring, -) {.ffiRaw.} = + self: LogosDelivery, + appName: string, + appVersion: uint32, + contentTopicName: string, + encoding: string, +): Future[Result[string, string]] {.ffi.} = let topic = ( - await ctx.myLib[].waku.buildContentTopic( - $appName, uint32(appVersion), $contentTopicName, $encoding - ) + await self.waku.buildContentTopic(appName, appVersion, contentTopicName, encoding) ).valueOr: return err(error) return ok(string(topic)) proc waku_pubsub_topic( - ctx: ptr FFIContext[LogosDelivery], - callback: FFICallBack, - userData: pointer, - topicName: cstring, -) {.ffiRaw.} = - let topic = (await ctx.myLib[].waku.buildPubsubTopic($topicName)).valueOr: + self: LogosDelivery, topicName: string +): Future[Result[string, string]] {.ffi.} = + let topic = (await self.waku.buildPubsubTopic(topicName)).valueOr: return err(error) return ok(string(topic)) diff --git a/library/kernel_api/protocols/store_api.nim b/library/kernel_api/protocols/store_api.nim index 4d6accadd..d9245001e 100644 --- a/library/kernel_api/protocols/store_api.nim +++ b/library/kernel_api/protocols/store_api.nim @@ -65,15 +65,10 @@ func fromJsonNode(jsonContent: JsonNode): Result[StoreQueryRequest, string] = ) proc waku_store_query( - ctx: ptr FFIContext[LogosDelivery], - callback: FFICallBack, - userData: pointer, - jsonQuery: cstring, - peerAddr: cstring, - timeoutMs: cint, -) {.ffiRaw.} = + self: LogosDelivery, jsonQuery: string, peerAddr: string, timeoutMs: int32 +): Future[Result[string, string]] {.ffi.} = let jsonContentRes = catch: - parseJson($jsonQuery) + parseJson(jsonQuery) if jsonContentRes.isErr(): return err("StoreRequest failed parsing store request: " & jsonContentRes.error.msg) @@ -81,7 +76,7 @@ proc waku_store_query( let storeQueryRequest = ?fromJsonNode(jsonContentRes.get()) let queryResponse = ( - await ctx.myLib[].waku.storeQuery(storeQueryRequest, $peerAddr, int(timeoutMs)) + await self.waku.storeQuery(storeQueryRequest, peerAddr, int(timeoutMs)) ).valueOr: return err("StoreRequest failed store query: " & error) diff --git a/library/liblogosdelivery.h b/library/liblogosdelivery.h index 154b88774..d70172529 100644 --- a/library/liblogosdelivery.h +++ b/library/liblogosdelivery.h @@ -1,6 +1,10 @@ - -// Generated manually and inspired by libwaku.h -// Header file for Logos Messaging API (LMAPI) library +// Public C header for the Logos Messaging API (LMAPI) library. +// +// The call surface is generated from the {.ffi.} annotations in library/*.nim +// and written to generated/logosdelivery.h by `make liblogosdelivery`. That file +// is a build artifact, not checked in, so build the library before you compile +// against this header. This file adds the event-listener ABI, which nim-ffi +// exports from declareLibrary but does not emit into the `abi = c` header. #pragma once #ifndef __liblogosdelivery__ #define __liblogosdelivery__ @@ -8,145 +12,50 @@ #include #include -// The possible returned values for the functions that return int -#define RET_OK 0 -#define RET_ERR 1 -#define RET_MISSING_CALLBACK 2 +#include "generated/logosdelivery.h" + +// Kept as aliases of the generated NIMFFI_RET_* codes so existing callers that +// use the short names keep compiling. Guarded because the legacy libwaku header +// defines the same names with the same values. +#ifndef RET_OK +#define RET_OK NIMFFI_RET_OK +#endif +#ifndef RET_ERR +#define RET_ERR NIMFFI_RET_ERR +#endif +#ifndef RET_MISSING_CALLBACK +#define RET_MISSING_CALLBACK NIMFFI_RET_MISSING_CALLBACK +#endif #ifdef __cplusplus extern "C" { #endif + // Raw result-delivery callback used by the event API. `msg` is a byte run of + // `len` bytes, not NUL-terminated, and is valid only for the duration of the + // call. typedef void (*FFICallBack)(int callerRet, const char *msg, size_t len, void *userData); - // Creates a new instance of the node from the given configuration JSON. - // Returns a pointer to the Context needed by the rest of the API functions. - // The configuration is a JSON object with these optional keys: - // "mode": "Core" | "Edge" (messaging role; defaults to "Core") - // "preset": "" (e.g. "twn") - // "messagingOverrides": { ... } (per-field messaging config overrides) - // "channelsOverrides": { ... } (per-field reliable-channel overrides) - // Override keys accept the config field name or its CLI switch name (e.g. - // "clusterId" or "cluster-id"). Unknown keys are rejected. - // Example: {"mode":"Core","messagingOverrides":{"cluster-id":42,"log-level":"INFO"}} - void *logosdelivery_create_node( - const char *configJson, - FFICallBack callback, - void *userData); - - // Starts the node. - int logosdelivery_start_node(void *ctx, - FFICallBack callback, - void *userData); - - // Stops the node. - int logosdelivery_stop_node(void *ctx, - FFICallBack callback, - void *userData); - - // Destroys an instance of a node created with logosdelivery_create_node - int logosdelivery_destroy(void *ctx, - FFICallBack callback, - void *userData); - - // Subscribe to a content topic. - // contentTopic: string representing the content topic (e.g., "/myapp/1/chat/proto") - int logosdelivery_subscribe(void *ctx, - FFICallBack callback, - void *userData, - const char *contentTopic); - - // Unsubscribe from a content topic. - int logosdelivery_unsubscribe(void *ctx, - FFICallBack callback, - void *userData, - const char *contentTopic); - - // Send a message. - // messageJson: JSON string with the following structure: - // { - // "contentTopic": "/myapp/1/chat/proto", - // "payload": "base64-encoded-payload", - // "ephemeral": false - // } - // Returns a request ID that can be used to track the message delivery. - int logosdelivery_send(void *ctx, - FFICallBack callback, - void *userData, - const char *messageJson); - - // --- Reliable Channels API (stable surface) --- - - // Create a reliable channel. Returns the channel id. - int logosdelivery_channel_create(void *ctx, - FFICallBack callback, - void *userData, - const char *channelId, - const char *contentTopic, - const char *senderId); - - // Check whether a reliable channel is currently open. Returns "true" or - // "false"; an unknown channel id is not an error. - int logosdelivery_channel_exists(void *ctx, - FFICallBack callback, - void *userData, - const char *channelId); - - // Send a message on a reliable channel. - // messageJson: { "payload": "base64-encoded-payload", "ephemeral": false } - // Returns a request ID that can be used to track delivery. - int logosdelivery_channel_send(void *ctx, - FFICallBack callback, - void *userData, - const char *channelId, - const char *messageJson); - - // Close a reliable channel: stops its SDS loops; persisted state survives, so - // re-creating the channel restores it. - int logosdelivery_channel_close(void *ctx, - FFICallBack callback, - void *userData, - const char *channelId); - - // Channel lifecycle events are delivered through a per-event listener, - // registered by name: "onChannelMessageReceived" (payload base64-encoded), - // "onChannelMessageSent", "onChannelMessageError". + // Events are delivered through a per-event listener registry. Register one + // callback per event name of interest; see the README for the full list. + // Channel lifecycle events are "onChannelMessageReceived" (payload + // base64-encoded), "onChannelMessageSent" and "onChannelMessageError". // 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. + // (0 on an invalid context). `ctx` is the context handle returned by + // logosdelivery_create_node. // 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); + 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, - void *userData); - - // Given a node info ID, retrieves the corresponding info. - int logosdelivery_get_node_info(void *ctx, - FFICallBack callback, - void *userData, - const char *nodeInfoId); - - // Retrieves the list of available configurations. - int logosdelivery_get_available_configs(void *ctx, - FFICallBack callback, - void *userData); - - // NOTE: the low-level kernel API (waku_*) lives in the separate, advanced - // header liblogosdelivery_kernel.h. It is intentionally not declared here so - // this header only promises the stable Messaging / Reliable Channels surface. + uint64_t listenerId); #ifdef __cplusplus } diff --git a/library/liblogosdelivery.nim b/library/liblogosdelivery.nim index 57279c189..853311aef 100644 --- a/library/liblogosdelivery.nim +++ b/library/liblogosdelivery.nim @@ -32,3 +32,6 @@ include # logosdelivery_* surface in ./logos_delivery_api/node_api. The former # waku_new / waku_start / waku_stop / waku_destroy entry points were removed to # avoid maintaining two parallel node-lifecycle APIs. + +# Emits the `abi = c` dispatch wrappers, so it must stay the last FFI call here. +genBindings() diff --git a/library/liblogosdelivery_kernel.h b/library/liblogosdelivery_kernel.h index 95b2982e7..577fa21ff 100644 --- a/library/liblogosdelivery_kernel.h +++ b/library/liblogosdelivery_kernel.h @@ -1,16 +1,17 @@ - -// liblogosdelivery_kernel.h — Kernel / advanced API (low-level, per-protocol). +// liblogosdelivery_kernel.h — compatibility alias for liblogosdelivery.h. // -// ⚠️ USE AT YOUR OWN RISK — UNSUPPORTED, UNSTABLE SURFACE. +// This header used to declare the low-level `waku_*` kernel API separately from +// the stable messaging surface, so that including it was a deliberate opt-in. +// That split no longer exists: the call surface is generated as one header, and +// liblogosdelivery.h declares every entry point. This file is kept only so +// existing includes keep resolving. // -// These `waku_*` functions are the low-level kernel API. They are NOT part of -// the stable, supported Messaging / Reliable Channels surface declared in -// liblogosdelivery.h. They expose per-protocol internals (relay, filter, -// lightpush, store, discovery, peer management) and may change or be removed -// at ANY time, without notice or a deprecation cycle. -// -// Including this header is a deliberate opt-in into the advanced tier. If you -// only need messaging, include liblogosdelivery.h and nothing here. +// The tiering still holds as a support promise, even though the compiler no +// longer enforces it. The `waku_*` functions expose per-protocol internals +// (relay, filter, lightpush, store, discovery, peer management) and may change +// or be removed at ANY time, without notice or a deprecation cycle. Only the +// messaging and reliable-channel entry points are +// supported. // // See https://github.com/logos-messaging/logos-delivery/issues/3851 for the // tiering rationale. @@ -18,224 +19,6 @@ #ifndef __liblogosdelivery_kernel__ #define __liblogosdelivery_kernel__ -// Shared FFICallBack typedef and RET_* return codes live in the stable header. #include "liblogosdelivery.h" -#ifdef __cplusplus -extern "C" -{ -#endif - - // NOTE: node lifecycle (create / start / stop / destroy) is unified and lives - // only in the stable header. Use logosdelivery_create_node, - // logosdelivery_start_node, logosdelivery_stop_node and logosdelivery_destroy - // (declared in liblogosdelivery.h, included above) regardless of whether you - // drive the node through the messaging surface or this kernel API. - - int waku_version(void *ctx, - FFICallBack callback, - void *userData); - - // NOTE: event callbacks are registered via logosdelivery_add_event_listener - // (declared above) which the waku_* API shares. - - int waku_content_topic(void *ctx, - FFICallBack callback, - void *userData, - const char *appName, - unsigned int appVersion, - const char *contentTopicName, - const char *encoding); - - int waku_pubsub_topic(void *ctx, - FFICallBack callback, - void *userData, - const char *topicName); - - int waku_default_pubsub_topic(void *ctx, - FFICallBack callback, - void *userData); - - int waku_relay_publish(void *ctx, - FFICallBack callback, - void *userData, - const char *pubSubTopic, - const char *jsonWakuMessage, - unsigned int timeoutMs); - - int waku_lightpush_publish(void *ctx, - FFICallBack callback, - void *userData, - const char *pubSubTopic, - const char *jsonWakuMessage); - - int waku_relay_subscribe(void *ctx, - FFICallBack callback, - void *userData, - const char *pubSubTopic); - - int waku_relay_add_protected_shard(void *ctx, - FFICallBack callback, - void *userData, - int clusterId, - int shardId, - char *publicKey); - - int waku_relay_unsubscribe(void *ctx, - FFICallBack callback, - void *userData, - const char *pubSubTopic); - - int waku_filter_subscribe(void *ctx, - FFICallBack callback, - void *userData, - const char *pubSubTopic, - const char *contentTopics); - - int waku_filter_unsubscribe(void *ctx, - FFICallBack callback, - void *userData, - const char *pubSubTopic, - const char *contentTopics); - - int waku_filter_unsubscribe_all(void *ctx, - FFICallBack callback, - void *userData); - - int waku_relay_get_num_connected_peers(void *ctx, - FFICallBack callback, - void *userData, - const char *pubSubTopic); - - int waku_relay_get_connected_peers(void *ctx, - FFICallBack callback, - void *userData, - const char *pubSubTopic); - - int waku_relay_get_num_peers_in_mesh(void *ctx, - FFICallBack callback, - void *userData, - const char *pubSubTopic); - - int waku_relay_get_peers_in_mesh(void *ctx, - FFICallBack callback, - void *userData, - const char *pubSubTopic); - - int waku_store_query(void *ctx, - FFICallBack callback, - void *userData, - const char *jsonQuery, - const char *peerAddr, - int timeoutMs); - - int waku_connect(void *ctx, - FFICallBack callback, - void *userData, - const char *peerMultiAddr, - unsigned int timeoutMs); - - int waku_disconnect_peer_by_id(void *ctx, - FFICallBack callback, - void *userData, - const char *peerId); - - int waku_disconnect_all_peers(void *ctx, - FFICallBack callback, - void *userData); - - int waku_dial_peer(void *ctx, - FFICallBack callback, - void *userData, - const char *peerMultiAddr, - const char *protocol, - int timeoutMs); - - int waku_dial_peer_by_id(void *ctx, - FFICallBack callback, - void *userData, - const char *peerId, - const char *protocol, - int timeoutMs); - - int waku_get_peerids_from_peerstore(void *ctx, - FFICallBack callback, - void *userData); - - int waku_get_connected_peers_info(void *ctx, - FFICallBack callback, - void *userData); - - int waku_get_peerids_by_protocol(void *ctx, - FFICallBack callback, - void *userData, - const char *protocol); - - int waku_listen_addresses(void *ctx, - FFICallBack callback, - void *userData); - - int waku_get_connected_peers(void *ctx, - FFICallBack callback, - void *userData); - - // Returns a list of multiaddress given a url to a DNS discoverable ENR tree - // Parameters - // char* entTreeUrl: URL containing a discoverable ENR tree - // char* nameDnsServer: The nameserver to resolve the ENR tree url. - // int timeoutMs: Timeout value in milliseconds to execute the call. - int waku_dns_discovery(void *ctx, - FFICallBack callback, - void *userData, - const char *entTreeUrl, - const char *nameDnsServer, - int timeoutMs); - - // Updates the bootnode list used for discovering new peers via DiscoveryV5 - // bootnodes - JSON array containing the bootnode ENRs i.e. `["enr:...", "enr:..."]` - int waku_discv5_update_bootnodes(void *ctx, - FFICallBack callback, - void *userData, - char *bootnodes); - - int waku_start_discv5(void *ctx, - FFICallBack callback, - void *userData); - - int waku_stop_discv5(void *ctx, - FFICallBack callback, - void *userData); - - // Retrieves the ENR information - int waku_get_my_enr(void *ctx, - FFICallBack callback, - void *userData); - - int waku_get_my_peerid(void *ctx, - FFICallBack callback, - void *userData); - - int waku_get_metrics(void *ctx, - FFICallBack callback, - void *userData); - - int waku_peer_exchange_request(void *ctx, - FFICallBack callback, - void *userData, - int numPeers); - - int waku_ping_peer(void *ctx, - FFICallBack callback, - void *userData, - const char *peerAddr, - int timeoutMs); - - int waku_is_online(void *ctx, - FFICallBack callback, - void *userData); - -#ifdef __cplusplus -} -#endif - #endif /* __liblogosdelivery_kernel__ */ diff --git a/library/logos_delivery_api/debug_api.nim b/library/logos_delivery_api/debug_api.nim index 75c6b921b..26c7e99b6 100644 --- a/library/logos_delivery_api/debug_api.nim +++ b/library/logos_delivery_api/debug_api.nim @@ -3,46 +3,37 @@ import logos_delivery/waku/factory/waku_state_info import tools/confutils/[cli_args, config_option_meta] proc logosdelivery_get_available_node_info_ids( - ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer -) {.ffiRaw.} = - ## Returns the list of all available node info item ids that - ## can be queried with `get_node_info_item`. - requireInitializedNode(ctx, "GetNodeInfoIds"): - return err(errMsg) + self: LogosDelivery +): Future[Result[string, string]] {.ffi.} = + ## Returns, as a JSON array of strings, all available node info item ids that + ## can be queried with `get_node_info`. + var ids = newJArray() + for id in self.waku.stateInfo.getAllPossibleInfoItemIds(): + ids.add(%($id)) - return ok($ctx.myLib[].waku.stateInfo.getAllPossibleInfoItemIds()) + return ok($ids) proc logosdelivery_get_node_info( - ctx: ptr FFIContext[LogosDelivery], - callback: FFICallBack, - userData: pointer, - nodeInfoId: cstring, -) {.ffiRaw.} = + self: LogosDelivery, nodeInfoId: string +): Future[Result[string, string]] {.ffi.} = ## Returns the content of the node info item with the given id if it exists. - requireInitializedNode(ctx, "GetNodeInfoItem"): - return err(errMsg) - + ## The content is a plain string, not JSON: a peer id, an ENR URI, a + ## comma-separated multiaddress list or the Prometheus metrics text. let infoItemIdEnum = try: - parseEnum[NodeInfoId]($nodeInfoId) + parseEnum[NodeInfoId](nodeInfoId) except ValueError: - return err("Invalid node info id: " & $nodeInfoId) + return err("Invalid node info id: " & nodeInfoId) - return ok(ctx.myLib[].waku.stateInfo.getNodeInfoItem(infoItemIdEnum)) + return ok(self.waku.stateInfo.getNodeInfoItem(infoItemIdEnum)) proc logosdelivery_get_available_configs( - ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer -) {.ffiRaw.} = + self: LogosDelivery +): Future[Result[string, string]] {.ffi.} = ## Returns information about the accepted config items. - requireInitializedNode(ctx, "GetAvailableConfigs"): - return err(errMsg) - let optionMetas: seq[ConfigOptionMeta] = extractConfigOptionMeta(WakuNodeConf) var configOptionDetails = newJArray() - # for confField, confValue in fieldPairs(conf): - # defaultConfig[confField] = $confValue - for meta in optionMetas: configOptionDetails.add( %*{ @@ -52,5 +43,4 @@ proc logosdelivery_get_available_configs( var jsonNode = newJObject() jsonNode["configOptions"] = configOptionDetails - let asString = pretty(jsonNode) return ok(pretty(jsonNode)) diff --git a/library/logos_delivery_api/messaging_api.nim b/library/logos_delivery_api/messaging_api.nim index e0601340b..eea4b10b5 100644 --- a/library/logos_delivery_api/messaging_api.nim +++ b/library/logos_delivery_api/messaging_api.nim @@ -1,6 +1,5 @@ import std/[json] import chronos, results, ffi -import stew/byteutils import logos_delivery/waku/common/base64, logos_delivery/waku/waku, @@ -9,63 +8,45 @@ import ../declare_lib proc logosdelivery_subscribe( - ctx: ptr FFIContext[LogosDelivery], - callback: FFICallBack, - userData: pointer, - contentTopicStr: cstring, -) {.ffiRaw.} = - requireInitializedNode(ctx, "Subscribe"): - return err(errMsg) - - requireMessaging(ctx, "Subscribe"): + self: LogosDelivery, contentTopicStr: string +): Future[Result[string, string]] {.ffi.} = + requireMessaging(self, "Subscribe"): return err(errMsg) # ContentTopic is just a string type alias - let contentTopic = ContentTopic($contentTopicStr) + let contentTopic = ContentTopic(contentTopicStr) - (await ctx.myLib[].messagingClient.subscribe(contentTopic)).isOkOr: + (await self.messagingClient.subscribe(contentTopic)).isOkOr: let errMsg = $error return err("Subscribe failed: " & errMsg) return ok("") proc logosdelivery_unsubscribe( - ctx: ptr FFIContext[LogosDelivery], - callback: FFICallBack, - userData: pointer, - contentTopicStr: cstring, -) {.ffiRaw.} = - requireInitializedNode(ctx, "Unsubscribe"): - return err(errMsg) - - requireMessaging(ctx, "Unsubscribe"): + self: LogosDelivery, contentTopicStr: string +): Future[Result[string, string]] {.ffi.} = + requireMessaging(self, "Unsubscribe"): return err(errMsg) # ContentTopic is just a string type alias - let contentTopic = ContentTopic($contentTopicStr) + let contentTopic = ContentTopic(contentTopicStr) - ctx.myLib[].messagingClient.unsubscribe(contentTopic).isOkOr: + self.messagingClient.unsubscribe(contentTopic).isOkOr: let errMsg = $error return err("Unsubscribe failed: " & errMsg) return ok("") proc logosdelivery_send( - ctx: ptr FFIContext[LogosDelivery], - callback: FFICallBack, - userData: pointer, - messageJson: cstring, -) {.ffiRaw.} = - requireInitializedNode(ctx, "Send"): - return err(errMsg) - - requireMessaging(ctx, "Send"): + self: LogosDelivery, messageJson: string +): Future[Result[string, string]] {.ffi.} = + requireMessaging(self, "Send"): return err(errMsg) ## Parse the message JSON and send the message var jsonNode: JsonNode try: - jsonNode = parseJson($messageJson) + jsonNode = parseJson(messageJson) except Exception as e: return err("Failed to parse message JSON: " & e.msg) @@ -93,7 +74,7 @@ proc logosdelivery_send( ) # Send the message via the messaging layer's own API. - let requestId = (await ctx.myLib[].messagingClient.send(envelope)).valueOr: + let requestId = (await self.messagingClient.send(envelope)).valueOr: let errMsg = $error return err("Send failed: " & errMsg) diff --git a/library/logos_delivery_api/node_api.nim b/library/logos_delivery_api/node_api.nim index b687471f6..277b0a556 100644 --- a/library/logos_delivery_api/node_api.nim +++ b/library/logos_delivery_api/node_api.nim @@ -16,123 +16,74 @@ import proc `%`*(id: RequestId): JsonNode = %($id) -registerReqFFI(CreateNodeRequest, ctx: ptr FFIContext[LogosDelivery]): - proc(configJson: cstring): Future[Result[string, string]] {.async.} = - let conf = parseLogosDeliveryConf($configJson).valueOr: - error "Failed to parse Logos Delivery configuration JSON", - error = error, configJson = $configJson - return err("failed parseLogosDeliveryConf " & error) - - ctx.myLib[] = (await LogosDelivery.new(conf)).valueOr: - let errMsg = $error - chronicles.error "CreateNodeRequest failed", err = errMsg - return err(errMsg) - - return ok("") - -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) - - # 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 - - ## always need to invoke the callback although we don't retrieve value to the caller - callback(RET_OK, nil, 0, userData) - - return RET_OK - proc logosdelivery_create_node( - configJson: cstring, callback: FFICallback, userData: pointer -): pointer {.dynlib, exportc, cdecl.} = - initializeLibrary() + configJson: string +): Future[Result[LogosDelivery, string]] {.ffiCtor.} = + let conf = parseLogosDeliveryConf(configJson).valueOr: + error "Failed to parse Logos Delivery configuration JSON", + error = error, configJson = configJson + return err("failed parseLogosDeliveryConf " & error) - if callback.isNil(): - echo "error: missing callback in logosdelivery_create_node" - return nil - - 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 - - ctx.userData = userData - - ffi.sendRequestToFFIThread( - ctx, CreateNodeRequest.ffiNewReq(callback, userData, configJson) - ).isOkOr: - 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.recycleFFIContext(LogosDeliveryFFIPool, ctx).isOkOr: - chronicles.error "Error in recycleFFIContext after sendRequestToFFIThread during creation", - err = $error - return nil - - return ctx - -proc logosdelivery_start_node( - ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer -) {.ffiRaw.} = - requireInitializedNode(ctx, "START_NODE"): + let lib = (await LogosDelivery.new(conf)).valueOr: + let errMsg = $error + chronicles.error "CreateNodeRequest failed", err = errMsg return err(errMsg) - # setting up outgoing event listeners - let sentListener = MessageSentEvent.listen( - ctx.myLib[].waku.brokerCtx, + return ok(lib) + +proc logosdelivery_destroy(self: LogosDelivery) {.ffiDtor.} = + discard + +proc registerFFIEventListeners(self: LogosDelivery): Result[void, string] = + ## Bridges every broker event the library re-publishes onto the FFI event + ## registry. Keep in step with `dropFFIEventListeners`. + MessageSentEvent.listen( + self.waku.brokerCtx, proc(event: MessageSentEvent) {.async: (raises: []).} = emitEvent("onMessageSent"): $newJsonEvent("message_sent", event), - ).valueOr: + ).isOkOr: chronicles.error "MessageSentEvent.listen failed", err = $error return err("MessageSentEvent.listen failed: " & $error) - let errorListener = MessageErrorEvent.listen( - ctx.myLib[].waku.brokerCtx, + MessageErrorEvent.listen( + self.waku.brokerCtx, proc(event: MessageErrorEvent) {.async: (raises: []).} = emitEvent("onMessageError"): $newJsonEvent("message_error", event), - ).valueOr: + ).isOkOr: chronicles.error "MessageErrorEvent.listen failed", err = $error return err("MessageErrorEvent.listen failed: " & $error) - let propagatedListener = MessagePropagatedEvent.listen( - ctx.myLib[].waku.brokerCtx, + MessagePropagatedEvent.listen( + self.waku.brokerCtx, proc(event: MessagePropagatedEvent) {.async: (raises: []).} = emitEvent("onMessagePropagated"): $newJsonEvent("message_propagated", event), - ).valueOr: + ).isOkOr: chronicles.error "MessagePropagatedEvent.listen failed", err = $error return err("MessagePropagatedEvent.listen failed: " & $error) - let receivedListener = MessageReceivedEvent.listen( - ctx.myLib[].waku.brokerCtx, + MessageReceivedEvent.listen( + self.waku.brokerCtx, proc(event: MessageReceivedEvent) {.async: (raises: []).} = emitEvent("onMessageReceived"): $newJsonEvent("message_received", event), - ).valueOr: + ).isOkOr: chronicles.error "MessageReceivedEvent.listen failed", err = $error return err("MessageReceivedEvent.listen failed: " & $error) - let ConnectionStatusChangeListener = EventConnectionStatusChange.listen( - ctx.myLib[].waku.brokerCtx, + EventConnectionStatusChange.listen( + self.waku.brokerCtx, proc(event: EventConnectionStatusChange) {.async: (raises: []).} = emitEvent("onConnectionStatusChange"): $newJsonEvent("connection_status_change", event), - ).valueOr: + ).isOkOr: chronicles.error "ConnectionStatusChange.listen failed", err = $error return err("ConnectionStatusChange.listen failed: " & $error) - let shardTopicHealthListener = EventShardTopicHealthChange.listen( - ctx.myLib[].waku.brokerCtx, + EventShardTopicHealthChange.listen( + self.waku.brokerCtx, proc(event: EventShardTopicHealthChange) {.async: (raises: []).} = emitEvent("onTopicHealthChange"): $( @@ -142,12 +93,12 @@ proc logosdelivery_start_node( "topicHealth": $event.health, } ), - ).valueOr: + ).isOkOr: chronicles.error "EventShardTopicHealthChange.listen failed", err = $error return err("EventShardTopicHealthChange.listen failed: " & $error) - let peerEventListener = WakuPeerEvent.listen( - ctx.myLib[].waku.brokerCtx, + WakuPeerEvent.listen( + self.waku.brokerCtx, proc(event: WakuPeerEvent) {.async: (raises: []).} = emitEvent("onConnectionChange"): $( @@ -157,12 +108,12 @@ proc logosdelivery_start_node( "peerEvent": $event.kind, } ), - ).valueOr: + ).isOkOr: chronicles.error "WakuPeerEvent.listen failed", err = $error return err("WakuPeerEvent.listen failed: " & $error) - let channelReceivedListener = ChannelMessageReceivedEvent.listen( - ctx.myLib[].waku.brokerCtx, + ChannelMessageReceivedEvent.listen( + self.waku.brokerCtx, proc(event: ChannelMessageReceivedEvent) {.async: (raises: []).} = emitEvent("onChannelMessageReceived"): $( @@ -173,52 +124,61 @@ proc logosdelivery_start_node( "payload": string(base64.encode(event.payload)), } ), - ).valueOr: + ).isOkOr: chronicles.error "ChannelMessageReceivedEvent.listen failed", err = $error return err("ChannelMessageReceivedEvent.listen failed: " & $error) - let channelSentListener = ChannelMessageSentEvent.listen( - ctx.myLib[].waku.brokerCtx, + ChannelMessageSentEvent.listen( + self.waku.brokerCtx, proc(event: ChannelMessageSentEvent) {.async: (raises: []).} = emitEvent("onChannelMessageSent"): $newJsonEvent("channel_message_sent", event), - ).valueOr: + ).isOkOr: chronicles.error "ChannelMessageSentEvent.listen failed", err = $error return err("ChannelMessageSentEvent.listen failed: " & $error) - let channelErrorListener = ChannelMessageErrorEvent.listen( - ctx.myLib[].waku.brokerCtx, + ChannelMessageErrorEvent.listen( + self.waku.brokerCtx, proc(event: ChannelMessageErrorEvent) {.async: (raises: []).} = emitEvent("onChannelMessageError"): $newJsonEvent("channel_message_error", event), - ).valueOr: + ).isOkOr: chronicles.error "ChannelMessageErrorEvent.listen failed", err = $error return err("ChannelMessageErrorEvent.listen failed: " & $error) - (await ctx.myLib[].start()).isOkOr: + return ok() + +proc dropFFIEventListeners(self: LogosDelivery) {.async.} = + ## Reverse of `registerFFIEventListeners`. + await MessageErrorEvent.dropAllListeners(self.waku.brokerCtx) + await MessageSentEvent.dropAllListeners(self.waku.brokerCtx) + await MessagePropagatedEvent.dropAllListeners(self.waku.brokerCtx) + await MessageReceivedEvent.dropAllListeners(self.waku.brokerCtx) + await EventConnectionStatusChange.dropAllListeners(self.waku.brokerCtx) + await EventShardTopicHealthChange.dropAllListeners(self.waku.brokerCtx) + await WakuPeerEvent.dropAllListeners(self.waku.brokerCtx) + await ChannelMessageReceivedEvent.dropAllListeners(self.waku.brokerCtx) + await ChannelMessageSentEvent.dropAllListeners(self.waku.brokerCtx) + await ChannelMessageErrorEvent.dropAllListeners(self.waku.brokerCtx) + +proc logosdelivery_start_node( + self: LogosDelivery +): Future[Result[string, string]] {.ffi.} = + self.registerFFIEventListeners().isOkOr: + return err(error) + + (await self.start()).isOkOr: let errMsg = $error chronicles.error "START_NODE failed", err = errMsg return err("failed to start: " & errMsg) return ok("") proc logosdelivery_stop_node( - ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer -) {.ffiRaw.} = - requireInitializedNode(ctx, "STOP_NODE"): - return err(errMsg) + self: LogosDelivery +): Future[Result[string, string]] {.ffi.} = + await self.dropFFIEventListeners() - await MessageErrorEvent.dropAllListeners(ctx.myLib[].waku.brokerCtx) - await MessageSentEvent.dropAllListeners(ctx.myLib[].waku.brokerCtx) - await MessagePropagatedEvent.dropAllListeners(ctx.myLib[].waku.brokerCtx) - await MessageReceivedEvent.dropAllListeners(ctx.myLib[].waku.brokerCtx) - await EventConnectionStatusChange.dropAllListeners(ctx.myLib[].waku.brokerCtx) - await EventShardTopicHealthChange.dropAllListeners(ctx.myLib[].waku.brokerCtx) - await WakuPeerEvent.dropAllListeners(ctx.myLib[].waku.brokerCtx) - await ChannelMessageReceivedEvent.dropAllListeners(ctx.myLib[].waku.brokerCtx) - await ChannelMessageSentEvent.dropAllListeners(ctx.myLib[].waku.brokerCtx) - await ChannelMessageErrorEvent.dropAllListeners(ctx.myLib[].waku.brokerCtx) - - (await ctx.myLib[].stop()).isOkOr: + (await self.stop()).isOkOr: let errMsg = $error chronicles.error "STOP_NODE failed", err = errMsg return err("failed to stop: " & errMsg) diff --git a/logos_delivery.nimble b/logos_delivery.nimble index 72babe751..cd216f688 100644 --- a/logos_delivery.nimble +++ b/logos_delivery.nimble @@ -61,7 +61,7 @@ requires "nim >= 2.2.4", # Packages not on nimble (use git URLs) -requires "https://github.com/logos-messaging/nim-ffi#aad9374354a5e3d98964a9adf80766a12f8f200d" # v0.3.0-rc.1 +requires "https://github.com/logos-messaging/nim-ffi#53515de17af0ef3e88b2aec9675b8163dddc14ae" # v0.3.0-rc.2 requires "https://github.com/logos-messaging/nim-sds.git#b12f5ee07c5b764303b51fb948b32a4ade1de3b5" @@ -106,14 +106,25 @@ proc buildBinary(name: string, srcDir = "./", params = "") = exec "nim c --out:build/" & name & " --mm:refc " & getMyCPU() & getNimParams() & " " & params & " " & srcDir & name & ".nim" +## Emitted by `genBindings()` during the library build, so the header can never +## drift from the Nim signatures. Not checked in: it is a build artifact. +const cBindingsDir = "library/generated" + +## `-d:ffiSrcPath` is required: without it nim-ffi derives the path with +## `relativePath`, which needs `getcwd` at compile time and fails to build. +const cBindingsFlags = + " -d:ffiGenBindings -d:targetLang=c -d:ffiOutputDir=" & cBindingsDir & + " -d:ffiSrcPath=../liblogosdelivery.nim " + proc buildLibrary(lib_name: string, srcDir = "./", params = "", `type` = "static", srcFile = "liblogosdelivery.nim", mainPrefix = "liblogosdelivery") = if not dirExists "build": mkDir "build" + mkDir cBindingsDir if `type` == "static": exec "nim c" & " --out:build/" & lib_name & " --threads:on --app:staticlib --opt:speed --noMain --mm:refc --header -d:metrics --nimMainPrefix:" & mainPrefix & " --skipParentCfg:off -d:discv5_protocol_id=d5waku " & - getMyCPU() & getNimParams() & srcDir & "/" & srcFile + cBindingsFlags & getMyCPU() & getNimParams() & srcDir & "/" & srcFile else: # -Bsymbolic binds the library's references to its own symbols at link # time. Without it, a host process that already loads OpenSSL (e.g. @@ -123,7 +134,7 @@ proc buildLibrary(lib_name: string, srcDir = "./", params = "", `type` = "static let elfFlags = when defined(linux): "--passL:-Wl,-Bsymbolic " else: "" exec "nim c" & " --out:build/" & lib_name & " --threads:on --app:lib --opt:speed --noMain --mm:refc --header -d:metrics --nimMainPrefix:" & mainPrefix & " --skipParentCfg:off -d:discv5_protocol_id=d5waku " & - elfFlags & getMyCPU() & getNimParams() & " " & srcDir & "/" & srcFile + elfFlags & cBindingsFlags & getMyCPU() & getNimParams() & " " & srcDir & "/" & srcFile proc buildLibDynamicWindows(libName: string, folderName: string) = buildLibrary libName & ".dll", folderName, diff --git a/nimble.lock b/nimble.lock index 0bf286688..8d924d507 100644 --- a/nimble.lock +++ b/nimble.lock @@ -644,7 +644,7 @@ }, "ffi": { "version": "0.3.0", - "vcsRevision": "aad9374354a5e3d98964a9adf80766a12f8f200d", + "vcsRevision": "53515de17af0ef3e88b2aec9675b8163dddc14ae", "url": "https://github.com/logos-messaging/nim-ffi", "downloadMethod": "git", "dependencies": [ @@ -655,7 +655,7 @@ "cbor_serialization" ], "checksums": { - "sha1": "db5fc50aa4717418e481cb0b1f7ca36f2d76586c" + "sha1": "1d84ceaf8594f4970c5a37f916003ffc0531dc4e" } }, "boringssl": { diff --git a/nix/deps.nix b/nix/deps.nix index b8e3928c5..858d4b58b 100644 --- a/nix/deps.nix +++ b/nix/deps.nix @@ -285,8 +285,8 @@ ffi = pkgs.fetchgit { url = "https://github.com/logos-messaging/nim-ffi"; - rev = "aad9374354a5e3d98964a9adf80766a12f8f200d"; - sha256 = "075ax4spvzr7idd5b5sncpkr7b3163qncr8fsxv9d02dix4ailqc"; + rev = "53515de17af0ef3e88b2aec9675b8163dddc14ae"; + sha256 = "0ncf9j7fhgd3nswr4rh19jx77dl974sajphdl04cb602hshgj5ij"; fetchSubmodules = true; }; diff --git a/nix/submodules.json b/nix/submodules.json index e6a40b523..c9ac2d814 100644 --- a/nix/submodules.json +++ b/nix/submodules.json @@ -56,7 +56,7 @@ { "path": "vendor/nim-ffi", "url": "https://github.com/logos-messaging/nim-ffi", - "rev": "aad9374354a5e3d98964a9adf80766a12f8f200d" + "rev": "53515de17af0ef3e88b2aec9675b8163dddc14ae" } , { diff --git a/tests-e2e/src/node/wrapper_helpers.py b/tests-e2e/src/node/wrapper_helpers.py index 7c9dd1a2d..6c2e6558e 100644 --- a/tests-e2e/src/node/wrapper_helpers.py +++ b/tests-e2e/src/node/wrapper_helpers.py @@ -181,9 +181,9 @@ def get_node_multiaddr(node) -> str: list), this fails loudly instead of silently passing a malformed string downstream to staticnodes / add_peers. """ - result = node.get_node_info_raw("MyMultiaddresses") + result = node.get_node_info("MyMultiaddresses") if result.is_err(): - raise RuntimeError(f"get_node_info_raw failed: {result.err()}") + raise RuntimeError(f"get_node_info failed: {result.err()}") addr = result.ok_value.strip() if not addr or not addr.startswith("/"): diff --git a/tests-e2e/src/node/wrappers_manager.py b/tests-e2e/src/node/wrappers_manager.py index a0d20e0d6..7eafb2d66 100644 --- a/tests-e2e/src/node/wrappers_manager.py +++ b/tests-e2e/src/node/wrappers_manager.py @@ -74,25 +74,8 @@ class WrapperManager: def get_available_node_info_ids(self, *, timeout_s: float = 20.0) -> Result[list[str], str]: return self._node.get_available_node_info_ids(timeout_s=timeout_s) - def get_node_info(self, node_info_id: str, *, timeout_s: float = 20.0) -> Result[dict, str]: + def get_node_info(self, node_info_id: str, *, timeout_s: float = 20.0) -> Result[str, str]: return self._node.get_node_info(node_info_id, timeout_s=timeout_s) - def get_node_info_raw(self, node_info_id: str, *, timeout_s: float = 20.0) -> Result[str, str]: - """Like get_node_info but returns the raw string without JSON parsing.""" - from wrapper import lib, ffi, _new_cb_state, _wait_cb_raw # type: ignore[import] - - state = _new_cb_state() - cb = self._node._make_waiting_cb(state) - rc = lib.logosdelivery_get_node_info(self._node.ctx, cb, ffi.NULL, node_info_id.encode("utf-8")) - if rc != 0: - return Err(f"get_node_info_raw: immediate call failed (ret={rc})") - wait_result = _wait_cb_raw(state, "get_node_info_raw", timeout_s) - if wait_result.is_err(): - return Err(wait_result.err()) - cb_ret, cb_msg = wait_result.ok_value - if cb_ret != 0: - return Err(f"get_node_info_raw: callback failed (ret={cb_ret})") - return Ok(cb_msg.decode("utf-8") if cb_msg else "") - def get_available_configs(self, *, timeout_s: float = 20.0) -> Result[dict, str]: return self._node.get_available_configs(timeout_s=timeout_s) diff --git a/tests-e2e/vendor/logos-delivery-python-bindings/waku/wrapper.py b/tests-e2e/vendor/logos-delivery-python-bindings/waku/wrapper.py index f47d1a9b8..1133b4582 100644 --- a/tests-e2e/vendor/logos-delivery-python-bindings/waku/wrapper.py +++ b/tests-e2e/vendor/logos-delivery-python-bindings/waku/wrapper.py @@ -9,25 +9,40 @@ ffi = FFI() ffi.cdef( """ +// Raw FFICallBack, used by the event listener registry and by the +// scalar-fast-path exports (no string arguments). typedef void (*FFICallBack)(int callerRet, const char *msg, size_t len, void *userData); +// Reply callback of the `abi = c` exports that take arguments. `reply` and +// `errMsg` are NUL-terminated and valid only for the duration of the call. +typedef void (*ReplyFn)(int errCode, const char *reply, const char *errMsg, void *userData); + +// The constructor reports the context address as decimal text. +typedef void (*CreateRawFn)(int errCode, const char *ctxAddr, const char *errMsg, void *userData); + +typedef struct { const char *configJson; } CreateNodeCtorReq; +typedef struct { const char *contentTopicStr; } SubscribeReq; +typedef struct { const char *contentTopicStr; } UnsubscribeReq; +typedef struct { const char *messageJson; } SendReq; +typedef struct { const char *nodeInfoId; } GetNodeInfoReq; + void *logosdelivery_create_node( - const char *configJson, - FFICallBack callback, + const CreateNodeCtorReq *req, + CreateRawFn onCreated, void *userData ); -int logosdelivery_start_node( - void *ctx, - FFICallBack callback, - void *userData -); +int logosdelivery_destroy(void *ctx); -int logosdelivery_stop_node( - void *ctx, - FFICallBack callback, - void *userData -); +int logosdelivery_start_node(void *ctx, FFICallBack callback, void *userData); +int logosdelivery_stop_node(void *ctx, FFICallBack callback, void *userData); +int logosdelivery_get_available_node_info_ids(void *ctx, FFICallBack callback, void *userData); +int logosdelivery_get_available_configs(void *ctx, FFICallBack callback, void *userData); + +int logosdelivery_subscribe(void *ctx, ReplyFn onReply, void *userData, const SubscribeReq *req); +int logosdelivery_unsubscribe(void *ctx, ReplyFn onReply, void *userData, const UnsubscribeReq *req); +int logosdelivery_send(void *ctx, ReplyFn onReply, void *userData, const SendReq *req); +int logosdelivery_get_node_info(void *ctx, ReplyFn onReply, void *userData, const GetNodeInfoReq *req); uint64_t logosdelivery_add_event_listener( void *ctx, @@ -40,52 +55,6 @@ int logosdelivery_remove_event_listener( void *ctx, uint64_t listenerId ); - -int logosdelivery_destroy( - void *ctx, - FFICallBack callback, - void *userData -); - -int logosdelivery_subscribe( - void *ctx, - FFICallBack callback, - void *userData, - const char *contentTopic -); - -int logosdelivery_unsubscribe( - void *ctx, - FFICallBack callback, - void *userData, - const char *contentTopic -); - -int logosdelivery_send( - void *ctx, - FFICallBack callback, - void *userData, - const char *messageJson -); - -int logosdelivery_get_available_node_info_ids( - void *ctx, - FFICallBack callback, - void *userData -); - -int logosdelivery_get_node_info( - void *ctx, - FFICallBack callback, - void *userData, - const char *nodeInfoId -); - -int logosdelivery_get_available_configs( - void *ctx, - FFICallBack callback, - void *userData -); """ ) @@ -93,17 +62,15 @@ _repo_root = Path(__file__).resolve().parents[1] lib = ffi.dlopen(str(_repo_root / "lib" / "liblogosdelivery.so")) CallbackType = ffi.callback("void(int, const char*, size_t, void*)") +ReplyCallbackType = ffi.callback("void(int, const char*, const char*, 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. +# Non-terminal progress tick (~every 5s while a request is in flight), always +# followed by a terminal RET_OK/RET_ERR. The waiting callbacks drop it so a slow +# call (start_node most of all) is not latched as a result. 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. +# Every event the library emits. Since 0.3.0 a listener is registered per event +# name, so an `event_cb` that wants them all registers once per name. EVENT_NAMES = ( "onMessageSent", "onMessageError", @@ -118,43 +85,6 @@ EVENT_NAMES = ( "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 { @@ -173,17 +103,10 @@ def _wait_cb_raw( if not ok: return Err(f"{op_name}: timeout after {timeout_s}s") - cb_ret = state["ret"] - if cb_ret is None: + if state["ret"] is None: return Err(f"{op_name}: callback ret is None") - 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}") + return Ok((state["ret"], state["msg"])) def _wait_cb_ok(state, op_name: str, timeout_s: float = 20.0) -> Result[int, str]: @@ -220,15 +143,29 @@ class NodeWrapper: return CallbackType(c_cb) + @staticmethod + def _make_waiting_reply_cb(state): + def c_cb(err_code, reply_p, err_p, userData): + if int(err_code) == RET_STALE_WARN: + return + + text_p = reply_p if int(err_code) == 0 else err_p + msg = ffi.string(text_p) if text_p != ffi.NULL else b"" + + if not state["done"].is_set(): + state["ret"] = int(err_code) + state["msg"] = msg + state["done"].set() + + return ReplyCallbackType(c_cb) + @staticmethod def _make_event_cb(py_callback): def c_cb(ret, char_p, length, userData): msg = ffi.buffer(char_p, length)[:] if char_p != ffi.NULL else b"" py_callback(int(ret), msg) - handler = CallbackType(c_cb) - _PINNED_EVENT_CALLBACKS.append(handler) - return handler + return CallbackType(c_cb) @classmethod def create_node( @@ -242,41 +179,40 @@ class NodeWrapper: config_buffer = ffi.new("char[]", config_json.encode("utf-8")) state = _new_cb_state() - cb = cls._make_waiting_cb(state) + cb = cls._make_waiting_reply_cb(state) - ctx = lib.logosdelivery_create_node( - config_buffer, - cb, - ffi.NULL, - ) + req = ffi.new("CreateNodeCtorReq *", {"configJson": config_buffer}) + lib.logosdelivery_create_node(req, cb, ffi.NULL) + + wait_result = _wait_cb_ok(state, "create_node", timeout_s) + if wait_result.is_err(): + return Err(wait_result.err()) + + # The constructor reports the context address as decimal text. + try: + ctx = ffi.cast("void *", int(state["msg"].decode("utf-8"))) + except Exception as e: + return Err(f"create_node: invalid context address: {e}") if ctx == ffi.NULL: return Err("create_node: ctx is NULL") - node = cls(ctx, config_buffer, None) + event_cb_handler = None + listener_ids = [] + if event_cb is not None: + event_cb_handler = cls._make_event_cb(event_cb) + for event_name in EVENT_NAMES: + listener_id = lib.logosdelivery_add_event_listener( + ctx, + event_name.encode("utf-8"), + event_cb_handler, + ffi.NULL, + ) + if listener_id == 0: + return Err(f"create_node: add_event_listener({event_name}) failed") + listener_ids.append(listener_id) - wait_result = _wait_cb_ok(state, "create_node", timeout_s) - if wait_result.is_err(): - node.destroy() - return Err(wait_result.err()) - - 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_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(node) + return Ok(cls(ctx, config_buffer, event_cb_handler, listener_ids)) @classmethod def create_and_start( @@ -327,7 +263,7 @@ class NodeWrapper: def destroy(self, *, timeout_s: float = 20.0) -> Result[int, str]: if self.ctx == ffi.NULL: - return Ok(RET_OK) + return Ok(0) # Drop the listeners first so the event thread cannot reach the Python # callback once the context is gone. @@ -335,19 +271,12 @@ class NodeWrapper: lib.logosdelivery_remove_event_listener(self.ctx, listener_id) self._listener_ids = () - state = _new_cb_state() - cb = self._make_waiting_cb(state) - - rc = lib.logosdelivery_destroy(self.ctx, cb, ffi.NULL) + rc = lib.logosdelivery_destroy(self.ctx) if rc != 0: - return Err(f"destroy: immediate call failed (ret={rc})") - - wait_result = _wait_cb_ok(state, "destroy", timeout_s) - if wait_result.is_err(): - return Err(wait_result.err()) + return Err(f"destroy: call failed (ret={rc})") self.ctx = ffi.NULL - return wait_result + return Ok(rc) def stop_and_destroy(self, *, timeout_s: float = 20.0) -> Result[int, str]: stop_result = self.stop_node(timeout_s=timeout_s) @@ -363,14 +292,11 @@ class NodeWrapper: def subscribe_content_topic(self, content_topic: str, *, timeout_s: float = 20.0) -> Result[int, str]: state = _new_cb_state() - cb = self._make_waiting_cb(state) + cb = self._make_waiting_reply_cb(state) - rc = lib.logosdelivery_subscribe( - self.ctx, - cb, - ffi.NULL, - content_topic.encode("utf-8"), - ) + topic_buffer = ffi.new("char[]", content_topic.encode("utf-8")) + req = ffi.new("SubscribeReq *", {"contentTopicStr": topic_buffer}) + rc = lib.logosdelivery_subscribe(self.ctx, cb, ffi.NULL, req) if rc != 0: return Err(f"subscribe_content_topic: immediate call failed (ret={rc})") @@ -378,14 +304,11 @@ class NodeWrapper: def unsubscribe_content_topic(self, content_topic: str, *, timeout_s: float = 20.0) -> Result[int, str]: state = _new_cb_state() - cb = self._make_waiting_cb(state) + cb = self._make_waiting_reply_cb(state) - rc = lib.logosdelivery_unsubscribe( - self.ctx, - cb, - ffi.NULL, - content_topic.encode("utf-8"), - ) + topic_buffer = ffi.new("char[]", content_topic.encode("utf-8")) + req = ffi.new("UnsubscribeReq *", {"contentTopicStr": topic_buffer}) + rc = lib.logosdelivery_unsubscribe(self.ctx, cb, ffi.NULL, req) if rc != 0: return Err(f"unsubscribe_content_topic: immediate call failed (ret={rc})") @@ -393,16 +316,13 @@ class NodeWrapper: def send_message(self, message: dict, *, timeout_s: float = 20.0) -> Result[str, str]: state = _new_cb_state() - cb = self._make_waiting_cb(state) + cb = self._make_waiting_reply_cb(state) message_json = json.dumps(message, separators=(",", ":"), ensure_ascii=False) - rc = lib.logosdelivery_send( - self.ctx, - cb, - ffi.NULL, - message_json.encode("utf-8"), - ) + message_buffer = ffi.new("char[]", message_json.encode("utf-8")) + req = ffi.new("SendReq *", {"messageJson": message_buffer}) + rc = lib.logosdelivery_send(self.ctx, cb, ffi.NULL, req) if rc != 0: return Err(f"send_message: immediate call failed (ret={rc})") @@ -436,20 +356,17 @@ class NodeWrapper: return Err("get_available_node_info_ids: empty response") try: - return Ok(json.loads(cb_msg.decode("utf-8").strip().lstrip("@"))) + return Ok(json.loads(cb_msg.decode("utf-8"))) except Exception as e: return Err(f"get_available_node_info_ids: invalid response: {e}") - def get_node_info(self, node_info_id: str, *, timeout_s: float = 20.0) -> Result[dict, str]: + def get_node_info(self, node_info_id: str, *, timeout_s: float = 20.0) -> Result[str, str]: state = _new_cb_state() - cb = self._make_waiting_cb(state) + cb = self._make_waiting_reply_cb(state) - rc = lib.logosdelivery_get_node_info( - self.ctx, - cb, - ffi.NULL, - node_info_id.encode("utf-8"), - ) + info_id_buffer = ffi.new("char[]", node_info_id.encode("utf-8")) + req = ffi.new("GetNodeInfoReq *", {"nodeInfoId": info_id_buffer}) + rc = lib.logosdelivery_get_node_info(self.ctx, cb, ffi.NULL, req) if rc != 0: return Err(f"get_node_info: immediate call failed (ret={rc})") @@ -461,15 +378,10 @@ class NodeWrapper: if cb_ret != 0: return Err(f"get_node_info: callback failed (ret={cb_ret}) msg={cb_msg!r}") - if not cb_msg: - return Err("get_node_info: empty response") - - try: - result = json.loads(cb_msg.decode("utf-8")) - except Exception as e: - return Err(f"get_node_info: invalid json: {e}") - - return Ok(result) + # The item is a plain string, not JSON: a peer id, an ENR URI, a + # comma-separated multiaddress list or the Prometheus metrics text. + # MyMixPubKey is legitimately empty when mix is not mounted. + return Ok(cb_msg.decode("utf-8")) def get_available_configs(self, *, timeout_s: float = 20.0) -> Result[dict, str]: state = _new_cb_state()