RPC (Direct Reply-To)

package main

import (
	"context"
	"fmt"
	"log"
	"os"
	"strings"
	"time"

	amqp "github.com/rabbitmq/amqp091-go"
)

const (
	rpcQueue = "rpc-queue"
	replyTo  = "amq.rabbitmq.reply-to"
	calls    = 5
)

func amqpURL() string {
	if v := os.Getenv("KUBEMQ_AMQP_URL"); v != "" {
		return v
	}
	return "amqp://guest:guest@localhost:5672/"
}

func main() {
	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
	defer cancel()

	conn, err := amqp.Dial(amqpURL())
	if err != nil {
		log.Fatalf("dial: %v", err)
	}
	defer func() { _ = conn.Close() }()

	// Responder: consume rpc-queue, reply to the minted reply-to address.
	serverCh, _ := conn.Channel()
	if _, err := serverCh.QueueDeclare(rpcQueue, false, false, false, false, nil); err != nil {
		log.Fatalf("declare rpc-queue: %v", err)
	}
	requests, err := serverCh.Consume(rpcQueue, "responder", true, false, false, false, nil)
	if err != nil {
		log.Fatalf("responder consume: %v", err)
	}
	go func() {
		for req := range requests {
			// The requester's reply-to is rewritten to a minted opaque address.
			if !strings.HasPrefix(req.ReplyTo, "amq.rabbitmq.reply-to.g1.") {
				log.Printf("warning: responder expected a minted address, got %q", req.ReplyTo)
			}
			_ = serverCh.PublishWithContext(ctx, "", req.ReplyTo, false, false, amqp.Publishing{
				ContentType:   "text/plain",
				CorrelationId: req.CorrelationId,
				Body:          append([]byte("echo:"), req.Body...),
			})
		}
	}()

	// Requester: consume the pseudo-queue with no-ack (manual ack → 406).
	clientCh, _ := conn.Channel()
	replies, err := clientCh.Consume(replyTo, "", true, false, false, false, nil)
	if err != nil {
		log.Fatalf("consuming %s requires no-ack: %v", replyTo, err)
	}

	for i := 1; i <= calls; i++ {
		corr := fmt.Sprintf("corr-%d", i)
		body := fmt.Sprintf("request-%d", i)
		if err := clientCh.PublishWithContext(ctx, "", rpcQueue, false, false, amqp.Publishing{
			ContentType:   "text/plain",
			CorrelationId: corr,
			ReplyTo:       replyTo,
			Body:          []byte(body),
		}); err != nil {
			log.Fatalf("request %d: %v", i, err)
		}
		select {
		case reply := <-replies:
			if reply.CorrelationId != corr {
				log.Fatalf("call %d: correlation-id %q != %q", i, reply.CorrelationId, corr)
			}
			log.Printf(" [rpc] %s (corr=%s) → %s", body, corr, reply.Body)
		case <-time.After(30 * time.Second):
			log.Fatalf("timed out waiting for reply %d", i)
		}
	}
	log.Printf(" [✓] %d RPC calls round-tripped over %s, correlation-id matched", calls, replyTo)

	if _, err := serverCh.QueueDelete(rpcQueue, false, false, false); err != nil {
		log.Printf("warning: rpc-queue delete: %v", err)
	}
}

import os

import pika

URL = os.environ.get("KUBEMQ_AMQP_URL", "amqp://guest:guest@localhost:5672/")
RPC_QUEUE = "rpc-queue"
REPLY_TO = "amq.rabbitmq.reply-to"
CALLS = 5


