Build
Streaming triggers
A Live stream trigger keeps a persistent subscription to a message system and starts a run for each message as it arrives.
Supported stream sources
| Source | What it connects to | Main settings |
|---|---|---|
| Salesforce (Change Data Capture) | Salesforce CDC and Platform Events | Topic (for example /data/AccountChangeEvent), instance URL, org (tenant) id |
| Kafka / Azure Event Hubs | Kafka, Confluent Cloud, Redpanda, Event Hubs | Brokers, topic, TLS, SASL authentication (none, PLAIN, SCRAM-SHA-256 or SCRAM-SHA-512) and user, where to start the first time (latest or earliest), what to do if the saved position has expired |
| AWS Kinesis | Kinesis Data Streams | Stream name, region, access key id, start position |
| AWS SQS | Standard and FIFO queues | Queue URL, region, wait time, batch size, messages in flight, visibility timeout |
| MQTT | Mosquitto, HiveMQ, other brokers | Broker URL (mqtt:// or mqtts://), topic with +/# wildcards, client id, QoS 0/1/2 |
| Redis Stream | Redis streams | URL, stream key, consumer group, how work is shared |
| Redis (key changes / CDC) | Redis keyspace notifications | URL, database, key pattern, include values |
| WebSocket | Any WebSocket feed | URL, optional first message to subscribe, the field that identifies a message |
| Server-Sent Events | Any SSE endpoint | URL |
| RabbitMQ / AMQP 0-9-1 | RabbitMQ and compatible brokers | URL, queue, virtual host, user, prefetch |
| AMQP 1.0 | Azure Service Bus, ActiveMQ, Qpid, Solace | URL, source address, user, messages in flight |
| NATS / JetStream | NATS core and JetStream | URL, subject, mode (core or JetStream), queue group, stream, durable consumer |
| BullMQ (Redis) | BullMQ job queues | URL, queue, prefix, concurrency, lock and stall settings, rate limit |
| Google Cloud Pub/Sub | Pub/Sub subscriptions | Subscription (full name), batch size |
Setting up a streaming trigger
- Add the trigger. Add Live stream from the Triggers group. If you imported an AsyncAPI spec, Start from an imported API event fills in the details.
- Choose the Stream source and fill in its settings.
- Choose the Connection that holds the password, token or keys. Credentials always live in the connection, never in the trigger settings.
- Activate. Save and set the flow Active. The subscription opens and each message starts a run.
How it behaves
- No duplicates. Each message is identified by its id, offset or sequence number; a message delivered twice starts only one run.
- Resume where it left off. Offset-based sources (Kafka, Kinesis, Redis Streams) remember their position per partition or shard; acknowledgement-based sources (RabbitMQ, AMQP 1.0, SQS, Pub/Sub) acknowledge a message only after it is safely accepted. For WebSocket and SSE, name the field that identifies a message so repeats can be recognized.
- Backpressure. When your workspace is busy, intake slows down instead of dropping messages.
JMS brokers through AMQP 1.0
Brokers that serve Java applications through JMS (ActiveMQ Artemis, ActiveMQ Classic, Apache Qpid, Solace, Azure Service Bus and RabbitMQ 4) also speak AMQP 1.0, the open protocol their JMS clients use on the wire. The AMQP 1.0 connector sends and receives on them directly, with no Java runtime, in the cloud or on an On-Prem Worker, so messages exchanged with JMS applications keep their properties and types.
- Enable AMQP 1.0 on the broker. ActiveMQ, Qpid and Solace offer it as an acceptor or service you switch on; Azure Service Bus and RabbitMQ 4 have it built in. Create an AMQP 1.0 connection with the host, port (usually 5672, or 5671 with TLS), user and password.
- Address the destination. Put the queue or topic in Target to send and in Source to receive, written the way the broker names it (see the table below).
- Send like a JMS producer. Use Send a message, fill the JMS headers in their AMQP fields (see the mapping below) and put JMS properties in Application properties. For a text message put the text in AMQP value body; for a bytes message use Data body (base64) with the content type
application/octet-stream. - Receive like a JMS consumer. Start a flow for every message with the AMQP 1.0 stream source, or read a batch with Receive messages and choose the Settlement outcome (Accept (remove) takes the message off the queue, Release (redeliver) leaves it for the next reader).
| Broker | Queue | Topic |
|---|---|---|
| ActiveMQ Artemis | The address name, for example orders | The address name; read it through a queue bound to that address |
| ActiveMQ Classic | queue://orders (or just orders) | topic://prices |
| Azure Service Bus | The queue name | Send to the topic name; receive from prices/Subscriptions/my-subscription |
| RabbitMQ 4 | /queues/orders | Send to /exchanges/prices/routing-key; receive from a queue bound to that exchange |
| JMS header | AMQP 1.0 field |
|---|---|
JMSMessageID | Message id |
JMSCorrelationID | Correlation id |
JMSReplyTo | Reply-to |
JMSDeliveryMode persistent | Durable on |
JMSPriority | Priority (JMS uses 0 to 9) |
JMSExpiration | Time to live (ms) or Absolute expiry time (Unix ms) |
JMSXGroupID, JMSXGroupSeq | Group id, Group sequence |
| JMS message properties | Application properties |
A JMS application reads the message type from the x-opt-jms-msg-type message annotation. To make sure it sees a text message, add it under Message annotations as {"$amqp": "byte", "value": 5} (3 is a bytes message, 2 a map message). Received messages show every section the same way, so a flow can branch on the annotation and the application properties a JMS producer set.
Selectors and durable subscriptions. JMS message selectors are not applied by the broker for these reads. Filter in the flow with a Filter step, or let the broker route the wanted messages to their own queue. For a durable topic subscription, receive from a queue or subscription bound to the topic, as the table shows.