Implementing RPC client and server in RabbitMQ - Mini Project: Build & Apply

Implementing RPC client and server

📖 Scenario: You are building a simple Remote Procedure Call (RPC) system using RabbitMQ. This system allows a client to send a request to a server to calculate the square of a number. The server receives the request, processes it, and sends back the result.This is like ordering a coffee at a cafe: you tell the barista your order (request), they prepare it (process), and then give it back to you (response).

🎯 Goal: Build a basic RPC client and server using RabbitMQ in Python. The client will send a number, and the server will return the square of that number.

📋 What You'll Learn

Create a RabbitMQ connection and channel

Set up a queue for RPC requests

Implement the server to listen for requests and send back results

Implement the client to send requests and wait for responses

💡 Why This Matters

🌍 Real World

RPC systems are used in microservices to allow different services to communicate and request actions from each other asynchronously.

💼 Career

Understanding how to implement RPC with message brokers like RabbitMQ is valuable for backend developers and DevOps engineers working with distributed systems.

1

Setup RabbitMQ connection and declare RPC queue

Create a RabbitMQ connection using pika.BlockingConnection with default parameters. Then create a channel from this connection and declare a queue named rpc_queue with durable=True.

RabbitMQ

# Your code here

Hint

Use pika.ConnectionParameters('localhost') to connect to RabbitMQ on your local machine.

2

Define the server callback to process RPC requests

Define a function called on_request that takes ch, method, props, and body as parameters. Inside, convert body to an integer, calculate its square, and publish the result back to the props.reply_to queue with the same correlation_id. Finally, acknowledge the message with ch.basic_ack.

RabbitMQ

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.queue_declare(queue='rpc_queue', durable=True)

# Define the on_request callback function here
# Your code here

Hint

Use props.reply_to to send the response back to the client queue.

3

Start consuming RPC requests on the server

Use channel.basic_qos(prefetch_count=1) to limit unacknowledged messages. Then use channel.basic_consume with queue='rpc_queue' and on_request as the callback. Finally, start consuming with channel.start_consuming().

RabbitMQ

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.queue_declare(queue='rpc_queue', durable=True)

def on_request(ch, method, props, body):
    n = int(body)
    response = n * n

    ch.basic_publish(exchange='',
                     routing_key=props.reply_to,
                     properties=pika.BasicProperties(correlation_id=props.correlation_id),
                     body=str(response))
    ch.basic_ack(delivery_tag=method.delivery_tag)

# Set QoS and start consuming
# Your code here

Hint

Use basic_qos to process one message at a time.

4

Implement the RPC client to send request and receive response

Create a class RpcClient with a constructor that connects to RabbitMQ and declares a unique callback queue. Implement a method call that sends a number as a request to rpc_queue with a unique correlation_id and waits for the response on the callback queue. Print the response received from the server for the number 7.

RabbitMQ

# Your code here

Hint

Use uuid.uuid4() to generate a unique correlation ID.

Use process_data_events() to wait for the server response.

Practice

1. What is the main role of the RPC server in RabbitMQ RPC communication?

easy

A. To store messages permanently without processing

B. To send requests and wait for responses

C. To receive requests, process them, and send back responses

D. To monitor network traffic between clients

Solution

1. Step 1: Understand RPC server role

   The RPC server listens for requests on a queue and processes them.

2. Step 2: Identify server response behavior

   After processing, the server sends the result back to the client via a reply queue.

3. Final Answer:

   To receive requests, process them, and send back responses -> Option C

4. Quick Check:

   RPC server handles requests and replies [OK]

Hint: Server processes requests and replies, client sends requests [OK]

Common Mistakes:

• Confusing client and server roles
• Thinking server only stores messages
• Assuming server sends requests

2. Which of the following is the correct way to declare a callback queue for RPC replies in Python pika client?

easy

A. channel.queue_declare(queue='rpc_queue')

B. result = channel.queue_declare(queue='', exclusive=True)

C. channel.basic_publish(exchange='', routing_key='rpc_queue')