def main() -> None:
    # Responder: consume requests, echo back to the minted reply address.
    server_conn = pika.BlockingConnection(pika.URLParameters(URL))
    server_ch = server_conn.channel()
    server_ch.queue_declare(queue=RPC_QUEUE, durable=False)

    def on_request(ch, method, props, body):
        # The connector rewrites reply-to to an opaque minted address; the
        # responder only ever sees that, never the pseudo-queue name.
        ch.basic_publish(
            exchange="",
            routing_key=props.reply_to,
            body=b"echo:" + body,
            properties=pika.BasicProperties(content_type="text/plain", correlation_id=props.correlation_id),
        )
        ch.basic_ack(method.delivery_tag)

    server_ch.basic_consume(queue=RPC_QUEUE, on_message_callback=on_request)

    # Requester: consume amq.rabbitmq.reply-to (no-ack is mandatory).
    client_conn = pika.BlockingConnection(pika.URLParameters(URL))
    client_ch = client_conn.channel()
    responses: dict[str, str] = {}
    client_ch.basic_consume(queue=REPLY_TO, on_message_callback=lambda c, m, p, b: responses.__setitem__(p.correlation_id, b.decode()), auto_ack=True)

    print(" [client] issuing RPC requests over amq.rabbitmq.reply-to:")
    for i in range(1, CALLS + 1):
        corr = f"corr-{i}"
        client_ch.basic_publish(
            exchange="",
            routing_key=RPC_QUEUE,
            body=f"request-{i}".encode(),
            properties=pika.BasicProperties(content_type="text/plain", correlation_id=corr, reply_to=REPLY_TO),
        )
        # Pump the responder to handle the request, then the client to receive.
        while corr not in responses:
            server_conn.process_data_events(time_limit=1)
            client_conn.process_data_events(time_limit=1)
        print(f"   request-{i} [{corr}] -> {responses[corr]!r}")

    print(f" [x] all {CALLS} responses correlation-id matched (native AMQP RPC)")
    server_ch.queue_delete(queue=RPC_QUEUE)
    client_conn.close()
    server_conn.close()


if __name__ == "__main__":
    main()

import java.nio.charset.StandardCharsets;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.TimeUnit;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;

public final class Main {
    private static final String RPC_QUEUE = "rpc-queue";
    private static final String REPLY_TO = "amq.rabbitmq.reply-to";
    private static final int CALLS = 5;

    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setUri(System.getenv().getOrDefault("KUBEMQ_AMQP_URL", "amqp://guest:guest@localhost:5672/"));
        if (factory.getVirtualHost() == null || factory.getVirtualHost().isEmpty()) {
            factory.setVirtualHost("/");
        }

        try (Connection serverConn = factory.newConnection();
                Connection clientConn = factory.newConnection()) {
            // Responder: consume requests, reply to the minted address.
            Channel serverCh = serverConn.createChannel();
            serverCh.queueDeclare(RPC_QUEUE, false, false, false, null);
            DeliverCallback onRequest = (tag, req) -> {
                String replyAddr = req.getProperties().getReplyTo(); // minted address
                AMQP.BasicProperties replyProps = new AMQP.BasicProperties.Builder()
                        .contentType("text/plain")
                        .correlationId(req.getProperties().getCorrelationId())
                        .build();
                byte[] body = ("echo:" + new String(req.getBody(), StandardCharsets.UTF_8)).getBytes(StandardCharsets.UTF_8);
                serverCh.basicPublish("", replyAddr, replyProps, body);
            };
            serverCh.basicConsume(RPC_QUEUE, true, onRequest, tag -> { });

            // Requester: consume the pseudo-queue (auto-ack), then call.
            Channel clientCh = clientConn.createChannel();
            ConcurrentHashMap<String, SynchronousQueue<String>> pending = new ConcurrentHashMap<>();
            DeliverCallback onReply = (tag, reply) -> {
                SynchronousQueue<String> slot = pending.get(reply.getProperties().getCorrelationId());
                if (slot != null) {
                    slot.offer(new String(reply.getBody(), StandardCharsets.UTF_8));
                }
            };
            clientCh.basicConsume(REPLY_TO, true, onReply, tag -> { });

            for (int i = 1; i <= CALLS; i++) {
                String corr = "corr-" + i;
                String request = "request-" + i;
                SynchronousQueue<String> slot = new SynchronousQueue<>();
                pending.put(corr, slot);
                AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
                        .contentType("text/plain")
                        .correlationId(corr)
                        .replyTo(REPLY_TO)
                        .build();
                clientCh.basicPublish("", RPC_QUEUE, props, request.getBytes(StandardCharsets.UTF_8));

                String reply = slot.poll(30, TimeUnit.SECONDS);
                pending.remove(corr);
                if (reply == null) throw new IllegalStateException("timed out on reply " + i);
                System.out.println("[rpc] " + corr + ": '" + request + "' -> '" + reply + "'");
            }
            System.out.println("[x] " + CALLS + " RPC calls matched by correlation-id over amq.rabbitmq.reply-to");
        }
    }
}

