Viren Baraiya
126 lines
4.4 KiB
Markdown
126 lines
4.4 KiB
Markdown
# Workflow Message Queue (WMQ)
|
|
|
|
**tl;dr** — every workflow now has a queue. You can use this queue to turn your workflow into an event loop: it sits idle, waiting for messages, processes each one, then goes back to waiting.
|
|
|
|
## How it works
|
|
|
|
WMQ adds a persistent message queue to every running Conductor workflow. While the workflow is active you can push messages to it from anywhere — another service, a Kafka consumer, a webhook handler, a human — and the workflow will pick them up and act on them.
|
|
|
|
Two pieces make this work:
|
|
|
|
1. **`POST /api/workflow/{workflowId}/messages`** — an HTTP endpoint exposed by Conductor that accepts a JSON payload and enqueues it on the workflow's queue.
|
|
2. **`PULL_WORKFLOW_MESSAGES`** — a new Conductor system task that blocks until messages arrive, then completes with `output.messages` containing the batch.
|
|
|
|
## Prerequisites
|
|
|
|
WMQ is disabled by default. Enable it on the Conductor server before registering a workflow that uses `PULL_WORKFLOW_MESSAGES` or calling the push endpoint:
|
|
|
|
```properties
|
|
conductor.workflow-message-queue.enabled=true
|
|
```
|
|
|
|
When this property is `false`, Conductor does not register the system task or the HTTP endpoint; the endpoint returns `404 Not Found`.
|
|
|
|
## Using WMQ
|
|
|
|
Add a `PULL_WORKFLOW_MESSAGES` task to your workflow definition:
|
|
|
|
```json
|
|
{
|
|
"name": "wait_for_message",
|
|
"taskReferenceName": "wait_for_message_ref",
|
|
"type": "PULL_WORKFLOW_MESSAGES",
|
|
"inputParameters": {
|
|
"batchSize": 1
|
|
}
|
|
}
|
|
```
|
|
|
|
Then push to it:
|
|
|
|
```bash
|
|
curl -X POST http://localhost:8080/api/workflow/{workflowId}/messages \
|
|
-H "Content-Type: application/json" \
|
|
-d '{"text": "hello"}'
|
|
```
|
|
|
|
The task completes with:
|
|
|
|
```json
|
|
{
|
|
"messages": [
|
|
{
|
|
"id": "3f2504e0-4f89-11d3-9a0c-0305e82c3301",
|
|
"workflowId": "8e2c14e1-...",
|
|
"payload": { "text": "hello" },
|
|
"receivedAt": "2025-06-15T10:30:00Z"
|
|
}
|
|
],
|
|
"count": 1
|
|
}
|
|
```
|
|
|
|
Your workflow accesses the user data via `output.messages[0].payload`. The `id` and `receivedAt` fields are added by Conductor at ingestion time.
|
|
|
|
**Push errors:**
|
|
- `404 Not Found` — the workflow ID does not exist, or the WMQ feature is disabled.
|
|
- `409 Conflict` — workflow is not in `RUNNING` state (completed, failed, terminated, etc.). The message is not stored.
|
|
- `429 Too Many Requests` — queue is full (`maxQueueSize` reached). Caller must back off and retry.
|
|
|
|
### Event loop pattern
|
|
|
|
For workflows that process an unbounded stream of messages, wrap the task in a `DO_WHILE`:
|
|
|
|
```json
|
|
{
|
|
"name": "message_loop",
|
|
"taskReferenceName": "message_loop_ref",
|
|
"type": "DO_WHILE",
|
|
"loopCondition": "$.message_loop_ref['iteration'] < 100",
|
|
"loopOver": [
|
|
{
|
|
"name": "pull_message",
|
|
"taskReferenceName": "pull_message_ref",
|
|
"type": "PULL_WORKFLOW_MESSAGES",
|
|
"inputParameters": { "batchSize": 1 }
|
|
},
|
|
{
|
|
"name": "process_message",
|
|
"taskReferenceName": "process_message_ref",
|
|
"type": "INLINE",
|
|
"inputParameters": {
|
|
"evaluatorType": "javascript",
|
|
"expression": "function e() { return { payload: $.messages[0].payload }; } e();",
|
|
"messages": "${pull_message_ref.output.messages}"
|
|
}
|
|
}
|
|
]
|
|
}
|
|
```
|
|
|
|
The loop parks on `PULL_WORKFLOW_MESSAGES` until the next message arrives.
|
|
|
|
## Using WMQ with agents
|
|
|
|
WMQ is framework-neutral. Use `PULL_WORKFLOW_MESSAGES` in the Conductor graph to park execution until a message arrives, then pass the returned payload to the next task. For SDK-authored agents, see [Conductor Agents](../devguide/ai/conductor-agents.md) and keep framework-specific runtime code in its maintained SDK example.
|
|
|
|
### Kafka bridge example
|
|
|
|
The pattern also works as a bridge from external event streams. A Kafka consumer can translate each record into a `POST /api/workflow/{workflowId}/messages` request using the payload shape shown above. Keep that consumer implementation in its owning SDK or service repository; it is independent of the framework used by the workflow's agent steps.
|
|
|
|
## Configuration
|
|
|
|
```properties
|
|
conductor.workflow-message-queue.enabled=true
|
|
conductor.workflow-message-queue.maxQueueSize=1000
|
|
conductor.workflow-message-queue.ttlSeconds=86400
|
|
conductor.workflow-message-queue.maxBatchSize=100
|
|
```
|
|
|
|
| Property | Default | Description |
|
|
|---|---|---|
|
|
| `enabled` | `false` | Enable the WMQ feature |
|
|
| `maxQueueSize` | `1000` | Max messages queued per workflow |
|
|
| `ttlSeconds` | `86400` | Message TTL (24 h) |
|
|
| `maxBatchSize` | `100` | Max messages returned per `PULL_WORKFLOW_MESSAGES` poll |
|