D. channel.basic_consume(queue='rpc_queue', on_message_callback=callback)

Solution

1. Step 1: Identify callback queue declaration

   For RPC replies, a temporary exclusive queue is declared with empty name and exclusive=True.

2. Step 2: Match syntax with pika usage

   Using queue_declare with queue='' and exclusive=True creates a unique callback queue.

3. Final Answer:

   result = channel.queue_declare(queue='', exclusive=True) -> Option B

4. Quick Check:

   Exclusive empty queue for callback = result = channel.queue_declare(queue='', exclusive=True) [OK]

Hint: Use empty queue name with exclusive=True for callback queue [OK]

Common Mistakes:

• Declaring fixed queue name for callback
• Using basic_publish instead of queue_declare
• Not setting exclusive=True for callback queue

3. Given this Python RPC client snippet, what will be printed?

corr_id = str(uuid.uuid4())
channel.basic_publish(
    exchange='',
    routing_key='rpc_queue',
    properties=pika.BasicProperties(
        reply_to=callback_queue,
        correlation_id=corr_id
    ),
    body='5'
)
print('Sent request with correlation_id:', corr_id)

medium

A. Sent request with correlation_id:

B. Sent request with correlation_id: rpc_queue

C. Sent request with correlation_id: 5

D. Error: correlation_id not defined

Solution

1. Step 1: Understand correlation_id assignment

   corr_id is assigned a new UUID string using uuid.uuid4(), which is random.

2. Step 2: Analyze print statement output

   The print outputs the fixed string plus the generated corr_id value.

3. Final Answer:

   Sent request with correlation_id: <random UUID> -> Option A

4. Quick Check:

   corr_id is UUID string printed [OK]

Hint: corr_id is a UUID string, printed as is [OK]

Common Mistakes:

• Confusing correlation_id with routing_key
• Printing message body instead of corr_id
• Assuming correlation_id is undefined

4. In this RPC server code snippet, what is the main error?

def on_request(ch, method, props, body):
    n = int(body)
    response = fib(n)
    ch.basic_publish(
        exchange='',
        routing_key=props.reply_to,
        properties=pika.BasicProperties(correlation_id=props.correlation_id),
        body=str(response)
    )
    ch.basic_ack(delivery_tag=method.delivery_tag)

channel.basic_consume(queue='rpc_queue', on_message_callback=on_request)

medium

A. Missing call to start consuming messages

B. Incorrect use of correlation_id property

C. No acknowledgment of message delivery

D. Using wrong queue name in basic_consume

Solution

1. Step 1: Check message consumption setup

   basic_consume is set but start_consuming() is not called to begin processing.

2. Step 2: Verify other parts

   Correlation_id usage and ack are correct; queue name matches.

3. Final Answer:

   Missing call to start consuming messages -> Option A

4. Quick Check:

   start_consuming() needed to process messages [OK]

Hint: Always call start_consuming() after basic_consume [OK]

Common Mistakes:

• Forgetting start_consuming() call
• Misusing correlation_id property
• Not acknowledging messages

5. You want to implement an RPC client that sends multiple requests concurrently and matches replies correctly. Which approach best ensures correct matching of replies to requests?

hard

A. Create a new reply queue for each request without correlation_id

B. Send all requests with the same correlation_id to simplify matching

C. Use a single shared reply queue without correlation_id

D. Use a unique correlation_id for each request and check it on reply

Solution

1. Step 1: Understand concurrency in RPC client

   Multiple requests need unique IDs to identify replies correctly.

2. Step 2: Evaluate options for reply matching

   Using unique correlation_id per request allows client to match replies properly.

3. Final Answer:

   Use a unique correlation_id for each request and check it on reply -> Option D

4. Quick Check:

   Unique correlation_id matches replies to requests [OK]

Hint: Assign unique correlation_id per request for reply matching [OK]

Common Mistakes:

• Using same correlation_id for all requests
• Not using correlation_id at all
• Creating many reply queues unnecessarily