import amqp from "amqplib";

const RPC_QUEUE = "rpc-queue";
const REPLY_TO = "amq.rabbitmq.reply-to";
const CALLS = 5;

function url(): string {
  return process.env["KUBEMQ_AMQP_URL"] ?? "amqp://guest:guest@localhost:5672/";
}

async function main(): Promise<void> {
  const connection = await amqp.connect(url());

  // Responder: consume rpc-queue, echo back to the reply-to address.
  const serverCh = await connection.createChannel();
  await serverCh.assertQueue(RPC_QUEUE, { durable: false });
  await serverCh.consume(RPC_QUEUE, (req) => {
    if (req === null) return;
    const replyAddr = req.properties.replyTo; // minted address
    serverCh.publish("", replyAddr, Buffer.concat([Buffer.from("echo:"), req.content]), {
      contentType: "text/plain",
      correlationId: req.properties.correlationId,
    });
  }, { noAck: true });

  // Requester: consume amq.rabbitmq.reply-to with no-ack (manual ack → 406).
  const clientCh = await connection.createChannel();
  const pending = new Map<string, (body: string) => void>();
  await clientCh.consume(REPLY_TO, (reply) => {
    if (reply === null) return;
    const resolve = pending.get(reply.properties.correlationId);
    if (resolve) {
      pending.delete(reply.properties.correlationId);
      resolve(reply.content.toString());
    }
  }, { noAck: true });

  for (let i = 1; i <= CALLS; i++) {
    const corr = `corr-${i}`;
    const body = `request-${i}`;
    const response = new Promise<string>((resolve) => pending.set(corr, resolve));
    clientCh.publish("", RPC_QUEUE, Buffer.from(body), {
      contentType: "text/plain",
      correlationId: corr,
      replyTo: REPLY_TO,
    });
    console.log(`[client] ${body} (${corr}) -> ${await response}`);
  }

  console.log(`[client] completed ${CALLS} RPC round-trips, all correlation-id matched`);
  await serverCh.deleteQueue(RPC_QUEUE);
  await connection.close();
}

main().catch((err) => {
  console.error(err);
  process.exit(1);
});

using System.Collections.Concurrent;
using System.Text;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;

const string requestQueue = "rpc-queue";
const string replyToPseudoQueue = "amq.rabbitmq.reply-to";
const int calls = 5;

static string Url() =>
    Environment.GetEnvironmentVariable("KUBEMQ_AMQP_URL") is { Length: > 0 } v
        ? v
        : "amqp://guest:guest@localhost:5672/";

var factory = new ConnectionFactory { Uri = new Uri(Url()) };
await using var connection = await factory.CreateConnectionAsync("rpc");

// Responder: consume rpc-queue, echo back to the delivered reply-to.
await using var serverCh = await connection.CreateChannelAsync();
await serverCh.QueueDeclareAsync(requestQueue, durable: false, exclusive: false, autoDelete: false);
var serverConsumer = new AsyncEventingBasicConsumer(serverCh);
serverConsumer.ReceivedAsync += async (_, ea) =>
{
    var replyTo = ea.BasicProperties.ReplyTo ?? ""; // minted address
    var replyProps = new BasicProperties
    {
        ContentType = "text/plain",
        CorrelationId = ea.BasicProperties.CorrelationId,
    };
    var replyBody = Encoding.UTF8.GetBytes("echo:" + Encoding.UTF8.GetString(ea.Body.Span));
    await serverCh.BasicPublishAsync("", replyTo, mandatory: false, basicProperties: replyProps, body: replyBody);
};
await serverCh.BasicConsumeAsync(requestQueue, autoAck: true, consumer: serverConsumer);

