Package org.acemq.amqp.core

Class Requester

java.lang.Object

org.acemq.amqp.core.Requester

All Implemented Interfaces:

AutoCloseable

public final class Requester
extends Object
implements AutoCloseable

Asks a question and waits for the answer.

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 is blocked like an HTTP client, and the failure modes are a message broker's. If the two services can talk over HTTP or gRPC, they should — those tools have timeouts, load balancing, circuit breakers and tracing that a messaging library will not match.

What it is genuinely for: reaching a service that is only on the broker — no HTTP endpoint, behind a firewall, or one worker among many where the broker is already doing the load balancing. That case is real, and doing it by hand means reply queues, correlation ids and a timeout that somebody always forgets.

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

One reply queue serves every request from this instance, and replies are matched by correlation id. A requester per connection is the intended shape; one per call would create and destroy a queue for every question asked.

Method Summary

Modifier and Type	Method	Description
void	close()
String	replyQueue()
<Q, A> A	request(String exchange, String routingKey, Q request, Class<A> responseType, Duration timeout)	Asks, and waits.
<Q, A> CompletableFuture<A>	requestAsync(String exchange, String routingKey, Q request, Class<A> responseType)	Asks, and hands back a future.
long	timedOut()
long	unmatched()

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Method Details

request

public <Q,
A> A request(String exchange,
 String routingKey,
 Q request,
 Class<A> responseType,
 Duration timeout)

Asks, and waits.

Type Parameters:

Q - request type

A - response type

Parameters:

exchange - exchange to publish the request to, or the empty string for a queue

routingKey - routing key, or the queue name when the exchange is empty

request - the request payload

responseType - type to decode the reply into

timeout - how long to wait before giving up

Returns:

the reply

Throws:

RequestTimedOutException - if no reply arrives in time

ResponderFailedException - if the responder answered with a failure

requestAsync

public <Q,
A> CompletableFuture<A> requestAsync(String exchange,
 String routingKey,
 Q request,
 Class<A> responseType)

Asks, and hands back a future.

The future never completes on its own if no reply arrives — a timeout belongs to the caller, and CompletableFuture.orTimeout(long, java.util.concurrent.TimeUnit) is the usual way to attach one.

Returns:

a future completed with the decoded reply

replyQueue

public String replyQueue()

Returns:

the queue replies arrive on; useful in a log line, and nowhere else

timedOut

public long timedOut()

Returns:

how many requests gave up waiting

unmatched

public long unmatched()

Returns:

replies that arrived with nobody waiting for them, usually after a timeout

close

public void close()

Specified by:

close in interface AutoCloseable