Skip to content

Event-driven recipes

Read Event orchestration first for action support, provider configuration, delivery, and idempotency semantics.

Publish an internal event

{
  "name": "publish_order_event",
  "version": 1,
  "schemaVersion": 2,
  "inputParameters": ["orderId", "status"],
  "tasks": [
    {
      "name": "publish_order_status",
      "taskReferenceName": "publish_order_status",
      "type": "EVENT",
      "sink": "conductor:order-status",
      "inputParameters": {
        "orderId": "${workflow.input.orderId}",
        "status": "${workflow.input.status}",
        "eventVersion": 1
      }
    }
  ]
}

Register and run the workflow. Its conductor:order-status sink expands to conductor:publish_order_event:order-status.

Start a workflow from the event

Register the target workflow first:

{
  "name": "fulfill_order",
  "version": 1,
  "schemaVersion": 2,
  "inputParameters": ["orderId", "sourceEventId"],
  "tasks": [
    {
      "name": "record_fulfillment_start",
      "taskReferenceName": "record_fulfillment_start",
      "type": "SET_VARIABLE",
      "inputParameters": {
        "orderId": "${workflow.input.orderId}",
        "sourceEventId": "${workflow.input.sourceEventId}"
      }
    }
  ],
  "outputParameters": {
    "orderId": "${workflow.input.orderId}"
  }
}

{
  "name": "start_fulfillment_on_order_ready",
  "event": "conductor:publish_order_event:order-status",
  "condition": "$.status == 'READY'",
  "actions": [
    {
      "action": "start_workflow",
      "start_workflow": {
        "name": "fulfill_order",
        "version": 1,
        "correlationId": "${orderId}",
        "input": {
          "orderId": "${orderId}",
          "sourceEventId": "${workflowInstanceId}"
        }
      }
    }
  ],
  "active": true
}

curl -sS -X POST '<YOUR-CLUSTER-URL>/api/event' \
  -H 'Content-Type: application/json' \
  --data-binary @docs/devguide/cookbook/examples/events/start-workflow-handler.json

The payload expression is rooted directly at the Event task's published JSON.

Wait for an external approval

Workflow:

{
  "name": "wait_for_order_approval",
  "version": 1,
  "schemaVersion": 2,
  "inputParameters": ["orderId"],
  "tasks": [
    {
      "name": "wait_for_approval",
      "taskReferenceName": "approval",
      "type": "WAIT"
    }
  ],
  "outputParameters": {
    "approval": "${approval.output}"
  }
}

Handler:

{
  "name": "complete_order_approval",
  "event": "kafka:order-approvals",
  "condition": "$.approved == true",
  "actions": [
    {
      "action": "complete_task",
      "complete_task": {
        "workflowId": "${workflowId}",
        "taskRefName": "approval",
        "output": {
          "approved": "${approved}",
          "approvedBy": "${approvedBy}",
          "eventId": "${eventId}"
        }
      }
    }
  ],
  "active": true
}

Representative broker payload:

{
  "eventId": "approval-7f3d",
  "workflowId": "6f3f6db1-2b5f-4b34-a145-82b7ae814e91",
  "approved": true,
  "approvedBy": "reviewer@example.com"
}

Replace the representative workflowId with the ID returned when the waiting workflow starts. A correlation ID alone cannot target the WAIT task.

Use an external provider

Change event/sink to a registered provider identifier and its provider-specific URI, for example kafka:order-approvals, sqs:https://sqs.us-east-1.amazonaws.com/123/order-events, nats:orders.ready, jsm:orders.ready, nats_stream:orders.ready, amqp_queue:orders, or amqp_exchange:orders. Enable the matching module and properties described in the guide.