Pub/Sub (Fanout)

package main

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

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

const exchange = "logs"
const messages = 5

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

func subscribe(conn *amqp.Connection) (string, <-chan amqp.Delivery) {
	ch, err := conn.Channel()
	if err != nil {
		log.Fatalf("subscriber channel: %v", err)
	}
	if err := ch.ExchangeDeclare(exchange, "fanout", false, false, false, false, nil); err != nil {
		log.Fatalf("declare exchange: %v", err)
	}
	q, err := ch.QueueDeclare("", false, false, true, false, nil) // server-named, exclusive
	if err != nil {
		log.Fatalf("declare queue: %v", err)
	}
	if err := ch.QueueBind(q.Name, "", exchange, false, nil); err != nil { // key ignored
		log.Fatalf("bind: %v", err)
	}
	msgs, err := ch.Consume(q.Name, "", true, false, false, false, nil) // auto-ack
	if err != nil {
		log.Fatalf("consume: %v", err)
	}
	return q.Name, msgs
}

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

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

	nameA, subA := subscribe(conn)
	nameB, subB := subscribe(conn)
	log.Printf(" [s] bound exclusive queues %q and %q to fanout %q", nameA, nameB, exchange)

	// Publish on a confirm channel so every broadcast is accepted before we stop.
	pubCh, _ := conn.Channel()
	if err := pubCh.Confirm(false); err != nil {
		log.Fatalf("confirm select: %v", err)
	}
	confirms := pubCh.NotifyPublish(make(chan amqp.Confirmation, messages))
	for i := 0; i < messages; i++ {
		if err := pubCh.PublishWithContext(ctx, exchange, "", false, false, amqp.Publishing{
			ContentType: "text/plain",
			Body:        []byte(fmt.Sprintf("log-%d", i)),
		}); err != nil {
			log.Fatalf("publish %d: %v", i, err)
		}
	}
	for i := 0; i < messages; i++ {
		<-confirms
	}
	log.Printf(" [x] Broadcast %d messages to fanout %q", messages, exchange)

	// Each subscriber independently receives all 5 copies.
	for label, msgs := range map[string]<-chan amqp.Delivery{"A": subA, "B": subB} {
		got := 0
		for got < messages {
			select {
			case <-msgs:
				got++
			case <-ctx.Done():
				log.Fatalf("subscriber %s timed out (%d/%d)", label, got, messages)
			}
		}
		log.Printf(" [s%s] received all %d broadcasts", label, got)
	}
}

import os

import pika

URL = os.environ.get("KUBEMQ_AMQP_URL", "amqp://guest:guest@localhost:5672/")
EXCHANGE = "logs"
SUBSCRIBERS = 2
MESSAGES = 5


def main() -> None:
    # Subscribers — each on its own connection with a server-named exclusive
    # queue bound to the fanout exchange.
    subs = []
    for _ in range(SUBSCRIBERS):
        conn = pika.BlockingConnection(pika.URLParameters(URL))
        ch = conn.channel()
        ch.exchange_declare(exchange=EXCHANGE, exchange_type="fanout", durable=False)
        queue = ch.queue_declare(queue="", exclusive=True).method.queue  # amq.gen-*
        ch.queue_bind(exchange=EXCHANGE, queue=queue, routing_key="")  # key ignored
        print(f" [sub] bound exclusive queue {queue!r}")
        subs.append((conn, ch, queue))

    # Publisher — broadcast 5 messages on a confirm channel.
    pub_conn = pika.BlockingConnection(pika.URLParameters(URL))
    pub_ch = pub_conn.channel()
    pub_ch.exchange_declare(exchange=EXCHANGE, exchange_type="fanout", durable=False)
    pub_ch.confirm_delivery()
    for i in range(MESSAGES):
        pub_ch.basic_publish(exchange=EXCHANGE, routing_key="", body=f"log-{i}".encode())
    print(f" [x] Broadcast {MESSAGES} messages to {EXCHANGE!r}")

    # Each subscriber independently receives every copy.
    for idx, (conn, ch, queue) in enumerate(subs):
        got = 0
        for method, _props, _body in ch.consume(queue, inactivity_timeout=30, auto_ack=True):
            if method is None:
                raise SystemExit(f"subscriber {idx}: timed out ({got}/{MESSAGES})")
            got += 1
            if got >= MESSAGES:
                break
        ch.cancel()
        print(f" [sub {idx}] received all {got} broadcasts")
        ch.close()
        conn.close()

    pub_ch.close()
    pub_conn.close()


if __name__ == "__main__":
    main()

import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

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

