cleveragents-core/.opencode/skills/programming-patterns/references/messaging/request-reply.md

Request-Reply Pattern

Problem

Standard message queues are fire-and-forget — the sender publishes a message and moves on. But sometimes the sender needs a response: a query result, a validation outcome, or a computed value. Without a structured way to correlate requests with responses, implementing RPC over messaging becomes ad-hoc and error-prone.

Solution

The sender includes a correlation ID and a reply-to address (queue/callback) with each request. The receiver processes the request and sends the response back to the specified reply address, including the same correlation ID. The sender matches responses to their original requests using the correlation ID.

When to Use

• RPC-style communication over message infrastructure
• When you want the decoupling of messaging but need responses
• Async queries across services
• When the response might take significant time (long-running computations)
• Gateway patterns that aggregate responses from multiple services

When to Avoid

• Simple fire-and-forget work (use plain message queue)
• When low latency is critical (direct HTTP/gRPC is more efficient)
• When the response is not needed
• High-throughput scenarios where correlation tracking becomes expensive

Pseudocode

# Sender
function send_request(payload):
    correlation_id = generate_uuid()
    pending[correlation_id] = create_future()
    send(message={payload, correlation_id, reply_to=my_reply_queue})
    return pending[correlation_id].await(timeout)

# Receiver
function on_message(message):
    result = process(message.payload)
    send(reply_to=message.reply_to,
         message={result, correlation_id=message.correlation_id})

Python Implementation

import uuid
import threading
import time
from dataclasses import dataclass, field
from typing import Any, Optional
from queue import Queue, Empty


@dataclass
class Message:
    payload: Any
    correlation_id: str = field(default_factory=lambda: str(uuid.uuid4())[:8])
    reply_to: Optional[str] = None


class MessageBroker:
    """Simple in-memory broker with named queues."""

    def __init__(self):
        self._queues: dict[str, Queue] = {}

    def get_queue(self, name: str) -> Queue:
        if name not in self._queues:
            self._queues[name] = Queue()
        return self._queues[name]

    def send(self, queue_name: str, message: Message):
        self.get_queue(queue_name).put(message)

    def receive(self, queue_name: str, timeout: float = 2.0) -> Optional[Message]:
        try:
            return self.get_queue(queue_name).get(timeout=timeout)
        except Empty:
            return None


class RequestReplyClient:
    def __init__(self, broker: MessageBroker, reply_queue: str):
        self.broker = broker
        self.reply_queue = reply_queue
        self._pending: dict[str, threading.Event] = {}
        self._responses: dict[str, Any] = {}

        # Start reply listener
        self._running = True
        self._listener = threading.Thread(target=self._listen, daemon=True)
        self._listener.start()

    def _listen(self):
        while self._running:
            msg = self.broker.receive(self.reply_queue, timeout=0.5)
            if msg and msg.correlation_id in self._pending:
                self._responses[msg.correlation_id] = msg.payload
                self._pending[msg.correlation_id].set()

    def request(self, target_queue: str, payload: Any, timeout: float = 5.0) -> Any:
        corr_id = str(uuid.uuid4())[:8]
        event = threading.Event()
        self._pending[corr_id] = event

        msg = Message(payload=payload, correlation_id=corr_id, reply_to=self.reply_queue)
        print(f"  [Client] Sending request (corr={corr_id}): {payload}")
        self.broker.send(target_queue, msg)

        if not event.wait(timeout=timeout):
            raise TimeoutError(f"No reply for correlation {corr_id}")

        response = self._responses.pop(corr_id)
        del self._pending[corr_id]
        return response

    def stop(self):
        self._running = False


class RequestReplyServer:
    def __init__(self, broker: MessageBroker, listen_queue: str, handler):
        self.broker = broker
        self.listen_queue = listen_queue
        self.handler = handler
        self._running = True
        self._thread = threading.Thread(target=self._process, daemon=True)
        self._thread.start()

    def _process(self):
        while self._running:
            msg = self.broker.receive(self.listen_queue, timeout=0.5)
            if msg is None:
                continue
            print(f"  [Server] Received request (corr={msg.correlation_id}): {msg.payload}")
            result = self.handler(msg.payload)
            reply = Message(payload=result, correlation_id=msg.correlation_id)
            print(f"  [Server] Sending reply (corr={msg.correlation_id}): {result}")
            self.broker.send(msg.reply_to, reply)

    def stop(self):
        self._running = False


