Subscriptions

Subscription patterns determine which events are delivered to a subscriber. Ironflow uses NATS-style wildcards for flexible pattern matching.

Wildcard Syntax

Token	Description	Example
*	Matches one segment.	system.run.*.created
>	Matches all remaining segments.	events:>

---

Connection Lifecycle

For real-time subscriptions, the SDK maintains a persistent connection (WebSocket or ConnectRPC stream).

TypeScript

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

// 1. Connect to the stream
await ironflow.connect();

// 2. Monitor connection state
const off = ironflow.onConnectionChange((state) => {
  console.log(`Connection state: ${state}`); // 'connected', 'reconnecting', etc.
});

// 3. Disconnect when finished
ironflow.disconnect();
off();

Go

// Connect with a timeout context
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()

if err := subClient.Connect(ctx); err != nil {
    log.Fatal(err)
}
defer subClient.Close()

Python

from ironflow import IronflowRPC

# The client opens a stream on first iteration and closes every stream on exit.
with IronflowRPC(server_url="http://localhost:9123") as rpc:
    ...

Python has no connection-state callback: each subscription is its own ConnectRPC stream. Do not set a client-wide timeout here: it bounds a whole subscription, not the gap between events.

---

Pattern Helpers

TypeScript

Browser (Singleton):

import { ironflow } from "@ironflow/browser";
await ironflow.subscribe("events:order.*", { onEvent: (e) => {} });

Node.js (Instance):

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

const subClient = createSubscriptionClient({
  serverUrl: "http://localhost:9123",
  apiKey: "...",
});
await subClient.connect();
const sub = await subClient.subscribe("system.run.>", { onEvent: (e) => {} });

Go

sub, err := subClient.Subscribe(ctx, ironflow.Patterns.AllRuns(), nil)

Python

from ironflow.rpc.v1 import SubscribeRequest

events = rpc.pubsub.subscribe(SubscribeRequest(pattern="system.run.>"))

---

Replay Then Live

Set replay to receive the last N events of the pattern first. The same subscription then delivers new events as they arrive. replay is a count, not a position. To resume after a known event, use startAfterSequence (see below).

TypeScript

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

Go

sub, err := subClient.Subscribe(ctx, "topic:orders", &ironflow.SubscribeOptions{
    Replay: 10,
})

Python

from ironflow.rpc.v1 import SubscribeOptions, SubscribeRequest

for event in rpc.pubsub.subscribe(SubscribeRequest(
    pattern="topic:orders",
    options=SubscribeOptions(replay=10),
)):
    print(event.topic, event.data.to_python())

---

Lifecycle & Storage

Regular subscriptions are ephemeral and stored in-memory on the server.

1. Disconnect: If a client disconnects, the subscription is removed from the server.
2. Reconnect: SDKs restore the pattern, filter, consumer group, metadata, acknowledgment, backpressure, and namespace options.
3. Position: The start position depends on how you subscribed.

replay applies only to the initial subscription. Reapplying the same count on every reconnect would repeat an arbitrary historical window. On reconnect:

• A consumer group resumes from its server-owned durable position.
• A fan-out subscription with startAfterSequence resumes after the last sequence delivered to the caller. Delivery remains at least once.
• An unpositioned fan-out subscription starts at the current tail and may miss events published while it was offline.

Use startAfterSequence for positioned fan-out delivery. Do not combine it with replay or a consumer group.

onEvent is not awaited, so advance the cursor only after your own work completes. An async handler runs unsupervised: the next event can be delivered before it settles, and a rejection becomes an unhandled rejection rather than reaching onError. Serialize the handler yourself if the cursor must never run ahead of the work it stands for.

TypeScript

await ironflow.subscribe("events:>", {
  startAfterSequence: lastDeliveredSequence,
  onEvent: (event) => console.log(event),
});

Go

sub, err := subClient.Subscribe(ctx, "events:>", &ironflow.SubscribeOptions{
    StartAfterSequence: &lastDeliveredSequence,
})

Python

from ironflow.rpc.v1 import SubscribeOptions, SubscribeRequest

events = rpc.pubsub.subscribe(SubscribeRequest(
    pattern="events:>",
    options=SubscribeOptions(start_after_sequence=last_delivered_sequence),
))

---

Resource Limits

Two server-side caps apply to subscriptions. Both are fixed defaults — there is no flag, env var or YAML key that changes them.

Limit	Value	On breach
Subscriptions per WebSocket connection, plain subscriptions and consumer-group joins together	100	The subscribe is rejected with LIMIT_EXCEEDED. A ConnectRPC Subscribe call is its own stream and is not counted.
replay count, on WebSocket and ConnectRPC	1000	The subscribe is rejected with the message replay limit exceeded. ConnectRPC returns invalid_argument. WebSocket maps no code for this one, so it arrives as internal.

Use startAfterSequence rather than a large replay when you need to resume a long gap — it is a cursor, not a bounded backlog request.