public final class Main {
    private static final String EXCHANGE = "logs";
    private static final int SUBSCRIBERS = 2;
    private static final int MESSAGES = 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("/"); // trailing "/" parses to an empty vhost
        }

        Connection connection = factory.newConnection();

        // Subscribers — each with a server-named exclusive queue bound to fanout.
        List<CountDownLatch> latches = new ArrayList<>();
        for (int i = 0; i < SUBSCRIBERS; i++) {
            Channel ch = connection.createChannel();
            ch.exchangeDeclare(EXCHANGE, "fanout", false);
            String queue = ch.queueDeclare("", false, true, true, null).getQueue(); // amq.gen-*
            ch.queueBind(queue, EXCHANGE, ""); // key ignored for fanout
            CountDownLatch latch = new CountDownLatch(MESSAGES);
            ch.basicConsume(queue, true, (tag, delivery) -> latch.countDown(), tag -> { });
            System.out.println("[sub] bound exclusive queue " + queue);
            latches.add(latch);
        }

        // Publisher — broadcast 5 messages on a confirm channel.
        Channel pub = connection.createChannel();
        pub.exchangeDeclare(EXCHANGE, "fanout", false);
        pub.confirmSelect();
        for (int i = 0; i < MESSAGES; i++) {
            pub.basicPublish(EXCHANGE, "", null, ("log-" + i).getBytes(StandardCharsets.UTF_8));
        }
        pub.waitForConfirmsOrDie(30_000);
        System.out.println("[x] Broadcast " + MESSAGES + " messages to fanout '" + EXCHANGE + "'");

        // Each subscriber independently receives all 5 copies.
        for (int i = 0; i < latches.size(); i++) {
            if (!latches.get(i).await(30, TimeUnit.SECONDS)) {
                throw new IllegalStateException("subscriber " + i + " did not receive all broadcasts");
            }
            System.out.println("[sub " + i + "] received all " + MESSAGES + " broadcasts");
        }
        connection.close();
    }
}

import amqp, { type Channel, type ChannelModel } from "amqplib";

const EXCHANGE = "logs";
const MESSAGES = ["log-0", "log-1", "log-2", "log-3", "log-4"];

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

async function subscribe(connection: ChannelModel): Promise<{ name: string; done: Promise<void> }> {
  const ch: Channel = await connection.createChannel();
  await ch.assertExchange(EXCHANGE, "fanout", { durable: false });
  const q = await ch.assertQueue("", { exclusive: true }); // amq.gen-*
  await ch.bindQueue(q.queue, EXCHANGE, ""); // routing key ignored for fanout
  let received = 0;
  const done = new Promise<void>((resolve) => {
    ch.consume(q.queue, (msg) => {
      if (msg === null) return;
      if (++received === MESSAGES.length) resolve();
    }, { noAck: true });
  });
  return { name: q.queue, done };
}

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

  const subA = await subscribe(connection);
  const subB = await subscribe(connection);
  console.log(`[sub] bound exclusive queues ${subA.name} and ${subB.name}`);

  // Publisher — broadcast on a confirm channel so no message is lost to an early close.
  const pubCh = await connection.createConfirmChannel();
  await pubCh.assertExchange(EXCHANGE, "fanout", { durable: false });
  for (const body of MESSAGES) pubCh.publish(EXCHANGE, "", Buffer.from(body));
  await pubCh.waitForConfirms();
  console.log(`[x] Broadcast ${MESSAGES.length} messages to fanout "${EXCHANGE}"`);

  await Promise.all([subA.done, subB.done]);
  console.log("both subscribers received all 5 broadcasts independently");
  await connection.close();
}

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

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

const string exchange = "logs";
const int subscribers = 2;
const int messages = 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("pub-sub-fanout");

// Subscribers — each with a server-named exclusive queue bound to the fanout.
var latches = new List<TaskCompletionSource>();
for (var i = 0; i < subscribers; i++)
{
    var ch = await connection.CreateChannelAsync();
    await ch.ExchangeDeclareAsync(exchange, ExchangeType.Fanout, durable: false);
    var queue = (await ch.QueueDeclareAsync("", durable: false, exclusive: true, autoDelete: true)).QueueName;
    await ch.QueueBindAsync(queue, exchange, ""); // routing key ignored for fanout
    var done = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
    var got = 0;
    var consumer = new AsyncEventingBasicConsumer(ch);
    consumer.ReceivedAsync += (_, _) =>
    {
        if (Interlocked.Increment(ref got) == messages) done.TrySetResult();
        return Task.CompletedTask;
    };
    await ch.BasicConsumeAsync(queue, autoAck: true, consumer: consumer);
    Console.WriteLine($"[sub] bound exclusive queue {queue}");
    latches.Add(done);
}

