rb_zenoh

rb_zenoh is a shared Zenoh-based communication package. Its main features are publish/subscribe, query/queryable, and ZenohRouter decorator-based routing.

Installation/Dependencies

• Python: >=3.12,<3.13
• Package dependencies: flatbuffers, psutil, rb_utils, rb_modules
• The _core native extension is not built during package installation. Build the _core*.so/_core*.pyd for the required platforms first, then include it in the package.

# For the current host
bash scripts/backend/build_cpp_core.sh zenoh

# For Docker/Linux
RB_CPP_CORE_TARGETS="linux/amd64 linux/arm64" bash scripts/backend/build_cpp_core.sh zenoh

The distributed rainbow-rb-zenoh is not built as a fat wheel; it is split into per-platform wheels. Each wheel contains only the single _core artifact for its platform.

Core Objects

• ZenohClient: low-level communication API
• ZenohRouter: decorator-based router
• SubscribeOptions: options controlling subscribe behavior

1) publish

from rb_zenoh.client import ZenohClient

client = ZenohClient()
client.publish("muscat/sample/topic", payload={"ok": True})

Publishing a FlatBuffer request:

from rb_flat_buffers.IPC.Request_MotionPause import Request_MotionPauseT

req = Request_MotionPauseT()
client.publish(
    "C500920/call_pause",
    flatbuffer_req_obj=req,
    flatbuffer_buf_size=64,
)

2) subscribe

from rb_zenoh.client import ZenohClient
from rb_zenoh.schema import SubscribeOptions

client = ZenohClient()

async def on_msg(*, topic, mv, obj_payload, dict_payload, attachment):
    print(topic, obj_payload, dict_payload, attachment)

client.subscribe(
    "muscat/sample/topic",
    on_msg,
    options=SubscribeOptions(dispatch="immediate"),
)

Callback signature:

• topic: topic string
• mv: raw payload (memoryview)
• obj_payload: object parsed via flatbuffer_obj_t=<FlatBuffer T class> (T | None)
• dict_payload: dict conversion result, regardless of whether flatbuffer_obj_t is set (dict | None)
• attachment: sender, sender_id information

3) queryable

def on_query(req=None, params=None):
    return {"status": "ok"}

client.queryable("muscat/sample/query", on_query)

Callback arguments:

• req: query payload body
  • Injected only when flatbuffer_req_t/flatbuffer_req_T_class is specified.
  • A FlatBuffer ...T object parsed internally via InitFromPackedBuf(...).
  • If flatbuffer_req_t is specified but the request has no payload, an error is raised.
• params: query parameters
  • The internal q.parameters value is injected as a dict[str, str].
  • Used to pass filters/options separately from the payload (req).

Notes:

• Arguments are auto-injected only when the callback parameters are named exactly req and params.
• Parameters declared with other names such as request or query_params are not injected.
• If neither is declared, the callback is invoked with no arguments.

4) query_one / query_all

query_one

res = client.query_one("muscat/program/state", timeout=0.3)
print(res.get("dict_payload"))

• Returns only the first reply
• Raises ZenohNoReply if there is no reply

query_all

res_list = client.query_all("*/health", timeout=0.5)
for item in res_list:
    print(item.get("key"), item.get("dict_payload"))

• Returns all replies received within the timeout as a list
• Returns an empty list if there are no replies

Which to use when

• Calling a single target: query_one
• Collecting from wildcards/multiple services: query_all

FlatBuffer Parameter Details

flatbuffer_req_obj

• Meaning: the FlatBuffer object to send as the request (an object supporting Pack())
• Used in: publish, query_one, query_all
• Internal behavior: packed with a builder, then sent as a bytes payload

flatbuffer_buf_size

• Meaning: initial buffer size of the FlatBuffer serialization builder
• Used in:
  • Request serialization: publish, query_one, query_all
  • queryable response serialization: queryable(..., flatbuffer_res_buf_size=...)
• Serialization may fail if it is too small
• Recommended: start from 64/128/256/512 and increase to fit the payload size

flatbuffer_res_T_class

• Meaning: specifies which FlatBuffer type to parse the response payload as
• Used in: query_one, query_all
• Type: must be a FlatBuffer Object API class (T class)
  • e.g. Response_FunctionsT, State_CoreT
  • Usually a class whose name ends in ...T
  • Parsed internally via InitFromPackedBuf(...)
• Result:
  • If specified: parsed object in obj_payload
  • If not specified: rely mainly on dict_payload/raw payload

Example:

from rb_flat_buffers.IPC.Request_MotionPause import Request_MotionPauseT
from rb_flat_buffers.IPC.Response_Functions import Response_FunctionsT

res = client.query_one(
    "C500920/call_pause",
    flatbuffer_req_obj=Request_MotionPauseT(),
    flatbuffer_res_T_class=Response_FunctionsT,
    flatbuffer_buf_size=64,
    timeout=0.3,
)
obj = res.get("obj_payload")

5) Using ZenohRouter

from rb_zenoh.router import ZenohRouter

router = ZenohRouter(prefix="muscat/common")

@router.subscribe("health")
async def on_health(*, topic, obj_payload, dict_payload, attachment):
    pass

@router.queryable("echo")
async def on_echo(params=None):
    return {"ok": True}

Lifecycle:

• On startup: await router.startup()
• On shutdown: await router.shutdown()

Exceptions/Notes

• query_one no-reply: ZenohNoReply
• Transport reconnection errors: ZenohTransportError
• ZenohClient is a per-process singleton, so avoid unnecessary frequent close() calls