Merge pull request #7 from logos-messaging/channel-API-wrappers

Channel api wrappers

.gitignore

@@ -3,4 +3,5 @@ dist/
 waku.egg-info/
 waku/__pycache__/
 /vendor/
+data/
 

vendor/logos-delivery

@@ -1 +1 @@
-Subproject commit 4099ff26382480ae2946f27087b430323330799e
+Subproject commit 53c084dfdb2c4366bbcb8e8ee967c13f4b9f294c

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)