pubsub

Topic-based publish/subscribe messaging. Registers the subscribe trigger type and the publish service function, replacing the engine builtin iii-pubsub. Any function bound to a subscribe trigger on a topic receives every event published to that topic; publish fans an event's data out to all of a topic's subscribers as fire-and-forget invocations.

Install

iii worker add pubsub

iii worker add fetches the binary, writes a config block into ~/.iii/config.yaml, and the engine starts the worker on the next iii start.

Quickstart

Bind a function to the subscribe trigger type with a topic, then call the publish function with { topic, data }:

use iii_sdk::protocol::{RegisterTriggerInput, TriggerRequest};
use iii_sdk::{InitOptions, RegisterFunction, errors::Error, register_worker};
use serde_json::{json, Value};

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let iii = register_worker("ws://localhost:49134", InitOptions::default());

    iii.register_function(
        "orders::on_created",
        RegisterFunction::new_async(|data: Value| async move {
            println!("order event: {data}");
            Ok::<Value, Error>(Value::Null)
        }),
    );

    iii.register_trigger(RegisterTriggerInput {
        trigger_type: iii_pubsub::TRIGGER_TYPE.to_string(), // "subscribe"
        function_id: "orders::on_created".into(),
        config: json!({ "topic": "orders" }),
        metadata: None,
    })?;

    // Any caller (function, worker, or the stream bridge) publishes with:
    iii.trigger(TriggerRequest {
        function_id: "publish".into(),
        payload: json!({ "topic": "orders", "data": { "id": 42 } }),
        action: None,
        timeout_ms: Some(5000),
    })
    .await?;

    tokio::signal::ctrl_c().await?;
    Ok(())
}

The subscriber receives the raw data value (no envelope). Delivery is fire-and-forget: publish returns a null result once dispatched and does not wait for or report subscriber outcomes. Publishing to a topic with no subscribers is a silent no-op; publishing with an empty topic fails with a topic_not_set error.

Trigger type

This worker always registers the subscribe trigger type. Bind a function to it with:

The publish function

Registered under the bare id publish (matching the builtin exactly, so existing callers and the engine's stream bridge keep working). Input:

Returns a null result on success.

Configuration

Configuration is owned by the configuration worker — edit it from the console (Configuration → Workers → pubsub) or seed it once via --config .yaml on first boot. The adapter hot-swaps at runtime: a change rebuilds the backend, re-subscribes the live subscriptions onto it before the swap (no delivery gap — a brief double-delivery overlap is accepted for fire-and-forget pub/sub), then tears down the previous backend. A build failure keeps the previous backend and config.

Requires removing the built-in iii-pubsub worker

The built-in iii-pubsub worker also owns the subscribe trigger type and the publish function. Two owners of the same trigger type on one engine collide — whichever registers last wins — so this worker requires iii-pubsub to be absent: omit it from the engine's config.yaml (a config that doesn't list a worker won't run it).

On boot, this worker queries the engine for connected workers and refuses to start with a clear error if iii-pubsub is still active, so a stale config fails loudly instead of silently racing the built-in worker for ownership of subscribe/publish.

Parity vs builtin