pubsub

Deprecated: pubsub is deprecated (pubsub) and will be removed in an upcoming release. Behavior is unchanged for now. Migration guide: https://github.com/iii-hq/iii/blob/6901a589dd39d9c34f9d5b6bd8bc82170ee21548/docs/next/upgrading/migrate-from-streams.mdx

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:

Field	Required	Default	Description
topic	yes	—	Topic to subscribe to. A missing or empty topic registers nothing (a warning is logged).
condition_function_id	no	—	Accepted for schema parity with the builtin; never evaluated.

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:

Field	Description
topic	Topic to publish to. Empty → topic_not_set error.
data	JSON payload delivered verbatim to each subscriber.

Returns a null result on success.

Configuration

Field	Default	Description
adapter.name	local	Backend: local (in-process broadcast) or redis (cross-instance Redis Pub/Sub).
adapter.config.redis_url	redis://localhost:6379	Redis connection URL (only for the redis adapter).

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

Behavior	Builtin	This worker
Trigger type	subscribe	same
Trigger config	{topic} (+ advertised condition_function_id, never evaluated)	same (still never evaluated)
Service functions	publish (bare id — the #[service] prefix is discarded by the macro)	publish (exact same id)
publish input	{topic, data}	same
publish success result	null	same
publish empty topic	error topic_not_set / "Topic is not set"	same (code prefixes the message: topic_not_set: Topic is not set)
Delivery payload	raw data, no envelope	same
Delivery semantics	fire-and-forget, results ignored, no retries	same
No subscribers	silent no-op	same
Subscribe with missing/empty topic	warn + register nothing (Ok)	same
Backends	local (in-process), redis	same
Redis: subscriptions per topic per instance	one (second warns, dropped)	same
Local unsubscribe	drops the ENTIRE topic entry (bug: kills co-subscribers)	removes only the given id (deliberate fix)
Adapter hot-swap	gated build-first, resubscribe-before-swap	same
Config entry id	legacy engine configuration	pubsub (new entry, seeded on first boot)
Config schema shape	hand-built oneOf union (local/redis)	plain derive — accepted values identical; explicit Console form