From 7fd83e40eb36e4dd836da7f46ac14c8fb1c383d2 Mon Sep 17 00:00:00 2001 From: Aya Hassan Date: Sun, 12 Jul 2026 18:20:52 +0200 Subject: [PATCH 1/2] Adding new wrappers for channel API , ignore generated data files --- .gitignore | 1 + vendor/logos-delivery | 2 +- waku/wrapper.py | 173 ++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 175 insertions(+), 1 deletion(-) diff --git a/.gitignore b/.gitignore index 5522b69..619cb76 100644 --- a/.gitignore +++ b/.gitignore @@ -3,4 +3,5 @@ dist/ waku.egg-info/ waku/__pycache__/ /vendor/ +data/ diff --git a/vendor/logos-delivery b/vendor/logos-delivery index 4099ff2..53c084d 160000 --- a/vendor/logos-delivery +++ b/vendor/logos-delivery @@ -1 +1 @@ -Subproject commit 4099ff26382480ae2946f27087b430323330799e +Subproject commit 53c084dfdb2c4366bbcb8e8ee967c13f4b9f294c diff --git a/waku/wrapper.py b/waku/wrapper.py index e92812e..89cd17f 100644 --- a/waku/wrapper.py +++ b/waku/wrapper.py @@ -80,6 +80,30 @@ int logosdelivery_get_available_configs( FFICallBack callback, void *userData ); + +int logosdelivery_channel_create( + void *ctx, + FFICallBack callback, + void *userData, + const char *channelId, + const char *contentTopic, + const char *senderId +); + +int logosdelivery_channel_send( + void *ctx, + FFICallBack callback, + void *userData, + const char *channelId, + const char *messageJson +); + +int logosdelivery_channel_close( + void *ctx, + FFICallBack callback, + void *userData, + const char *channelId +); """ ) @@ -413,3 +437,152 @@ class NodeWrapper: return Err(f"get_available_configs: invalid json: {e}") return Ok(result) + + def channel_create( + self, + channel_id: str, + content_topic: str, + sender_id: str, + *, + timeout_s: float = 20.0, + ) -> Result[str, str]: + state = _new_cb_state() + cb = self._make_waiting_cb(state) + + rc = lib.logosdelivery_channel_create( + self.ctx, + cb, + ffi.NULL, + channel_id.encode("utf-8"), + content_topic.encode("utf-8"), + sender_id.encode("utf-8"), + ) + if rc != 0: + return Err(f"channel_create: immediate call failed (ret={rc})") + + wait_result = _wait_cb_raw(state, f"channel_create({channel_id})", 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(cb_msg.decode("utf-8") if cb_msg else f"channel_create({channel_id}): callback failed (ret={cb_ret})") + + created_channel_id = cb_msg.decode("utf-8") if cb_msg else "" + return Ok(created_channel_id) + + def channel_send(self, channel_id: str, message: dict, *, timeout_s: float = 20.0) -> Result[str, str]: + state = _new_cb_state() + cb = self._make_waiting_cb(state) + + message_json = json.dumps(message, separators=(",", ":"), ensure_ascii=False) + + rc = lib.logosdelivery_channel_send( + self.ctx, + cb, + ffi.NULL, + channel_id.encode("utf-8"), + message_json.encode("utf-8"), + ) + if rc != 0: + return Err(f"channel_send: immediate call failed (ret={rc})") + + wait_result = _wait_cb_raw(state, f"channel_send({channel_id})", 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(cb_msg.decode("utf-8") if cb_msg else f"channel_send({channel_id}): callback failed (ret={cb_ret})") + + channel_req_id = cb_msg.decode("utf-8") if cb_msg else "" + return Ok(channel_req_id) + + def channel_close(self, channel_id: str, *, timeout_s: float = 20.0) -> Result[int, str]: + state = _new_cb_state() + cb = self._make_waiting_cb(state) + + rc = lib.logosdelivery_channel_close( + self.ctx, + cb, + ffi.NULL, + channel_id.encode("utf-8"), + ) + if rc != 0: + return Err(f"channel_close: immediate call failed (ret={rc})") + + wait_result = _wait_cb_raw(state, f"channel_close({channel_id})", 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(cb_msg.decode("utf-8") if cb_msg else f"channel_close({channel_id}): callback failed (ret={cb_ret})") + + return Ok(cb_ret) + + +# --------------------------------------------------------------------------- +# TEMPORARY manual smoke test for the channel wrappers. +# Run: python -m waku.wrapper (from the repo root, with lib/liblogosdelivery.so built) +# Remove before merging. +# --------------------------------------------------------------------------- +if __name__ == "__main__": + import base64 + import socket + + def _free_port() -> int: + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: + s.bind(("", 0)) + return s.getsockname()[1] + + def _show(label, result): + tag = "OK " if result.is_ok() else "ERR" + value = result.ok_value if result.is_ok() else result.err() + print(f"[{tag}] {label}: {value!r}") + + config = { + "logLevel": "INFO", + "listenAddress": "0.0.0.0", + "tcpPort": _free_port(), + "discv5UdpPort": _free_port(), + "restPort": _free_port(), + "restAddress": "0.0.0.0", + "clusterId": "198", + "relay": True, + "store": False, + "filter": False, + "lightpush": False, + "peerExchange": False, + "discv5Discovery": False, + } + + CHANNEL_ID = "smoke-channel" + CONTENT_TOPIC = "/test/1/channel/proto" + SENDER_ID = "smoke-sender" + + node_result = NodeWrapper.create_and_start(config) + if node_result.is_err(): + print(f"[ERR] create_and_start: {node_result.err()!r}") + raise SystemExit(1) + node = node_result.ok_value + print("[OK ] node created and started") + + try: + # RC01: create, then duplicate create rejected. + _show("channel_create", node.channel_create(CHANNEL_ID, CONTENT_TOPIC, SENDER_ID)) + _show("channel_create (duplicate)", node.channel_create(CHANNEL_ID, CONTENT_TOPIC, SENDER_ID)) + + # RC02: send on unknown channel. + payload = base64.b64encode(b"hello channel").decode("ascii") + _show("channel_send (unknown id)", node.channel_send("no-such-channel", {"payload": payload, "ephemeral": False})) + + # RC04: empty payload rejected; then a valid send returns a handle. + _show("channel_send (empty payload)", node.channel_send(CHANNEL_ID, {"payload": "", "ephemeral": False})) + _show("channel_send", node.channel_send(CHANNEL_ID, {"payload": payload, "ephemeral": False})) + + # RC03: close, then close unknown rejected. + _show("channel_close", node.channel_close(CHANNEL_ID)) + _show("channel_close (unknown id)", node.channel_close(CHANNEL_ID)) + finally: + _show("stop_and_destroy", node.stop_and_destroy()) From 694833acfb632105a4d5ab010edfdbe1dd46bbce Mon Sep 17 00:00:00 2001 From: Aya Hassan Date: Sun, 12 Jul 2026 18:23:19 +0200 Subject: [PATCH 2/2] remove the temp test code --- waku/wrapper.py | 66 ------------------------------------------------- 1 file changed, 66 deletions(-) diff --git a/waku/wrapper.py b/waku/wrapper.py index 89cd17f..7a7e128 100644 --- a/waku/wrapper.py +++ b/waku/wrapper.py @@ -520,69 +520,3 @@ class NodeWrapper: return Err(cb_msg.decode("utf-8") if cb_msg else f"channel_close({channel_id}): callback failed (ret={cb_ret})") return Ok(cb_ret) - - -# --------------------------------------------------------------------------- -# TEMPORARY manual smoke test for the channel wrappers. -# Run: python -m waku.wrapper (from the repo root, with lib/liblogosdelivery.so built) -# Remove before merging. -# --------------------------------------------------------------------------- -if __name__ == "__main__": - import base64 - import socket - - def _free_port() -> int: - with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: - s.bind(("", 0)) - return s.getsockname()[1] - - def _show(label, result): - tag = "OK " if result.is_ok() else "ERR" - value = result.ok_value if result.is_ok() else result.err() - print(f"[{tag}] {label}: {value!r}") - - config = { - "logLevel": "INFO", - "listenAddress": "0.0.0.0", - "tcpPort": _free_port(), - "discv5UdpPort": _free_port(), - "restPort": _free_port(), - "restAddress": "0.0.0.0", - "clusterId": "198", - "relay": True, - "store": False, - "filter": False, - "lightpush": False, - "peerExchange": False, - "discv5Discovery": False, - } - - CHANNEL_ID = "smoke-channel" - CONTENT_TOPIC = "/test/1/channel/proto" - SENDER_ID = "smoke-sender" - - node_result = NodeWrapper.create_and_start(config) - if node_result.is_err(): - print(f"[ERR] create_and_start: {node_result.err()!r}") - raise SystemExit(1) - node = node_result.ok_value - print("[OK ] node created and started") - - try: - # RC01: create, then duplicate create rejected. - _show("channel_create", node.channel_create(CHANNEL_ID, CONTENT_TOPIC, SENDER_ID)) - _show("channel_create (duplicate)", node.channel_create(CHANNEL_ID, CONTENT_TOPIC, SENDER_ID)) - - # RC02: send on unknown channel. - payload = base64.b64encode(b"hello channel").decode("ascii") - _show("channel_send (unknown id)", node.channel_send("no-such-channel", {"payload": payload, "ephemeral": False})) - - # RC04: empty payload rejected; then a valid send returns a handle. - _show("channel_send (empty payload)", node.channel_send(CHANNEL_ID, {"payload": "", "ephemeral": False})) - _show("channel_send", node.channel_send(CHANNEL_ID, {"payload": payload, "ephemeral": False})) - - # RC03: close, then close unknown rejected. - _show("channel_close", node.channel_close(CHANNEL_ID)) - _show("channel_close (unknown id)", node.channel_close(CHANNEL_ID)) - finally: - _show("stop_and_destroy", node.stop_and_destroy())