Skip to content

Egress RabbitMQ

Responsibility

Publishes messages to RabbitMQ for workflow service_call steps (egress.rabbitmq.publish and egress.rabbitmq.topic.send). When RABBITMQ_TRIGGER_ENABLED=true, the same image also runs an AMQP subscriber that forwards messages to the Orchestrator (used by the rabbitmq-trigger-subscriber deployment).

Trigger capabilities (registry)

The service registers these trigger capabilities (all use invocation.type: amqp in workflow YAML):

Capability Pattern See RabbitMQ tutorials
egress.rabbitmq.trigger.topic Durable queue bound to an exchange with a routing / topic pattern Routing, Topics
egress.rabbitmq.trigger.topic.dsav Topic receive filtered on required Domain / Service / Action / Version Topics + MeshFlows DSAV convention
egress.rabbitmq.trigger.queue Consume a named queue only (no exchange declare/bind in the subscriber) Hello World, Work Queues

Runtime mode is selected with RABBITMQ_TRIGGER_MODE: topic (default) or queue. Other RABBITMQ_TRIGGER_* variables configure URL, queue name, exchange, binding key, etc.

DSAV topic receive

egress.rabbitmq.trigger.topic.dsav is the designer capability for an AMQP Topic subscription trigger authored in flows/triggers/*.yaml with Domain / Service / Action / Version triggerFilters (and matching context:amqp.routing_key). Use with egress.rabbitmq.topic.send.

When the subscriber has no explicit Workflow header (and no static RABBITMQ_TRIGGER_WORKFLOW), it calls Orchestrator POST /invoke/amqp/dispatch, which starts every AMQP workflow whose triggerFilters match.

Egress capabilities (registry)

Capability Path Notes
egress.rabbitmq.publish /publish Generic exchange or queue publish
egress.rabbitmq.topic.send /publish/topic Required Domain/Service/Action/Version; forces routing key

Inbound Authentication

  • internal-only service, called by orchestrator

Outbound Authentication

  • AMQP URL credentials (user/pass in URL or secret ref)

Key Config

  • exchange, routing key, persistence settings
  • optional trigger loop: RABBITMQ_TRIGGER_MODE, RABBITMQ_TRIGGER_EXCHANGE, RABBITMQ_TRIGGER_QUEUE, RABBITMQ_TRIGGER_BINDING_KEY, RABBITMQ_TRIGGER_QUEUE_PASSIVE, …
  • dispatch path override: ORCHESTRATOR_AMQP_DISPATCH_PATH (default /invoke/amqp/dispatch)

Publish: exchange vs queue (egress.rabbitmq.publish)

Orchestrator calls POST /publish with publish_target:

  • exchange — declare the exchange and publish with a routing key (topic/direct/fanout).
  • queue — publish via the broker default exchange so the message is routed to the queue whose name equals the routing key (channel.default_exchange.publish(..., routing_key=queue_name)). This matches publishing to a named queue in the RabbitMQ tutorials without declaring a custom exchange.

Topic send with DSAV (egress.rabbitmq.topic.send)

Orchestrator calls POST /publish/topic with required domain, service, action, version, and message. The provider:

  1. Rejects incomplete DSAV (HTTP 400 / validation error).
  2. Sets AMQP headers Domain / Service / Action / Version.
  3. Forces routing_key = domain.service.action.version (lowercase).

Use this when you want MeshFlows to enforce DSAV routing end-to-end with the matching DSAV receive trigger.

Consumer ack vs “peek-lock” (trigger path)

The AMQP trigger subscriber processes one message, calls the Orchestrator synchronously, then ACKs on success or NACKs (optionally requeue) on failure. So the message is held un-acknowledged for the duration of the HTTP workflow run — similar in duration to ESB peek-lock, but it is not a separate “abandon/complete” API from inside arbitrary steps: the ack is owned by the subscriber, not by a rabbitmq_ack workflow step. Extending to deferred ack after async or multi-replica processing would need a new design (e.g. explicit delivery tokens and a callback API on the subscriber).

Health

  • GET /healthz
  • GET /readyz