RabbitMQ(8)Java Client - Remote Procedure Call(RPC)

RabbitMQ(8)Java Client - Remote Procedure Call(RPC)

We need to run a function on a remote computer and wait for the result. This pattern is commonly known as Remote Procedure Call or RPC.

Callback Queue
In order to receive a response we need to send a 'callback' queue address with the request.
callbackQueueName = channel.queueDeclare().getQueue();
BasicProperties props = new BasicProperties
                            .Builder()
                            .replyTo(callbackQueueName)
                            .build();
channel.basicPublish("", "rpc_queue", props, message.getBytes());

Send callback queue via properties.

Correlation Id
It is inefficient to create a callback queue for every RPC request. We will create a single callback queue per client.

CorrelationId, we are going to set it to a unique value for every request. Later, when we receive a message in the callback queue we'll look at this property, and based on that we will be able to match a response with a request.

Summary
Client  ------> rpc_queue --------> Server
         <-----  reply_to <----------- 

Putting it all together

Server Side:
package com.sillycat.easytalker.rabbitmq.rpc;

import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;

public class RPCServer {

private static final String RPC_QUEUE_NAME = "rpc_queue";

private final static String SERVER_HOST = "www.neptune.com";

private static int fib(int n) {
if (n == 0){
return 0;
}
if (n == 1){
return 1;
}
return fib(n - 1) + fib(n - 2);
}

public static void main(String[] argv) {
Connection connection = null;
Channel channel = null;
try {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost(SERVER_HOST);
connection = factory.newConnection();
channel = connection.createChannel();

channel.queueDeclare(RPC_QUEUE_NAME, false, false, false, null);
channel.basicQos(1);

QueueingConsumer consumer = new QueueingConsumer(channel);
channel.basicConsume(RPC_QUEUE_NAME, false, consumer);

System.out.println(" [x] Awaiting RPC requests");
while (true) {
String response = null;
QueueingConsumer.Delivery delivery = consumer.nextDelivery();
BasicProperties props = delivery.getProperties();
BasicProperties replyProps = new BasicProperties.Builder()
.correlationId(props.getCorrelationId()).build();
try {
String message = new String(delivery.getBody(), "UTF-8");
int n = Integer.parseInt(message);
System.out.println(" [.] fib(" + message + ")");
response = "" + fib(n);
} catch (Exception e) {
System.out.println(" [.] " + e.toString());
response = "";
} finally {
//send back the result
channel.basicPublish("", props.getReplyTo(), replyProps,
response.getBytes("UTF-8"));
channel.basicAck(delivery.getEnvelope().getDeliveryTag(),
false);
}
}
} catch (Exception e) {
e.printStackTrace();
} finally {
if (connection != null) {
try {
connection.close();
} catch (Exception ignore) {
}
}
}
}
}

Client Side
package com.sillycat.easytalker.rabbitmq.rpc;

import java.util.UUID;

import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;

public class RPCClient1 {

private Connection connection;
private Channel channel;

private String requestQueueName = "rpc_queue";

private String replyQueueName;
private QueueingConsumer consumer;

private final static String SERVER_HOST = "www.neptune.com";

public RPCClient1() throws Exception {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost(SERVER_HOST);
connection = factory.newConnection();
channel = connection.createChannel();

replyQueueName = channel.queueDeclare().getQueue();

consumer = new QueueingConsumer(channel);
channel.basicConsume(replyQueueName, true, consumer);
}

public String call(String message) throws Exception {
String response = null;
String corrId = UUID.randomUUID().toString();
BasicProperties props = new BasicProperties.Builder()
.correlationId(corrId).replyTo(replyQueueName).build();
channel.basicPublish("", requestQueueName, props, message.getBytes());

while (true) {
QueueingConsumer.Delivery delivery = consumer.nextDelivery();
if (delivery.getProperties().getCorrelationId().equals(corrId)) {
//if the id is the same, we will get the response
response = new String(delivery.getBody(), "UTF-8");
break;
}
}
return response;
}

public void close() throws Exception {
connection.close();
}

public static void main(String[] argv) {
RPCClient1 fibonacciRpc = null;
String response = null;
try {
fibonacciRpc = new RPCClient1();
System.out.println(" [x] Requesting fib(31)");
response = fibonacciRpc.call("31");
System.out.println(" [.] Got '" + response + "'");
} catch (Exception e) {
e.printStackTrace();
} finally {
if (fibonacciRpc != null) {
try {
fibonacciRpc.close();
} catch (Exception ignore) {
}
}
}
}
}

references:
http://www.rabbitmq.com/tutorials/tutorial-six-java.html