# --- Demo ---
broker = MessageBroker()

# Server: computes square of numbers
def compute_handler(payload):
    time.sleep(0.1)  # Simulate work
    return {"input": payload["value"], "result": payload["value"] ** 2}

server = RequestReplyServer(broker, "compute_queue", compute_handler)
client = RequestReplyClient(broker, "client_reply_queue")

print("=== Request-Reply Demo ===")
for val in [5, 12, 7]:
    response = client.request("compute_queue", {"value": val})
    print(f"  [Client] Got reply: {val}^2 = {response['result']}")
    print()

client.stop()
server.stop()
print("Done.")

Output:

=== Request-Reply Demo ===
  [Client] Sending request (corr=abc12345): {'value': 5}
  [Server] Received request (corr=abc12345): {'value': 5}
  [Server] Sending reply (corr=abc12345): {'input': 5, 'result': 25}
  [Client] Got reply: 5^2 = 25

  [Client] Sending request (corr=def67890): {'value': 12}
  [Server] Received request (corr=def67890): {'value': 12}
  [Server] Sending reply (corr=def67890): {'input': 12, 'result': 144}
  [Client] Got reply: 12^2 = 144

  [Client] Sending request (corr=ghi24680): {'value': 7}
  [Server] Received request (corr=ghi24680): {'value': 7}
  [Server] Sending reply (corr=ghi24680): {'input': 7, 'result': 49}
  [Client] Got reply: 7^2 = 49

Done.

Go Implementation

package main

import (
	"fmt"
	"sync"
	"time"
)

type Message struct {
	Payload       interface{}
	CorrelationID string
	ReplyTo       string
}

type Broker struct {
	mu     sync.Mutex
	queues map[string]chan Message
}

func NewBroker() *Broker {
	return &Broker{queues: make(map[string]chan Message)}
}

func (b *Broker) Queue(name string) chan Message {
	b.mu.Lock()
	defer b.mu.Unlock()
	if _, ok := b.queues[name]; !ok {
		b.queues[name] = make(chan Message, 100)
	}
	return b.queues[name]
}

func main() {
	broker := NewBroker()
	requestQ := broker.Queue("compute")
	replyQ := broker.Queue("replies")
	done := make(chan struct{})

	// Server goroutine
	go func() {
		for {
			select {
			case msg := <-requestQ:
				val := msg.Payload.(int)
				fmt.Printf("  [Server] Received (corr=%s): %d\n", msg.CorrelationID, val)
				time.Sleep(50 * time.Millisecond)
				result := val * val
				fmt.Printf("  [Server] Replying (corr=%s): %d\n", msg.CorrelationID, result)
				broker.Queue(msg.ReplyTo) <- Message{
					Payload:       result,
					CorrelationID: msg.CorrelationID,
				}
			case <-done:
				return
			}
		}
	}()

	// Client sends requests
	fmt.Println("=== Request-Reply Demo ===")
	for i, val := range []int{5, 12, 7} {
		corrID := fmt.Sprintf("req-%d", i+1)
		fmt.Printf("  [Client] Sending (corr=%s): %d\n", corrID, val)
		requestQ <- Message{Payload: val, CorrelationID: corrID, ReplyTo: "replies"}

		// Wait for matching reply
		reply := <-replyQ
		fmt.Printf("  [Client] Got reply: %d^2 = %v\n\n", val, reply.Payload)
	}

	close(done)
	fmt.Println("Done.")
}

Related Patterns

• Future/Promise — async request-reply is naturally modelled as a future: send the request, await the promise for the reply.
• Message Queue — request-reply is built on top of message queues; the correlation ID links request to reply.
• Timeout — always apply a timeout to request-reply so a lost reply doesn't block the caller indefinitely.