Topics

Topics provide a general-purpose messaging layer on top of Ironflow’s NATS infrastructure. Use them for service-to-service communication, analytics, or notifications — without triggering workflows.

Topics vs Events

Feature	emit()	publish()
Namespace	events:	topic:
Triggers Functions?	Yes	No
Use Case	Business logic triggers.	Real-time messages / broadcasting.

A topic name is at most 249 bytes. It cannot contain * or >, whitespace, control characters or an empty segment (a leading, trailing or doubled .). It cannot start with events:, system., entity:, public. or topic:, which Ironflow owns. A bad name fails with invalid_argument. See Reserved names.

---

Publishing to Topics

From Application Code

TypeScript

import { createClient } from "@ironflow/node";

const ironflow = createClient({ apiKey: process.env.IRONFLOW_API_KEY });

await ironflow.publish("notifications.email", {
  to: "user@example.com",
  body: "Welcome to Ironflow!",
});

Go

client := ironflow.NewClient(ironflow.ClientConfig{
    APIKey: os.Getenv("IRONFLOW_API_KEY"),
})

client.Publish(ctx, "notifications.email", map[string]string{
    "to": "user@example.com",
})

Python

import os
from ironflow import IronflowRPC
from ironflow.rpc.v1 import PublishRequest
from protobuf.wkt import Struct

with IronflowRPC(api_key=os.environ["IRONFLOW_API_KEY"]) as rpc:
    res = rpc.pubsub.publish(PublishRequest(
        topic="notifications.email",
        data=Struct.from_python({"to": "user@example.com", "body": "Welcome to Ironflow!"}),
    ))
    print(res.event_id, res.sequence)

rpc.pubsub also exposes subscriptions, topic queries and consumer-group operations.

CLI

# Publish inline JSON
ironflow topic publish notifications.email --data '{"to":"user@example.com"}'

# Or from a file, with an idempotency key
ironflow topic publish notifications.email -f payload.json --idempotency-key evt-123

# Discover what is being published
ironflow topic list
ironflow topic stats notifications.email

From Within a Workflow

Use step.publish() to emit a message as a durable step. This ensures that if the workflow retries, the message is not published more than once.

import { ironflow } from "@ironflow/node";

export const myFn = ironflow.createFunction(config, async ({ step }) => {
  // Durable publish — memoized and observable
  await step.publish("analytics.workflow_step", { status: "reached" });
});

// Durable publish: memoized under the step name "publish:analytics.workflow_step"
if err := ironflow.Publish(ctx, "analytics.workflow_step", map[string]string{"status": "reached"}); err != nil {
    return nil, err
}

from ironflow.worker import function


@function(id="my-fn", triggers=[{"event": "order.placed"}])
async def my_fn(ctx):
    # Durable publish: memoized and observable
    result = await ctx.step.publish("analytics.workflow_step", {"status": "reached"})
    print(result.event_id, result.sequence)

Pass idempotency_key= to make the publish idempotent on the server too. The server remembers a key for one topic in one environment, for IRONFLOW_RUN_RETENTION_DAYS days after the first publish (kept forever when unset or 0). A repeat returns duplicate with the first event_id.

Under serve(), the function needs server_url= or IRONFLOW_SERVER_URL for step.publish. Without it, the run fails with “server URL not configured for publish step”. Pull-mode workers already have their URL.

---

Subscribing to Topics

Topics use the topic: prefix for subscriptions.

TypeScript

import { ironflow } from "@ironflow/browser";

await ironflow.subscribe("topic:notifications.>", {
  onEvent: (event) => console.log("Broadcasting:", event.data),
});

Go

subClient := ironflow.NewSubscriptionClient(ironflow.SubscriptionClientConfig{WSURL: "ws://localhost:9123/ws"}) // API key from IRONFLOW_API_KEY
if err := subClient.Connect(ctx); err != nil {
    log.Fatal(err)
}
sub, err := subClient.Subscribe(ctx, "topic:notifications.>", nil)
if err != nil {
    log.Fatal(err)
}
for event := range sub.Events() {
    fmt.Println("Broadcasting:", string(event.Data))
}

Python

from ironflow import IronflowRPC
from ironflow.rpc.v1 import SubscribeRequest

with IronflowRPC(server_url="http://localhost:9123") as rpc:
    for event in rpc.pubsub.subscribe(SubscribeRequest(pattern="topic:notifications.>")):
        print("Broadcasting:", event.data.to_python())

CLI

# Listen to all developer topics
ironflow subscribe "topic:>"

Clean Patterns

Use topic:notifications.* to listen to specific categories, or topic:> to listen to all custom messages in the system.