feat(zenorb): zoci ext methods (#1164)

## changes
adds and documents extension methods for the Zenoh Orb Command
Interface: small helpers for sending commands using zenoh queries and
using strings/json underneath

small example, using utility methods:
- `Query::args`
- `Query::res`
- `zenorb::Session::command_raw`
- `Result<Reply, ReplyErr>::json`

```rust
use zenorb::zoci::{ReplyExt, ZociQueryExt};

blue.receiver(())
    .queryable("tuple", async |_ctx, query| {
        let args: (String, String) = query.args()?;
        query.res(&args).await?;

        Ok(())
    })
    .run()
    .await?;

let red = Zenorb::from_cfg(client_cfg.clone())
    .orb_id(orb_id.clone())
    .with_name("red")
    .await?;

let reply = red.command_raw("blue/tuple", "one two").await?;
let reply: Result<(String, String), String> = reply.json()?;
```

Author: vmenge

zenorb/README.md

@@ -1,171 +1,301 @@
 # zenorb
 
-A helper library for using [Zenoh](https://zenoh.io/) with orb-specific conventions. Just a small
-wrapper for delcaring publishers, queriers, queryables and subscribers. Tries to use native `zenoh` types
-as much as possible.
+`zenorb` is a small wrapper around [Zenoh](https://zenoh.io/) for orb services.
 
-## Design Decisions
+It keeps Zenoh's native types and behavior, but adds a few orb-specific
+conventions:
+
+- declared publisher and querier registries
+- shared context for subscribers and queryables
+- consistent orb/service key expression prefixes
+- a small command/reply layer for query-based interactions
 
-### 1. Publisher & Querier Registry
+## Design Decisions
 
-Zenoh encourages holding declared publishers and queriers across the application lifetime for performance optimizations. Without this, you'd need to either:
+### 1. Declared Publishers and Queriers
 
-1. Declare publishers/queriers on every send (inefficient)
-2. Manually manage a hashmap of declared publishers/queriers throughout your codebase (boilerplate)
+Zenoh performs best when publishers and queriers are declared once and kept for
+the lifetime of the process.
 
-`Sender` solves this by maintaining an internal registry of declared publishers and queriers. You declare them once at startup, then retrieve them by keyexpr when needed:
+Without a wrapper, each service has to either redeclare them repeatedly or build
+its own registry and thread that through the application. `Sender` provides that
+registry directly. You declare publishers and queriers at startup and retrieve
+them later by key expression.
 
 ```rust
-// At startup: declare all publishers and queriers
-let sender = session
+use zenoh::bytes::Encoding;
+use zenorb::Zenorb;
+
+let zenorb = Zenorb::from_cfg(cfg)
+    .orb_id(orb_id)
+    .with_name("banana")
+    .await?;
+
+let sender = zenorb
     .sender()
     .publisher("events")
     .publisher_with("metrics", |p| p.encoding(Encoding::APPLICATION_JSON))
-    .querier("other-service/status")
+    .querier("apple/status")
     .build()
     .await?;
 
-// Later: use them by keyexpr
+// Later
 sender.publisher("events")?.put(payload).await?;
-sender.querier("other-service/status")?.get().await?;
+sender.querier("apple/status")?.get().await?;
 ```
 
-`Sender` is `Clone` (wrapping an `Arc<Registry>`), so it can be passed around to different parts of your application.
+`Sender` is cheap to clone, so it can be shared freely across the application.
 
-The only time you should use Zenoh's publishers directly (without declaring) is when topics are dynamic and not known at startup.
+### 2. Shared Context for Receivers
 
-### 2. Context Injection & Automatic Error Logging
+`Receiver` lets you register subscribers and queryables that all receive the
+same cloned context.
 
-`Receiver` accepts a generic `Ctx` type that gets cloned and passed to every handler. This enables:
+That is useful for shared state such as database handles, caches, metrics
+clients, or test doubles. Handlers return `Result<()>`, and `zenorb` logs
+failures with the relevant key expression so each call site does not have to
+repeat the same error-handling boilerplate.
 
-- **Dependency injection**: Pass shared state, database connections, or test mocks
-- **Testability**: Inject test doubles without restructuring your handlers
+```rust
+#[derive(Clone)]
+struct AppCtx {
+    db: Db,
+    metrics: Metrics,
+}
 
-Additionally, handlers return `Result<()>`, and errors are automatically logged with the keyexpr context. This eliminates repetitive error handling boilerplate:
+zenorb
+    .receiver(AppCtx { db, metrics })
+    .subscriber("apple/events", async |ctx, sample| {
+        let event: Event = serde_json::from_slice(&sample.payload().to_bytes())?;
 
-```rust
-// Without zenorb: manual cloning and error handling everywhere
-let subscriber = session.declare_subscriber("events").await?;
-let db = db.clone();
-let metrics = metrics.clone();
-task::spawn(async move {
-    while let Ok(sample) = subscriber.recv_async().await {
-        let db = db.clone();
-        let metrics = metrics.clone();
-        let result = async move {
-            let data: Event = deserialize(&sample)?;
-            db.insert(&data).await?;
-            metrics.record(&data).await?;
-            Ok(())
-        };
-
-        if let Err(e) = result.await {
-            tracing::error!("Handler failed: {e}");
-        }
-    }
-});
-
-// With zenorb: context injection and automatic error logging
-session
-    .receiver(Ctx { db, metrics })
-    .subscriber("events", async |ctx, sample| {
-        let data: Event = deserialize(&sample)?;
-        ctx.db.insert(&data).await?;
-        ctx.metrics.record(&data).await?;
+        ctx.db.insert(&event).await?;
+        ctx.metrics.record(&event).await?;
 
         Ok(())
     })
+    .run()
+    .await?;
 ```
 
-### 3. Standardized Topic Format
-
-All orb sessions need to namespace their topics by orb ID and a name to avoid collisions. Without this library, every service would need to:
+### 3. Orb and Service Naming
 
-1. Carry around, `orb_id`, and `service_name` values
-2. Remember to format topics correctly
-3. Risk inconsistent formatting across services
+Orb services need stable key expressions so publishers, subscribers, queriers,
+and queryables all agree on the same paths.
 
-zenorb standardizes this in the `Session`:
+`Zenorb` carries the orb ID and service name once:
 
 ```rust
-let session = Session::from_cfg(cfg)
+use zenorb::Zenorb;
+
+let zenorb = Zenorb::from_cfg(cfg)
     .orb_id(orb_id)
-    .with_name("my-service")
+    .with_name("banana")
     .await?;
 ```
 
-Topic formats are then applied automatically:
+From there, `zenorb` applies the prefixes for you:
 
-| Type       | Format                      |
-| ---------- | --------------------------- |
-| Publisher  | `<orb-id>/<name>/<keyexpr>` |
-| Subscriber | `<orb-id>/<keyexpr>`        |
-| Queryable  | `<orb-id>/<name>/<keyexpr>` |
-| Querier    | `<orb-id>/<keyexpr>`        |
+| Type | Format |
+| --- | --- |
+| Publisher | `<orb-id>/<service>/<keyexpr>` |
+| Queryable | `<orb-id>/<service>/<keyexpr>` |
+| Subscriber | `<orb-id>/<keyexpr>` |
+| Querier | `<orb-id>/<keyexpr>` |
 
-Note that subscribers and queriers omit the service name since they typically listen to or query other services. The service name is part of the keyexpr you provide:
+That means publishers and queryables are service-scoped, while subscribers and
+queriers target the full service path you provide.
 
 ```rust
-// Service "banana" publishes to: ea2ea744/banana/events
-// Service "apple" subscribes to: ea2ea744/banana/events
-session
+// Service "banana" publishes to ea2ea744/banana/events
+sender.publisher("events")?.put(payload).await?;
+
+// Service "apple" subscribes to ea2ea744/banana/events
+apple
     .receiver(ctx)
-    .subscriber("banana/events", handler)  // keyexpr includes source service
+    .subscriber("banana/events", async |ctx, sample| {
+        Ok(())
+    })
     .run()
     .await?;
 ```
 
-## Example
+If `Receiver::queryable(...)` is too opinionated for a use case,
+`Zenorb::declare_queryable(...)` exposes the underlying Zenoh queryable builder
+while keeping the same service-scoped prefix.
+
+### 4. ZOCI - Zenoh Orb Command Interface
+
+Some queries are really commands: one request, one reply, typed success or typed
+error.
+
+`zenorb::zoci` defines a small convention for that pattern.
+
+Caller-side helpers:
+
+- `Sender::command(...)` sends a JSON payload through a declared querier
+- `Sender::command_raw(...)` sends a raw string payload through a declared querier
+- `Zenorb::command(...)` sends a JSON payload with `Zenorb::get(...)`
+- `Zenorb::command_raw(...)` sends a raw string payload with `Zenorb::get(...)`
+
+Handler-side helpers:
+
+- `query.json()` decodes a JSON request payload
+- `query.args()` decodes a space-delimited argument payload
+- `query.res(...)` sends a JSON success reply
+- `query.res_err(...)` sends a JSON error reply
+
+Reply-side helper:
+
+- `ReplyExt::json()` decodes a reply into `Result<OkType, ErrType>`
+
+The command API keeps Zenoh's two layers of failure visible:
+
+- outer `Result`: transport or session failure
+- inner `Result`: success reply or `reply_err(...)`
+
+#### ZOCI Example
 
 ```rust
-use orb_info::{orb_os_release::OrbRelease, OrbId};
-use zenoh::bytes::Encoding;
+use serde::{Deserialize, Serialize};
+use zenorb::{
+    zoci::{ReplyExt, ZociQueryExt},
+    Zenorb,
+};
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
+struct StatusRequest {
+    id: u64,
+    label: String,
+}
+
+let red = Zenorb::from_cfg(client_cfg.clone())
+    .orb_id(orb_id.clone())
+    .with_name("red")
+    .await?;
+
+let blue = Zenorb::from_cfg(client_cfg)
+    .orb_id(orb_id)
+    .with_name("blue")
+    .await?;
+
+let sender = red
+    .sender()
+    .querier("blue/status")
+    .build()
+    .await?;
+
+blue.receiver(())
+    .queryable("status", async |_ctx, query| {
+        let req: StatusRequest = query.json()?;
+        query.res(&req).await?;
+
+        Ok(())
+    })
+    .run()
+    .await?;
+
+let reply = sender
+    .command(
+        "blue/status",
+        &StatusRequest {
+            id: 7,
+            label: "banana".into(),
+        },
+    )
+    .await?;
+
+let reply: Result<StatusRequest, StatusRequest> = reply.json()?;
+```
+
+For lighter payloads, pair `command_raw(...)` with `query.args()`:
+
+```rust
+use zenorb::zoci::{ReplyExt, ZociQueryExt};
+
+blue.receiver(())
+    .queryable("tuple", async |_ctx, query| {
+        let args: (String, String) = query.args()?;
+        query.res(&args).await?;
+
+        Ok(())
+    })
+    .run()
+    .await?;
+
+let sender = red
+    .sender()
+    .querier("blue/tuple")
+    .build()
+    .await?;
+
+let reply = sender.command_raw("blue/tuple", "one two").await?;
+let reply: Result<(String, String), String> = reply.json()?;
+```
+
+## Basic Publish/Subscribe Example
+
+The examples above show the command/reply flow. The example below goes back to
+the core `zenorb` pattern: declared publishers on one service, subscribers on
+another, and JSON payload handling in the subscriber callback.
+
+```rust
+use orb_info::OrbId;
+use serde::{Deserialize, Serialize};
+use std::{str::FromStr, sync::Arc};
+use tokio::sync::Mutex;
+use zenorb::Zenorb;
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+struct Message {
+    text: String,
+}
 
 #[derive(Clone)]
 struct AppCtx {
-    received: Arc<Mutex<Vec<Message>>>,
+    received: Arc<Mutex<Vec<Message>>>
 }
 
 #[tokio::main]
-async fn main() -> Result<()> {
+async fn main() -> color_eyre::Result<()> {
     let cfg = zenorb::client_cfg(7447);
     let orb_id = OrbId::from_str("ea2ea744")?;
 
-    // Create sessions with two different names
-    let banana_session = Session::from_cfg(cfg.clone())
+    let banana = Zenorb::from_cfg(cfg.clone())
         .orb_id(orb_id.clone())
         .with_name("banana")
         .await?;
 
-    let apple_session = Session::from_cfg(cfg)
+    let apple = Zenorb::from_cfg(cfg)
         .orb_id(orb_id)
         .with_name("apple")
         .await?;
 
-    // Set up the sender with declared publishers
-    let sender = banana_session
+    let sender = banana
         .sender()
         .publisher("notifications")
         .build()
         .await?;
 
-    // Set up the receiver with context and handlers
-    let ctx = AppCtx { received: Arc::new(Mutex::new(vec![])) };
+    let ctx = AppCtx {
+        received: Arc::new(Mutex::new(vec![])),
+    };
 
-    apple_session
-        .receiver(ctx)
+    apple
+        .receiver(ctx.clone())
         .subscriber("banana/notifications", async |ctx, sample| {
             let msg: Message = serde_json::from_slice(&sample.payload().to_bytes())?;
             ctx.received.lock().await.push(msg);
+
             Ok(())
         })
         .run()
         .await?;
 
-    // Publish messages
     sender
         .publisher("notifications")?
-        .put(serde_json::to_vec(&Message::new("hello"))?)
+        .put(serde_json::to_vec(&Message {
+            text: "hello".to_string(),
+        })?)
         .await?;
 
     Ok(())

zenorb/src/lib.rs

@@ -1,3 +1,5 @@
+pub mod zoci;
+
 mod receiver;
 mod sender;
 mod session;