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..7a7e128 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,86 @@ 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)