// Requester: consume the pseudo-queue with no-ack (manual ack → 406).
await using var clientCh = await connection.CreateChannelAsync();
var pending = new ConcurrentDictionary<string, TaskCompletionSource<string>>();
var clientConsumer = new AsyncEventingBasicConsumer(clientCh);
clientConsumer.ReceivedAsync += (_, ea) =>
{
    if (pending.TryRemove(ea.BasicProperties.CorrelationId ?? "", out var tcs))
        tcs.TrySetResult(Encoding.UTF8.GetString(ea.Body.Span));
    return Task.CompletedTask;
};
await clientCh.BasicConsumeAsync(replyToPseudoQueue, autoAck: true, consumer: clientConsumer);

try
{
    for (var i = 1; i <= calls; i++)
    {
        var corr = $"corr-{i}";
        var tcs = new TaskCompletionSource<string>(TaskCreationOptions.RunContinuationsAsynchronously);
        pending[corr] = tcs;
        var reqProps = new BasicProperties
        {
            ContentType = "text/plain",
            CorrelationId = corr,
            ReplyTo = replyToPseudoQueue,
        };
        var request = $"request-{i}";
        await clientCh.BasicPublishAsync("", requestQueue, mandatory: false,
            basicProperties: reqProps, body: Encoding.UTF8.GetBytes(request));
        Console.WriteLine($"[rpc] {corr}: '{request}' → '{await tcs.Task}'");
    }
    Console.WriteLine($"[x] All {calls} RPC calls correlation-id matched over native amq.rabbitmq.reply-to");
}
finally
{
    await serverCh.QueueDeleteAsync(requestQueue, ifUnused: false, ifEmpty: false);
}

# frozen_string_literal: true
require "bunny"
require "amq/uri"

RPC_QUEUE = "rpc-queue"
REPLY_TO = "amq.rabbitmq.reply-to"
REQUEST_COUNT = 5

opts = AMQ::URI.parse(ENV.fetch("KUBEMQ_AMQP_URL", "amqp://guest:guest@localhost:5672/"))
opts[:vhost] = "/" if opts[:vhost].nil? || opts[:vhost].to_s.empty?
conn = Bunny.new(opts)
conn.start

# Responder: consume rpc-queue, echo back with the correlation-id.
server_ch = conn.create_channel
rpc_q = server_ch.queue(RPC_QUEUE, durable: false)
rpc_q.subscribe(manual_ack: false, block: false) do |_di, props, body|
  server_ch.default_exchange.publish("echo:#{body}", routing_key: props.reply_to, correlation_id: props.correlation_id)
end

# Requester: consume amq.rabbitmq.reply-to (no-ack MUST be true) and fire 5 requests.
client_ch = conn.create_channel
replies = {}
mutex = Mutex.new
cond = ConditionVariable.new
client_ch.basic_consume(REPLY_TO, "", true, false) do |_di, props, body|
  mutex.synchronize do
    replies[props.correlation_id] = body
    cond.signal
  end
end

requests = {}
REQUEST_COUNT.times do |i|
  cid = "corr-#{i + 1}"
  payload = "request-#{i + 1}"
  requests[cid] = payload
  client_ch.default_exchange.publish(payload, routing_key: RPC_QUEUE, reply_to: REPLY_TO, correlation_id: cid)
  puts " [>] request #{cid} body=#{payload.inspect}"
end

deadline = Time.now + 10
mutex.synchronize do
  cond.wait(mutex, 0.5) while replies.size < REQUEST_COUNT && Time.now < deadline
end

requests.keys.sort.each { |cid| puts " [<] response #{cid} -> #{replies[cid].inspect}" }
puts " [x] All #{REQUEST_COUNT} RPC round-trips matched by correlation-id"

rpc_q.delete
conn.close

