pubsub

Deprecated: pubsub is deprecated (pubsub) and will be removed in an upcoming release. Behavior is unchanged for now. Migration guide: https://iii.dev/docs/upgrading/migrate-from-streams

This covers the publish function, the subscribe trigger type and the pubsub configuration entry. For new code, use a worker-owned trigger type for fire-and-forget fan-out, or the queue worker (iii::durable::publish + durable:subscriber) when delivery must not be lost. While deprecated, this worker logs a warning (target iii::deprecation) once at startup, on publish calls and on subscribe registrations (not per delivered event), at most once every 10 minutes per entry point and caller. Payloads and topic names are never logged. Traffic on topic stream.events (the engine's internal iii-stream bridge) does not warn.

Topic-based publish/subscribe messaging. Registers the subscribe trigger type and the publish service function. This standalone worker replaces the legacy built-in pub/sub service. 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 trigger compose::add worker=pubsub

iii trigger compose::add resolves the worker and its dependencies, writes exact declarations to worker-compose.yaml, and reconciles the Compose project.

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 legacy built-in pub/sub service

The legacy built-in pub/sub service 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 it 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 the legacy built-in 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