`activemq://`¶

Producer + consumer for ActiveMQ brokers (both ActiveMQ Classic 5.x and ActiveMQ Artemis 2.x), speaking STOMP 1.2 via github.com/go-stomp/stomp/v3.

activemq:// is a thin alias on top of stomp:// — same wire protocol, same URI param surface, same defaults. The two schemes share every line of code; pick activemq:// when you want ActiveMQ targets to be explicit in your YAML.

Note on protocol choice. Helix speaks STOMP 1.2 rather than OpenWire (ActiveMQ Classic's native binary protocol) — Go has no first-class OpenWire client, and modern ActiveMQ Artemis ships with OpenWire disabled by default. For AMQP 1.0 use amqp://.

URI shape¶

activemq://<destination>?<params>

The destination is sent verbatim as the STOMP destination header. Common conventions:

Style	Example	What it means
JMS queue path	`activemq:///queue/orders`	Point-to-point queue.
JMS topic path	`activemq:///topic/events`	Pub/sub topic.
Bare name	`activemq://orders.in`	Broker maps to its default destination type (queue on ActiveMQ Classic).

Specify the broker URL via the brokerURL= query param (the authority before the ? is part of the destination path, NOT the broker host):

activemq:///queue/orders?brokerURL=tcp://mq:61613&login=app&passcode=…

Body shape¶

Request (producer):

Body type	Wire shape
`[]byte`	STOMP BytesMessage (`content-type` from headers, or auto-set if not present).
`string`	STOMP TextMessage.
`io.Reader`	Drained, then sent as BytesMessage.

Response (consumer — each inbound STOMP MESSAGE becomes one exchange):

Body	When
`[]byte`	Always — the STOMP body.
Headers	STOMP headers pass through subject to `headerIncludePattern` / `headerExcludePattern`. Reserved: `destination`, `message-id`, `subscription`, `reply-to`, `content-length`, `content-type`.

Operations¶

Producer-only — the wire is one-shot SEND per exchange. Consumer is a long-lived STOMP SUBSCRIBE that emits one Helix exchange per received MESSAGE.

Pattern	URI	Body in	Body out
Fire-and-forget send (default)	`to: activemq:///queue/orders`	bytes/string	unchanged
Request/reply (InOut)	`to: activemq:///queue/rpc?replyTo=/temp-queue/reply` + InOut exchange	bytes/string	reply body
Subscribe	`trigger: { uri: activemq:///queue/orders }`	n/a	bytes
Durable subscribe	`trigger: { uri: activemq:///topic/events?durable=true&subscriptionName=worker1&clientId=helix-1 }`	n/a	bytes

Headers¶

Inputs (producer SEND headers):

Header	Effect
Standard exchange `content-type`	Sent verbatim as STOMP `content-type`.
`correlation-id`	Sent verbatim. Used by InOut to match the reply.
`reply-to`	Overrides the URI `replyTo=` value for this message.
`priority`	Overrides URI `priority=` for this message.
`expires`	Overrides URI `timeToLive=` for this message (epoch-ms).
Custom keys	Pass-through subject to the include/exclude regex pair.

Outputs (consumer MESSAGE → exchange headers):

Reserved STOMP frame headers (destination, message-id, subscription, reply-to, content-length, content-type) always pass through. User headers follow headerIncludePattern / headerExcludePattern.

Parameters¶

Full URI-param surface is documented under stomp:// — activemq:// accepts the identical set. Quick reference:

Group	Keys
Connection	`brokerURL`, `brokerHost`, `brokerPort`, `host`, `login`, `passcode`, `version`, `clientId`, `heartBeat`, `connectTimeout`, `readTimeout`, `writeTimeout`, `maxReconnectAttempts`, `recoveryInterval`, `testConnectionOnStartup`
TLS	`tlsCAFile`, `tlsCertFile`, `tlsKeyFile`, `tlsInsecureSkipVerify`, `tlsMinVersion`, `tlsMaxVersion`, `tlsCipherSuites`
Common	`replyTo`, `disableReplyTo`, `requestTimeout`, `receiveTimeout`
Consumer	`ack`, `concurrentConsumers`, `selector`, `durable`, `subscriptionName`, `prefetch`
Producer	`persistent`, `priority`, `timeToLive`, `allowNullBody`, `useMessageIDAsCorrelationID`
Headers	`header.<NAME>`, `headerIncludePattern`, `headerExcludePattern`

Examples¶

Producer — fan a Kafka topic into an ActiveMQ queue¶

workflows:
  - id: kafka-to-activemq
    trigger:
      uri: kafka://orders.kafka?bootstrapServers=kafka:9092&groupId=mq-fanout
    steps:
      - target: "activemq:///queue/orders?brokerURL=tcp://mq:61613&persistent=true&priority=4"

Body is the Kafka value; STOMP persistent flag makes the broker journal the message before ack-ing the SEND.

Consumer — durable topic subscriber¶

workflows:
  - id: events-listener
    trigger:
      uri: activemq:///topic/events
      params:
        brokerURL: "tcp://mq:61613"
        clientId: "helix-events"
        durable: "true"
        subscriptionName: "events-helix-1"
        ack: "client"
        prefetch: "100"
    steps:
      - target: "log://event"

durable=true + subscriptionName= + clientId= together create a named durable subscription that survives broker restarts. ack=client means Helix sends an ACK frame only after the route returns successfully — failures cause redelivery.

Request / reply over a temporary queue¶

workflows:
  - id: rpc-caller
    trigger:
      uri: http-server://0.0.0.0:8080/quote
    steps:
      # InOut over activemq:// — Helix opens a temp reply queue, sends,
      # blocks for the matching correlation-id, returns the reply body.
      - target: "activemq:///queue/quote.req?brokerURL=tcp://mq:61613&replyTo=/temp-queue/quote.reply&requestTimeout=10s"

Selector — broker-side filter¶

trigger:
  uri: "activemq:///queue/orders?brokerURL=tcp://mq:61613&selector=region%3D%27EU%27"

The selector URI param maps directly to STOMP's selector SUBSCRIBE header. JMS selector syntax — broker-dependent feature support (Artemis: full SQL92 subset; ActiveMQ Classic: simpler).

Failure modes¶

Error	Cause	Fix
`activemq testConnectionOnStartup: …`	`testConnectionOnStartup=true` and broker rejected the CONNECT (bad auth, wrong port, TLS handshake fail).	Check `brokerURL`, `login`, `passcode`, TLS params.
`dial tcp …: connection refused`	Broker not running, wrong port, firewall.	Default STOMP port is 61613 (61612 for ssl://).
`CONNECT: invalid credentials`	login/passcode wrong.	Verify broker user.
`SUBSCRIBE: unknown destination` (Artemis)	Auto-create disabled and the address doesn't exist.	Pre-create on the broker, or enable auto-create in `broker.xml`.
Producer hangs at SEND	Broker is back-pressured (queue full) and Helix has `persistent=true`.	Raise broker quotas, lower message rate, or use `persistent=false` for non-critical data.
`durable=true requires subscriptionName`	Validation.	Set both `subscriptionName=` and `clientId=`.

When to pick `activemq://` vs `stomp://` vs `amqp://`¶

Use	Component
Broker is ActiveMQ (Classic or Artemis), and you want this to be explicit in YAML	`activemq://`
Broker is RabbitMQ-with-STOMP-plugin or HornetQ, and you want broker-agnostic config	`stomp://`
Broker is Artemis (or any AMQP 1.0 broker) and you want better message tracing / interop	`amqp://`
Broker is RabbitMQ-with-AMQP-0-9-1 (the default RabbitMQ protocol)	`rabbitmq://`

activemq://¶