use futures_lite::StreamExt;
use lapin::{
    options::{BasicConsumeOptions, BasicPublishOptions, QueueDeclareOptions, QueueDeleteOptions},
    types::FieldTable,
    BasicProperties, Connection, ConnectionProperties,
};
use std::collections::BTreeMap;

const RPC_QUEUE: &str = "rpc-queue";
const REPLY_TO: &str = "amq.rabbitmq.reply-to";
const REQUESTS: usize = 5;

fn amqp_url() -> String {
    let url = std::env::var("KUBEMQ_AMQP_URL")
        .unwrap_or_else(|_| "amqp://guest:guest@localhost:5672/".into());
    match url.rsplit_once('@').map_or(url.as_str(), |(_, host)| host) {
        host if host.ends_with('/') && !host.to_lowercase().ends_with("/%2f") => format!("{url}%2f"),
        _ => url,
    }
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let url = amqp_url();

    // Responder: consume rpc-queue and reply to each request's reply-to address.
    let responder_conn = Connection::connect(&url, ConnectionProperties::default()).await?;
    let responder_ch = responder_conn.create_channel().await?;
    responder_ch
        .queue_declare(RPC_QUEUE, QueueDeclareOptions::default(), FieldTable::default())
        .await?;
    let mut requests = responder_ch
        .basic_consume(RPC_QUEUE, "responder", BasicConsumeOptions { no_ack: true, ..Default::default() }, FieldTable::default())
        .await?;

    let responder = tokio::spawn(async move {
        let mut handled = 0usize;
        while handled < REQUESTS {
            let Some(delivery) = requests.next().await else { break };
            let delivery = delivery?;
            let reply_to = delivery.properties.reply_to().as_ref().map(|s| s.to_string()).unwrap_or_default();
            let corr = delivery.properties.correlation_id().clone();
            let mut body = b"echo:".to_vec();
            body.extend_from_slice(&delivery.data);
            // The responder only ever sees the minted reply-to address.
            responder_ch
                .basic_publish(
                    "",
                    reply_to.as_str(),
                    BasicPublishOptions::default(),
                    &body,
                    BasicProperties::default().with_correlation_id(corr.unwrap_or_else(|| "".into())),
                )
                .await?
                .await?;
            handled += 1;
        }
        Ok::<_, lapin::Error>(())
    });

    // Requester: consume amq.rabbitmq.reply-to (no-ack is mandatory; manual ack → 406).
    let requester_conn = Connection::connect(&url, ConnectionProperties::default()).await?;
    let requester_ch = requester_conn.create_channel().await?;
    let mut replies = requester_ch
        .basic_consume(REPLY_TO, "requester", BasicConsumeOptions { no_ack: true, ..Default::default() }, FieldTable::default())
        .await?;

    let mut pending: BTreeMap<String, String> = BTreeMap::new();
    for i in 1..=REQUESTS {
        let corr = format!("corr-{i}");
        let body = format!("request-{i}");
        pending.insert(corr.clone(), format!("echo:{body}"));
        requester_ch
            .basic_publish(
                "",
                RPC_QUEUE,
                BasicPublishOptions::default(),
                body.as_bytes(),
                BasicProperties::default()
                    .with_reply_to(REPLY_TO.into())
                    .with_correlation_id(corr.as_str().into()),
            )
            .await?
            .await?;
    }

    let mut matched = 0usize;
    while matched < REQUESTS {
        let delivery = replies.next().await.ok_or("reply stream closed")??;
        let corr = delivery.properties.correlation_id().as_ref().map(|s| s.to_string()).unwrap_or_default();
        let body = String::from_utf8_lossy(&delivery.data).into_owned();
        if let Some(want) = pending.remove(&corr) {
            assert_eq!(body, want, "reply must echo the request");
            println!("[requester] matched {corr}: {body}");
            matched += 1;
        }
    }
    println!("[x] All {REQUESTS} RPC request/response pairs matched by correlation-id");

    responder.await??;
    let _ = requester_ch.queue_delete(RPC_QUEUE, QueueDeleteOptions::default()).await;
    requester_conn.close(0, "done").await?;
    responder_conn.close(0, "done").await?;
    Ok(())
}