Implementing RPC Pattern

1. Creating Reply Queue

Property	Description	Default Value
reply_to	Specifies the queue name where the response should be sent.	Empty string
correlation_id	Unique identifier to match requests with responses.	Generated UUID

Example: Setting up reply queue in RabbitMQ

const amqp = require('amqplib');

async function setupReplyQueue() {
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createChannel();

  // Declare a queue for replies
  const replyQueue = 'rpc_reply_queue';
  await channel.assertQueue(replyQueue, { durable: false });

  // Set up consumer for reply queue
  channel.consume(replyQueue, (msg) => {
    if (msg && msg.properties.correlationId === correlationId) {
      console.log('Received response:', msg.content.toString());
    }
  }, { noAck: true });

  return channel;
}

setupReplyQueue().catch(console.error);

2. Setting Correlation ID

Property	Description	Value
correlation_id	Unique identifier for the request-response cycle	`string`

Example: Setting Correlation ID in RPC Request

const amqp = require('amqplib');

async function setupRPCClient() {
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createChannel();

  const queueName = 'rpc_queue';
  const correlationId = generateUuid();

  const msg = {
    correlation_id: correlationId,
    reply_to: 'rpc_client_queue',
    data: 'some_data'
  };

  channel.sendToQueue(queueName, Buffer.from(JSON.stringify(msg)), {
    correlationId: correlationId,
    replyTo: 'rpc_client_queue'
  });

  return channel;
}

function generateUuid() {
  return Math.random().toString(36).substring(2, 15) + Math.random().toString(36).substring(2, 15);
}

3. Using Reply-To Property

Property	Value	Description
reply_to	`string`	The queue name to which the reply should be sent. If not set, no reply is expected.

Example: Setting Reply-To Property

const amqp = require('amqplib');

async function setupRPCClient() {
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createChannel();

  // Declare a queue for replies
  await channel.assertQueue('rpc_queue', { durable: false });
  
  // Set up a callback to handle replies
  channel.consume('rpc_queue', (msg) => {
    console.log('Received reply:', msg.content.toString());
  }, { noAck: true });

  // Send a message with reply_to set
  const correlationId = 'req-123';
  channel.sendToQueue('rpc_queue', Buffer.from('Hello World'), {
    correlationId: correlationId,
    replyTo: 'rpc_queue'
  });
}

setupRPCClient().catch(console.error);

Note: The reply_to property must be set to a valid queue name where the client expects to receive the response.

4. Implementing RPC Server

Example: Basic RPC Server Implementation

const amqp = require('amqplib');

async function startRpcServer() {
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createChannel();

  const queueName = 'rpc_queue';
  await channel.assertQueue(queueName, { durable: false });
  
  // Consume messages from the queue
  channel.consume(queueName, async (msg) => {
    const response = await processRequest(msg.content.toString());
    
    // Send response back to the replyTo queue
    channel.sendToQueue(
      msg.properties.replyTo,
      Buffer.from(response),
      { correlationId: msg.properties.correlationId }
    );
    
    // Acknowledge the message
    channel.ack(msg);
  }, { noAck: false });
}

async function processRequest(data) {
  // Simulate some processing
  const result = data.split('').reverse().join('');
  return result;
}

startRpcServer().catch(console.error);

Property	Description	Values
queueName	Name of the queue to listen for RPC requests	`'rpc_queue'`
replyTo	Queue name to send the response to	`msg.properties.replyTo`
correlationId	Unique identifier to match request and response	`msg.properties.correlationId`

Note: The replyTo and correlationId properties are essential for the RPC pattern to work correctly.

Warning: Ensure that the RPC server acknowledges messages to prevent message loss.

RPC Server Workflow

Connect to RabbitMQ broker
Declare a queue for RPC requests
Consume messages from the queue
Process the request and generate a response
Send the response to the replyTo queue with matching correlationId
Acknowledge the received message

5. Implementing RPC Client

Property	Description	Default Value
reply_to	Queue name to which the server should send the response	Generated unique queue name
correlation_id	Unique identifier to match request with response	Generated UUID

Example: Basic RPC Client Implementation

const amqp = require('amqplib');

async function rpcClient() {
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createChannel();
  const queue = 'rpc_queue';

  // Declare the queue for RPC responses
  await channel.assertQueue('', { exclusive: true });
  const replyQueue = channel.currentQueue;

  // Set up correlation ID and response handler
  const correlationId = generateUuid();
  const responses = new Map();

  channel.consume(replyQueue, (msg) => {
    if (msg.properties.correlationId === correlationId) {
      responses.set('result', msg.content.toString());
    }
  }, { noAck: true });

  // Send the RPC request
  channel.sendToQueue(queue, Buffer.from('Hello World'), {
    reply_to: replyQueue,
    correlation_id: correlationId
  });

  // Wait for response
  while (!responses.has('result')) {
    await new Promise(resolve => setTimeout(resolve, 100));
  }

  console.log('Result:', responses.get('result'));
  await channel.close();
  await connection.close();
}

