logos-messaging/logos-messaging-python-bindings
1
0
Fork 0
mirror of https://github.com/logos-messaging/logos-messaging-python-bindings.git synced 2026-04-08 05:13:10 +00:00
Code Issues Packages Projects Releases Wiki Activity

Adding the test version of the wrapper file

Browse Source
This commit is contained in:
Aya Hassan 2026-03-30 11:19:16 +02:00
parent eb7ab05e8d
commit 17e526c4a1
1 changed files with 33 additions and 34 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

67
waku/wrapper.py
View File

@ -78,6 +78,12 @@ int logosdelivery_get_node_info(
void *userData,
const char *nodeInfoId
);
int logosdelivery_get_available_configs(
void *ctx,
FFICallBack callback,
void *userData
);
"""
)
@ -333,48 +339,40 @@ class NodeWrapper:
)
if rc < 0:
return Err(f"get_node_info: immediate call failed (ret={rc})")
return Err(f"call failed rc={rc}")
wait_result = _wait_cb_raw(state, "get_node_info", timeout_s)
wait = _wait_cb_raw(state, "get_node_info", timeout_s)
if wait.is_err():
return Err(wait.err())
cb_ret, cb_msg = wait.ok_value
if cb_ret != 0 or not cb_msg:
return Err(f"callback failed ret={cb_ret}")
try:
return Ok(json.loads(cb_msg.decode()))
except Exception:
return Err("invalid json")
def get_available_configs(self, *, timeout_s: float = 20.0) -> Result[bytes, str]:
state = _new_cb_state()
cb = self._make_waiting_cb(state)
rc = lib.logosdelivery_get_available_configs(self.ctx, cb, ffi.NULL)
if rc != 0:
return Err(f"get_available_configs failed: {rc}")
wait_result = _wait_cb_raw(state, "get_available_configs", 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: failed (ret={cb_ret}) msg={cb_msg!r}")
return Err(f"get_available_configs failed: {cb_ret}")
if not cb_msg:
return Err("get_node_info: empty response")
return Ok(cb_msg)
try:
result = json.loads(cb_msg.decode("utf-8"))
except Exception as e:
return Err(f"get_node_info: failed to parse response: {e}")
return Ok(result)
def debug_get_node_info(self, node_info_id: str, *, timeout_s: float = 20.0) -> Result[tuple, str]:
state = _new_cb_state()
cb = self._make_waiting_cb(state)
rc = lib.logosdelivery_get_node_info(
self.ctx,
cb,
ffi.NULL,
node_info_id.encode("utf-8"),
)
print(f"[DEBUG] get_node_info immediate rc={rc}")
wait_result = _wait_cb_raw(state, "get_node_info", timeout_s)
if wait_result.is_err():
return Err(wait_result.err())
cb_ret, cb_msg = wait_result.ok_value
print(f"[DEBUG] get_node_info callback ret={cb_ret}, msg={cb_msg}")
return Ok((rc, cb_ret, cb_msg))
def main():
config = {
"logLevel": "DEBUG",
@ -412,6 +410,7 @@ def main():
print(node.subscribe_content_topic(topic))
print(node.send_message(message))
#print(node.get_available_node_info_ids())
#print(node.debug_get_available_configs())
print(node.stop_node())
print(node.destroy())