Request and reply

try (Responder responder = mq.respond("pricing", Quote.class, quote -> price(quote));
        Requester requester = mq.requester()) {

    Price price = requester.request("", "pricing", quote, Price.class, Duration.ofSeconds(5));
}

Read this before using it

Request/reply over a broker is synchronous calling wearing asynchronous clothes, and it inherits the worst of both. The caller blocks like an HTTP client, and the failure modes are a message broker's.

If the two services can speak HTTP or gRPC, they should. Those tools have timeouts, retries, load balancing, circuit breakers, health checks and tracing that a messaging library will not match, and everyone already knows how to operate them.

What this is genuinely for:

• the callee is reachable only on the broker — no HTTP endpoint, behind a firewall, in a network you do not control;
• one worker among many, where the broker is already doing the load balancing;
• a system that is already message-driven and where adding an HTTP hop would mean a second failure domain.

Those cases are real. Doing them by hand means reply queues, correlation ids, and a timeout somebody always forgets.

How it works

The request names its reply queue twice — AMQP's own reply-to property and an acemq-reply-to header, always the same value — and carries a correlation id. The responder publishes the answer to that queue with the same id, and the requester matches it to the caller waiting for it.

A responder reads the header first and the property second. That order is the contract in all five libraries and it is not arbitrary. It comes from a split that is now closed: Go, Python and Ruby once wrote only the header, Java and .NET only the property, and a requester in one family could not be answered by a responder in the other. Since 0.5 all five write both — verified in the source rather than assumed. Writing both and reading either is what lets a requester in one language be answered by a responder in another, in both directions and whichever of the two is older. The header is preferred because it survives a hop that rebuilds the message — a retry rung, a dead-letter, a shovel — where the property does not.

Neither name is in the reserved x-acemq-* namespace, deliberately. The property is a real AMQP property, so a service that never heard of this library can answer a caller that uses it; the header is an ordinary application header, so it reaches a handler rather than being stripped on the way in.

One reply queue serves every request from a Requester. Build one per connection — one per call creates and destroys a queue for every question you ask.

Timeouts

requester.request("", "pricing", quote, Price.class, Duration.ofSeconds(5));

A timeout throws RequestTimedOutException, and its message says the thing that matters:

The request may still be queued, being handled, or already done with the reply lost on the way back.

A timeout is not a failure to handle the request. Retrying is a decision about whether the responder is idempotent, not a reflex. Where the work is not idempotent — taking money, sending an email — a timeout is a question for a human or for a shared idempotency store, not for a retry loop.

requestAsync hands back a CompletableFuture with no timeout attached, because the timeout belongs to the caller. orTimeout is the usual way to add one.

When the responder fails

A responder whose handler throws answers anyway: the reply carries the failure in the acemq-error header, as ExceptionType: message, with an empty body, and the request is then rejected to its dead-letter queue rather than retried. The caller gets ResponderFailedException at once, with the responder's text in failure(), instead of waiting out its timeout. The header is checked before the body is decoded, so a failure from a Go, Python or Ruby responder reads the same way whatever its body holds.

Since 0.7 a Java responder and requester both speak this; Go, Python and Ruby always have. Paired with an older Java peer, a failed request still ends the old way: an older responder sends no reply, so the caller times out; an older requester cannot decode the empty body and raises AceFatalException.

The numbers worth graphing

unmatched() rising while timedOut() rises is the signature of a responder that is slower than callers expect. Nothing is broken; the timeout is wrong.

What the counters promise

A responder reports two numbers, and both are safe to read the instant a round trip returns:

responder.answered()      // requests answered, counted before the reply left
responder.unanswerable()  // requests that named nowhere to reply

answered() is incremented before the reply is published, so a caller holding its answer can rely on the count already including it. The other order looks more natural and is wrong: it leaves a window where the reply is in the caller's hands and the responder still says nothing has been answered, which is a monitoring dashboard reporting an idle service that is demonstrably working. A publish that fails takes its increment back, so this counts replies that were sent rather than replies that were attempted.

The counters exist before the responder subscribes, so a request the broker hands over during start-up — what a queue with a backlog looks like from in here — is counted like any other. Neither number needs a wait before it can be trusted, and code that sleeps before reading one is working around a defect that is fixed.

Two libraries promise this, not five. Java and .NET expose answered and unanswerable with the ordering described above, and they promise it identically. Go, Python and Ruby expose neither number: a responder there counts nothing. Ruby goes as far as defining answered and timed_out as telemetry outcome names, and nothing writes them — a constant no code reaches reads as a supported feature. AceHeaders.CLAIM was the same trap here until it became a real envelope field; this one is still open.

Until that is closed, a dashboard that has to cover all five reads the responder's consume metrics instead — the queue is an ordinary queue and acemq.messages.consumed.total counts it — and the request span carries the round trip. Neither is as direct as the counters, and both work everywhere.

Concurrency

mq.respond("pricing", Quote.class, ConsumerOptions.prefetch(10).concurrency(4), this::price)

A responder handles one request at a time by default, and every caller behind a slow one is blocked. This is the setting to reach for first when request/reply feels slow.

Cleaning up

The reply queue is deleted when the requester closes, and carries x-expires so the broker removes it if the process dies first. Without that, a service that restarts often leaves thousands of orphaned queues — a real way to run a broker out of memory with nothing obviously wrong.

Closing a Responder drains it: a request being answered right now has a caller blocked on the other side, and dropping it turns their call into a timeout.

Related

• Publishing
• Consuming
• Reliability — idempotency, which decides whether a timeout can be retried