function generateUuid() {
  return Math.random().toString(36).substring(2, 15) +
         Math.random().toString(36).substring(2, 15);
}

Note: The client must listen for responses on a unique queue to avoid mixing up replies from different requests.

6. Using Direct Reply-To

Property	Description	Value
reply_to	Specifies the queue to which the reply should be sent	`direct_reply_to`

Example: Publishing RPC Request with Direct Reply-To

const amqp = require('amqplib');

async function rpcClient() {
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createChannel();

  const queue = 'rpc_queue';
  const msg = {num: 5};

  // Set reply_to to direct_reply_to for automatic reply queue creation
  const options = {
    replyTo: 'direct_reply_to',
    correlationId: '123'
  };

  channel.sendToQueue(queue, Buffer.from(JSON.stringify(msg)), options);

  // Listen for the reply on the auto-generated queue
  channel.consume('direct_reply_to', (msg) => {
    console.log('Received reply:', msg.content.toString());
    channel.close();
    connection.close();
  }, {noAck: true});
}

rpcClient();

Note: The direct_reply_to feature is available in RabbitMQ 3.8+ and requires the rabbitmq_direct_reply_to plugin to be enabled.

Warning: Using direct_reply_to can lead to resource exhaustion if not handled properly. Ensure that the client consumes replies promptly to avoid queue buildup.

7. Handling RPC Timeouts

Property	Value	Description
rpc_timeout	`30000`	Timeout in milliseconds for RPC calls
retry_attempts	`3`	Number of retry attempts before failing

Example: Configuring RPC Timeout

const amqp = require('amqplib');

async function setupRPC() {
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createChannel();

  // Set RPC timeout to 30 seconds
  const rpcTimeout = 30000;
  
  // Set up a queue for RPC responses
  await channel.assertQueue('rpc_queue', { durable: false });
  
  // Handle RPC requests
  channel.consume('rpc_queue', (msg) => {
    const response = processRequest(msg.content.toString());
    
    // Send response back with timeout handling
    channel.sendToQueue(
      msg.properties.replyTo,
      Buffer.from(response),
      { correlationId: msg.properties.correlationId }
    );
    
    channel.ack(msg);
  }, { noAck: false });
}

function processRequest(data) {
  // Simulate processing time
  return JSON.stringify({ result: 'success', data });
}

Note: If the RPC call exceeds the timeout, the client should handle the error gracefully and possibly retry the request.

Warning: Setting a timeout too low may cause premature failures during long-running operations.

8. Implementing Async RPC

Asynchronous Remote Procedure Call (RPC) in RabbitMQ allows clients to send requests and receive responses without blocking, using message queues for communication.

Component	Description	Notes
Request Queue	Queue where client sends RPC requests	Named using a unique identifier or UUID
Reply Queue	Queue where client listens for responses	Generated dynamically by the client (e.g., auto-named)
Correlation ID	Unique identifier to match request and response	Set in the `correlation_id` field of the message properties
RPC Server	Service that processes requests and sends responses	Consumes from request queue, publishes to reply queue

Example: Client-side Async RPC

const amqp = require('amqplib');
const uuid = require('uuid');

async function rpcClient() {
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createChannel();

  const queueName = 'rpc_queue';
  const replyQueue = 'reply_' + uuid.v4();

  await channel.assertQueue(replyQueue, { durable: false });
  await channel.assertQueue(queueName, { durable: true });

  const correlationId = uuid.v4();

  const msg = {
    operation: 'add',
    params: [2, 3]
  };

  channel.sendToQueue(queueName, Buffer.from(JSON.stringify(msg)), {
    correlationId: correlationId,
    replyTo: replyQueue
  });

  const response = await new Promise((resolve, reject) => {
    channel.consume(replyQueue, (msg) => {
      if (msg.properties.correlationId === correlationId) {
        resolve(JSON.parse(msg.content.toString()));
      }
    }, { noAck: true });
  });

  console.log('Result:', response);
}

Example: Server-side Async RPC

const amqp = require('amqplib');