// Publisher — broadcast on a confirm channel.
var confirmOpts = new CreateChannelOptions(
    publisherConfirmationsEnabled: true,
    publisherConfirmationTrackingEnabled: true);
await using var pub = await connection.CreateChannelAsync(confirmOpts);
await pub.ExchangeDeclareAsync(exchange, ExchangeType.Fanout, durable: false);
for (var i = 0; i < messages; i++)
    await pub.BasicPublishAsync(exchange, "", body: Encoding.UTF8.GetBytes($"log-{i}"));
Console.WriteLine($"[x] Broadcast {messages} messages to fanout '{exchange}'");

using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30));
await Task.WhenAll(latches.Select(l => l.Task.WaitAsync(cts.Token)));
Console.WriteLine("both subscribers received all broadcasts independently");

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

EXCHANGE = "logs"
MESSAGE_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

pub_ch = conn.create_channel
pub_ch.confirm_select
exchange = pub_ch.fanout(EXCHANGE, durable: false)

# Two subscribers, each with its own server-named exclusive queue.
subscribers = Array.new(2) do |idx|
  sub_ch = conn.create_channel
  queue = sub_ch.queue("", exclusive: true) # amq.gen-*
  queue.bind(exchange) # routing key ignored for fanout
  received = Queue.new
  queue.subscribe(manual_ack: false, block: false) { |_di, _props, _body| received.push(:msg) }
  puts " [*] Subscriber #{idx + 1} bound exclusive queue #{queue.name}"
  { id: idx + 1, received: received }
end

# Broadcast 5 messages, then wait for confirms.
MESSAGE_COUNT.times { |i| exchange.publish("log-#{i}", routing_key: "ignored") }
pub_ch.wait_for_confirms
puts " [x] Broadcast #{MESSAGE_COUNT} messages to fanout '#{EXCHANGE}'"

# Each subscriber independently receives all 5.
subscribers.each do |s|
  MESSAGE_COUNT.times { s[:received].pop }
  puts " [x] Subscriber #{s[:id]} received all #{MESSAGE_COUNT} broadcasts"
end

exchange.delete
conn.close

use futures_lite::StreamExt;
use lapin::{
    options::{
        BasicConsumeOptions, BasicPublishOptions, ConfirmSelectOptions, ExchangeDeclareOptions,
        QueueBindOptions, QueueDeclareOptions,
    },
    types::FieldTable,
    BasicProperties, Connection, ConnectionProperties, ExchangeKind,
};

const EXCHANGE: &str = "logs";
const MESSAGES: 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,
    }
}

async fn subscribe(conn: &Connection) -> Result<lapin::Consumer, Box<dyn std::error::Error>> {
    let ch = conn.create_channel().await?;
    ch.exchange_declare(EXCHANGE, ExchangeKind::Fanout, ExchangeDeclareOptions::default(), FieldTable::default())
        .await?;
    let queue = ch
        .queue_declare("", QueueDeclareOptions { exclusive: true, ..Default::default() }, FieldTable::default())
        .await?;
    ch.queue_bind(queue.name().as_str(), EXCHANGE, "", QueueBindOptions::default(), FieldTable::default())
        .await?;
    let consumer = ch
        .basic_consume(queue.name().as_str(), "", BasicConsumeOptions { no_ack: true, ..Default::default() }, FieldTable::default())
        .await?;
    println!("[sub] bound exclusive queue {}", queue.name());
    Ok(consumer)
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let conn = Connection::connect(&amqp_url(), ConnectionProperties::default()).await?;

    let mut sub_a = subscribe(&conn).await?;
    let mut sub_b = subscribe(&conn).await?;

    // Publisher — broadcast on a confirm channel.
    let pub_ch = conn.create_channel().await?;
    pub_ch.exchange_declare(EXCHANGE, ExchangeKind::Fanout, ExchangeDeclareOptions::default(), FieldTable::default())
        .await?;
    pub_ch.confirm_select(ConfirmSelectOptions::default()).await?;
    for i in 0..MESSAGES {
        pub_ch
            .basic_publish(EXCHANGE, "", BasicPublishOptions::default(), format!("log-{i}").as_bytes(), BasicProperties::default())
            .await?
            .await?;
    }
    println!("[x] Broadcast {MESSAGES} messages to fanout '{EXCHANGE}'");

    for (label, consumer) in [("A", &mut sub_a), ("B", &mut sub_b)] {
        for _ in 0..MESSAGES {
            consumer.next().await.ok_or("subscriber stream closed early")??;
        }
        println!("[s{label}] received all {MESSAGES} broadcasts");
    }
    conn.close(0, "done").await?;
    Ok(())
}