Events and Queues

Events broadcast a message to every subscriber; queues deliver each message to exactly one competing consumer. This guide builds one flow of each.

Broadcast with events

publish-event broadcasts the current message to a subject, and every flow with an events source on that subject receives its own copy: fan-out, not load balancing. samples/events.yaml wires one publisher to two subscribers:

flows:
  # Publisher: fire on a schedule and broadcast a notification.
  - name: emitter
    source:
      type: cron                 # implicit connector, no instance needed
      settings:
        schedule: "@every 3s"
        payload: '{"tick": string(now)}'
    process:
      - type: publish-event
        name: broadcast
        settings:
          # CEL subject, so a flow can route dynamically; here it is constant.
          subject: '"notifications"'

  # Subscriber A: receives every notification broadcast to the subject.
  - name: subscriber-a
    source:
      type: events               # implicit connector, no instance needed
      settings:
        subject: notifications
    process:
      - type: log
        name: emit-a
        settings:
          logger: out
          message: '"subscriber A saw tick " + body.tick'

Subscriber B is identical. Run it and watch both subscribers handle every tick:

octo run --config samples/events.yaml
# every few seconds:
#   "subscriber A saw tick ..."
#   "subscriber B saw tick ..."

The subject in publish-event is a CEL expression: quote a constant ('"notifications"') or compute it per message ('"orders." + vars.region'). Neither side declares a connector instance, because events has no global settings and the runtime resolves an implicit one on demand.

Broadcast fits independent reactions to the same fact (update a cache, send a notification, record metrics); adding a subscriber never affects the others.

Load-balance with queues

With queues, subscribers on a subject become competing consumers, and each message is handled by exactly one of them, cluster-wide. samples/queue-loadbalance.yaml uses this as the cross-replica analogue of flow-ref: an HTTP flow hands work to worker flows through queue subjects instead of calling them in-process.

process:
  # Fire-and-forget publish: the caller does not wait. Any replica
  # subscribed to the subject may pick it up.
  - type: queue-dispatch
    name: audit-async
    settings:
      subject: '"audit-work"'      # subject is a CEL expression

  # Request/reply: wait for one competing consumer to answer, and fold
  # the reply's body and variables back into this message.
  - type: queue-dispatch
    name: enrich-sync
    settings:
      subject: '"enrich-work"'
      awaitReply: true

Each worker is an ordinary flow fronted by a queue source:

- name: order-enricher
  source:
    type: queue                    # implicit connector, no instance needed
    settings:
      subject: enrich-work
      listeners: 8                 # concurrent handlers on this replica
  process:
    - type: set-payload
      name: normalize-order
      settings:
        value: >
          {
            "orderId":   vars.id,
            "tenant":    vars["X-Tenant"],
            "item":      body.item,
            "amount":    body.amount,
            "requestId": correlationID
          }

Try it:

octo run --config samples/queue-loadbalance.yaml
curl -s -X POST localhost:8080/api/v1/orders/42 \
  -H 'X-Tenant: acme' -H 'X-Request-Id: req-1' \
  -d '{"item":"widget","amount":1500}'

The semantics mirror flow-ref. The default dispatch publishes and moves on, the queue-side counterpart of oneWay; the order-auditor worker runs purely for its side effect. awaitReply: true opts into request/reply: whichever consumer is free handles the message, and its final body and variables are folded back into the dispatching flow before the next block runs. An optional timeout (e.g. 30s) bounds that wait. listeners on the queue source sets how many messages one replica handles concurrently; run more replicas of the whole service and the queue spreads messages across all of them.

Which one do I want?

If losing a second subscriber would change the meaning of your system, you want a queue; if adding a fifth subscriber should be free, you want events.

What backs them

In the standalone module (octo run on your machine) both are in-process. On the platform the same YAML is NATS-backed: events become a plain pub/sub subscription that reaches every replica, and queues become NATS queue subscriptions with competing consumers cluster-wide. See Clustering.

Where to go next