async function rpcServer() {
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createChannel();

  const queueName = 'rpc_queue';

  await channel.assertQueue(queueName, { durable: true });
  await channel.prefetch(1);

  channel.consume(queueName, async (msg) => {
    const data = JSON.parse(msg.content.toString());
    const result = await performOperation(data.operation, data.params);

    channel.sendToQueue(
      msg.properties.replyTo,
      Buffer.from(JSON.stringify(result)),
      { correlationId: msg.properties.correlationId }
    );

    channel.ack(msg);
  });
}

async function performOperation(operation, params) {
  if (operation === 'add') return params[0] + params[1];
  throw new Error('Unknown operation');
}

Note: The correlationId ensures that the client correctly matches responses to their respective requests, even when multiple concurrent RPC calls are made.

9. Handling RPC Failures

Failure Type	Description	Handling Strategy	Recovery Options
Timeouts	Request exceeds configured timeout	Client retries with exponential backoff	`retry` policy, `dead-letter` exchange
No Reply	Server does not respond within expected time	Client assumes failure and logs error	Requeue message, send to `dead-letter` queue
Server Crash	RPC server process crashes or becomes unresponsive	Client detects connection loss and retries	Use `publisher confirms`, `message persistence`
Invalid Request	Server receives malformed or invalid payload	Server sends error response to client	Client validates input before sending, use `schema validation`

Example: Client-side timeout handling

const amqp = require('amqplib');

async function rpcClient() {
  const conn = await amqp.connect('amqp://localhost');
  const ch = await conn.createChannel();

  const queue = 'rpc_queue';
  const msg = { operation: 'add', params: [2, 3] };

  // Set timeout for RPC call
  const timeout = 5000;

  const response = await ch.request(queue, msg, { timeout });
  
  console.log('Result:', response);
}

Example: Server-side error handling

const amqp = require('amqplib');

async function rpcServer() {
  const conn = await amqp.connect('amqp://localhost');
  const ch = await conn.createChannel();

  const queue = 'rpc_queue';

  ch.consume(queue, async (msg) => {
    try {
      const { operation, params } = JSON.parse(msg.content.toString());
      
      if (operation === 'add') {
        const result = params.reduce((a, b) => a + b, 0);
        ch.reply(msg, { result });
      } else {
        throw new Error('Unsupported operation');
      }
    } catch (err) {
      // Send error response back to client
      ch.reply(msg, { error: err.message });
    }
  }, { noAck: true });
}

Note: Always implement proper error handling and logging for RPC failures to ensure system reliability.

10. Monitoring RPC Performance

Example: RPC Server Metrics

const amqp = require('amqplib');
const client = require('prom-client');

const register = new client.Registry();
const rpcDuration = new client.Histogram({
  name: 'rpc_duration_seconds',
  help: 'Duration of RPC calls in seconds',
  labelNames: ['method'],
  buckets: [0.01, 0.05, 0.1, 0.5, 1, 2, 5]
});
register.registerMetric(rpcDuration);

async function startRPCServer() {
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createChannel();
  const queue = 'rpc_queue';

  await channel.assertQueue(queue, { durable: false });
  channel.prefetch(1);
  channel.consume(queue, async (msg) => {
    const method = msg.properties.headers.method || 'unknown';
    const start = Date.now();
    const response = await handleRPCRequest(msg.content.toString());
    const duration = (Date.now() - start) / 1000;
    rpcDuration.observe({ method }, duration);
    channel.sendToQueue(msg.properties.replyTo, Buffer.from(response), {
      correlationId: msg.properties.correlationId
    });
    channel.ack(msg);
  });
}

Metrics	Description	Collection Method
rpc_duration_seconds	Duration of RPC calls in seconds	Prometheus Histogram
rpc_requests_total	Total number of RPC requests handled	Prometheus Counter
rpc_errors_total	Total number of failed RPC requests	Prometheus Counter

Note: Use Prometheus client libraries to instrument your RPC services and expose metrics via a /metrics endpoint.

Example: RPC Client Metrics

const client = require('prom-client');

const rpcClientDuration = new client.Histogram({
  name: 'rpc_client_duration_seconds',
  help: 'Duration of RPC client calls in seconds',
  labelNames: ['method'],
  buckets: [0.01, 0.05, 0.1, 0.5, 1, 2, 5]
});
const register = new client.Registry();
register.registerMetric(rpcClientDuration);

async function callRPC(method, payload) {
  const start = Date.now();
  const response = await sendRequest(method, payload);
  const duration = (Date.now() - start) / 1000;
  rpcClientDuration.observe({ method }, duration);
  return response;
}

Key Performance Indicators (KPIs)

RPC Latency: Average time for an RPC call to complete.
RPC Throughput: Number of RPC calls processed per second.
Error Rate: Percentage of failed RPC calls.