Ironflow is pre-1.0: APIs can change between releases. Pin a version. What is ready →

Quick Script

This script allows you to verify your Ironflow connection and see events flowing in real-time. Use this as a diagnostic tool or a starting point for your own subscribers.

How to Run

Start Ironflow: Run ironflow serve in one terminal.
Run Subscriber: Run the code below in a second terminal.
Trigger Events: Use ironflow emit in a third terminal to see events appear.

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

async function main() {
  const client = createSubscriptionClient({
    serverUrl: "http://localhost:9123",
  });
  await client.connect();

  // Subscribe to all system run events
  await client.subscribe("system.run.>", {
    onEvent: (e) => console.log(`[${e.topic}]`, e.data),
    replay: 5,
  });

  // Subscribe to all user events
  await client.subscribe("events:>", {
    onEvent: (e) => console.log(`[${e.topic}]`, e.data),
    replay: 5,
  });

  console.log("Listening for events...");
}

main().catch(console.error);

func main() {
    client := ironflow.NewSubscriptionClient(ironflow.SubscriptionClientConfig{
        WSURL: "ws://localhost:9123/ws",
    })

    ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt)
    defer stop()

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

    // Subscribe to all events
    sub, _ := client.Subscribe(ctx, "events:>", &ironflow.SubscribeOptions{Replay: 5})

    fmt.Println("Listening for events...")
    for {
        select {
        case event := <-sub.Events():
            fmt.Printf("Received: %s\n", event.Topic)
        case <-ctx.Done():
            return
        }
    }
}

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


def main():
    with IronflowRPC(server_url="http://localhost:9123") as rpc:
        # Subscribe to all user events
        events = rpc.pubsub.subscribe(SubscribeRequest(
            pattern="events:>",
            options=SubscribeOptions(replay=5),
        ))
        print("Listening for events...")
        for event in events:
            print(f"Received: {event.topic}")


if __name__ == "__main__":
    main()  # Ctrl+C to stop