The following diagram provides an overview of the scenario that we will implement:
The sender will send a message to the default exchange with a routing key that matches the name of the designated request queue. The request receiver is a subscriber to the request queue. After a request message is received, the request receiver retrieves the value of the replyTo property from the message header, creates a response message, and sends it to the default exchange with a routing key that matches the replyTo property. This means that the replyTo property points to a queue that handles response messages and the sender is subscribed to that queue in order to receive a response.
Let's extend our Sender class with the following sendRequest() method, which sends a message to the request_exchange exchange, and the receiveResponse() method, which receives a message from the response_queue queue as follows:
private static final String REQUEST_QUEUE = "request_queue";
private static final String RESPONSE_QUEUE = "response_queue";
public void sendRequest(String requestQueue, String message, String correlationId) {
try {
channel.queueDeclare(REQUEST_QUEUE, false, false, false, null);
channel.queueDeclare(RESPONSE_QUEUE, false, false, false, null);
AMQP.BasicProperties amqpProps = new AMQP.BasicProperties();
amqpProps = amqpProps.builder()
.correlationId(String.valueOf(correlationId))
.replyTo(RESPONSE_QUEUE).build();
channel.basicPublish(DEFAULT_EXCHANGE, REQUEST_QUEUE, amqpProps, message.getBytes());
} catch (IOException e) {
LOGGER.error(e.getMessage(), e);
}
}
public String waitForResponse(final String correlationId) {
QueueingConsumer consumer = new QueueingConsumer(channel);
String result = null;
try {
channel.basicConsume(RESPONSE_QUEUE, true, consumer);
QueueingConsumer.Delivery delivery = consumer.nextDelivery(3000);
String message = new String(delivery.getBody());
if (delivery.getProperties() != null) {
String msgCorrelationId = delivery.getProperties()
.getCorrelationId();
if (!correlationId.equals(msgCorrelationId)) {
LOGGER.warn("Received response of another request.");
} else {
result = message;
}
}
LOGGER.info("Message received: " + message);
} catch (IOException e) {
LOGGER.error(e.getMessage(), e);
} catch (ShutdownSignalException e) {
LOGGER.error(e.getMessage(), e);
} catch (ConsumerCancelledException e) {
LOGGER.error(e.getMessage(), e);
} catch (InterruptedException e) {
LOGGER.error(e.getMessage(), e);
}
return result;
}The sendRequest() method crafts an AMQP.BasicProperties instance and provides the replyTo and correlationId properties. The correlationId must be a unique identifier that is passed back in the response message and can be used by the sender to determine the request for which a response is received.
The RequestReceiver class provides a sample implementation of a request receiver:
public class RequestReceiver {
private static final String DEFAULT_QUEUE = "";
private static final String REQUEST_QUEUE = "request_queue";
private final static Logger LOGGER =
LoggerFactory.getLogger(Sender.class);
private Connection connection = null;
private Channel channel = null;
public void initialize() {
try {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
connection = factory.newConnection();
channel = connection.createChannel();
} catch (IOException e) {
LOGGER.error(e.getMessage(), e);
}
}
. . .
}The receive() method is used to read a request message from a queue:
public void receive() {
if (channel == null) {
initialize();
}
String message = null;
try {
channel.queueDeclare(REQUEST_QUEUE, false, false, false, null);
QueueingConsumer consumer = new QueueingConsumer(channel);
channel.basicConsume(REQUEST_QUEUE, true, consumer);
QueueingConsumer.Delivery delivery = consumer.nextDelivery();
message = new String(delivery.getBody());
LOGGER.info("Request received: " + message);
// do something with the request message ...
BasicProperties properties = delivery.getProperties();
if (properties != null) {
AMQP.BasicProperties amqpProps = new AMQP.BasicProperties();
amqpProps = amqpProps.builder().correlationId(
String.valueOf(properties.getCorrelationId())).build();
channel.basicPublish(DEFAULT_QUEUE, properties.getReplyTo(), amqpProps, "Response message.".getBytes());
} else {
LOGGER.warn("Cannot determine response destination for message.");
}
} catch (IOException e) {
LOGGER.error(e.getMessage(), e);
} catch (ShutdownSignalException e) {
LOGGER.error(e.getMessage(), e);
} catch (ConsumerCancelledException e) {
LOGGER.error(e.getMessage(), e);
} catch (InterruptedException e) {
LOGGER.error(e.getMessage(), e);
}
}And again we have a destroy() method – it is important to make sure that you close your connections to the broker if you are no longer using them:
public void destroy() {
if (connection != null) {
try {
connection.close();
} catch (IOException e) {
LOGGER.warn(e.getMessage(), e);
}
}
}
}In order to send a request message we can use the RequestSenderDemo class:
public class RequestSenderDemo {
private static final String REQUEST_QUEUE = "request_queue";
public static String sendToRequestReplyQueue() {
Sender sender = new Sender();
sender.initialize();
sender.sendRequest(REQUEST_QUEUE, "Test message.", "MSG1");
String result = sender.waitForResponse("MSG1");
sender.destroy();
return result;
}
public static void main(String[] args) {
sendToRequestReplyQueue();
}
}In order to receive the request message and send a response message, you can use the RequestReceiverDemo class:
public class RequestReceiverDemo {
public static void main(String[] args) throws InterruptedException {
final RequestReceiver receiver = new RequestReceiver();
receiver.initialize();
receiver.receive();
receiver.destroy();
}
}