Topics

package main

import (
	"context"
	"log"
	"os"
	"sort"
	"time"

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

const exchange = "topic_logs"

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

func bindConsumer(conn *amqp.Connection, patterns ...string) <-chan amqp.Delivery {
	ch, err := conn.Channel()
	if err != nil {
		log.Fatalf("channel: %v", err)
	}
	q, err := ch.QueueDeclare("", false, false, true, false, nil) // server-named, exclusive
	if err != nil {
		log.Fatalf("declare: %v", err)
	}
	for _, p := range patterns {
		if err := ch.QueueBind(q.Name, p, exchange, false, nil); err != nil {
			log.Fatalf("bind %s: %v", p, err)
		}
	}
	msgs, err := ch.Consume(q.Name, "", true, false, false, false, nil)
	if err != nil {
		log.Fatalf("consume: %v", err)
	}
	return 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() }()

	ch, _ := conn.Channel()
	if err := ch.ExchangeDeclare(exchange, "topic", false, false, false, false, nil); err != nil {
		log.Fatalf("declare exchange: %v", err)
	}

	q1 := bindConsumer(conn, "*.orange.*")          // orange animals
	q2 := bindConsumer(conn, "*.*.rabbit", "lazy.#") // rabbits and everything lazy

	keys := []string{
		"quick.orange.rabbit",  // q1 + q2
		"lazy.orange.elephant", // q1 + q2 (lazy.# spans two words)
		"quick.orange.fox",     // q1 only
		"lazy.brown.fox",       // q2 only (lazy.#)
		"lazy.pink.rabbit",     // q2 only — matches BOTH q2 bindings, ONE copy
		"quick.brown.fox",      // nobody → silent drop
	}
	for _, key := range keys {
		if err := ch.PublishWithContext(ctx, exchange, key, false, false, amqp.Publishing{
			ContentType: "text/plain",
			Body:        []byte(key),
		}); err != nil {
			log.Fatalf("publish %s: %v", key, err)
		}
	}
	log.Printf(" [x] Published %d keys", len(keys))

	collect := func(label string, msgs <-chan amqp.Delivery, want []string) {
		seen := make(map[string]struct{}, len(want))
		for len(seen) < len(want) {
			select {
			case d := <-msgs:
				seen[string(d.Body)] = struct{}{}
			case <-ctx.Done():
				log.Fatalf("%s timed out (%d/%d)", label, len(seen), len(want))
			}
		}
		out := make([]string, 0, len(seen))
		for k := range seen {
			out = append(out, k)
		}
		sort.Strings(out)
		log.Printf(" [%s] matched: %v", label, out)
	}
	collect("q1", q1, []string{"quick.orange.rabbit", "lazy.orange.elephant", "quick.orange.fox"})
	collect("q2", q2, []string{"quick.orange.rabbit", "lazy.orange.elephant", "lazy.brown.fox", "lazy.pink.rabbit"})
	log.Printf(" [✓] lazy.pink.rabbit matched both q2 bindings → one copy; quick.brown.fox → dropped")
}

import os

import pika

URL = os.environ.get("KUBEMQ_AMQP_URL", "amqp://guest:guest@localhost:5672/")
EXCHANGE = "topic_logs"


def main() -> None:
    conn = pika.BlockingConnection(pika.URLParameters(URL))
    ch = conn.channel()
    ch.exchange_declare(exchange=EXCHANGE, exchange_type="topic", durable=False)

    def bind(*patterns: str) -> str:
        queue = ch.queue_declare(queue="", exclusive=True).method.queue
        for p in patterns:
            ch.queue_bind(exchange=EXCHANGE, queue=queue, routing_key=p)
        return queue

    q1 = bind("*.orange.*")            # orange animals
    q2 = bind("*.*.rabbit", "lazy.#")  # rabbits and everything lazy

    keys = [
        "quick.orange.rabbit",   # q1 + q2
        "lazy.orange.elephant",  # q1 + q2
        "quick.orange.fox",      # q1 only
        "lazy.brown.fox",        # q2 only
        "lazy.pink.rabbit",      # q2 only — matches BOTH q2 bindings, ONE copy
        "quick.brown.fox",       # nobody → silent drop
    ]
    for key in keys:
        ch.basic_publish(exchange=EXCHANGE, routing_key=key, body=key.encode())
    print(f" [x] Published {len(keys)} keys")

    def collect(label: str, queue: str, want: set[str]) -> None:
        seen: set[str] = set()
        for method, _props, body in ch.consume(queue, inactivity_timeout=30, auto_ack=True):
            if method is None:
                raise SystemExit(f"{label} timed out ({len(seen)}/{len(want)})")
            seen.add(body.decode())
            if len(seen) >= len(want):
                break
        ch.cancel()
        print(f" [{label}] matched: {sorted(seen)}")

    collect("q1", q1, {"quick.orange.rabbit", "lazy.orange.elephant", "quick.orange.fox"})
    collect("q2", q2, {"quick.orange.rabbit", "lazy.orange.elephant", "lazy.brown.fox", "lazy.pink.rabbit"})
    print(" [x] lazy.pink.rabbit matched both q2 bindings → one copy; quick.brown.fox → dropped")
    ch.close()
    conn.close()


if __name__ == "__main__":
    main()

import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

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

public final class Main {
    private static final String EXCHANGE = "topic_logs";

    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("/");
        }

        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        channel.exchangeDeclare(EXCHANGE, "topic", false);

        Set<String> q1 = ConcurrentHashMap.newKeySet();
        Set<String> q2 = ConcurrentHashMap.newKeySet();
        bind(connection, q1, "*.orange.*");
        bind(connection, q2, "*.*.rabbit", "lazy.#");

        String[] keys = {
            "quick.orange.rabbit",  // q1 + q2
            "lazy.orange.elephant", // q1 + q2
            "quick.orange.fox",     // q1 only
            "lazy.brown.fox",       // q2 only
            "lazy.pink.rabbit",     // q2 only — matches BOTH q2 bindings, ONE copy
            "quick.brown.fox",      // nobody → silent drop
        };
        for (String key : keys) {
            channel.basicPublish(EXCHANGE, key, null, key.getBytes(StandardCharsets.UTF_8));
        }
        System.out.println("[x] Published " + keys.length + " keys");

        await(q1, 3);
        await(q2, 4);
        System.out.println("[q1] matched: " + new HashSet<>(q1));
        System.out.println("[q2] matched: " + new HashSet<>(q2));
        System.out.println("[v] lazy.pink.rabbit matched both q2 bindings → one copy; quick.brown.fox → dropped");
        connection.close();
    }

    private static void bind(Connection connection, Set<String> sink, String... patterns) throws Exception {
        Channel ch = connection.createChannel();
        String queue = ch.queueDeclare("", false, true, true, null).getQueue();
        for (String p : patterns) {
            ch.queueBind(queue, EXCHANGE, p);
        }
        ch.basicConsume(queue, true,
                (tag, d) -> sink.add(new String(d.getBody(), StandardCharsets.UTF_8)),
                tag -> { });
    }

    private static void await(Set<String> sink, int expected) throws InterruptedException {
        long deadline = System.currentTimeMillis() + 30_000;
        while (sink.size() < expected && System.currentTimeMillis() < deadline) {
            Thread.sleep(50);
        }
        if (sink.size() < expected) {
            throw new IllegalStateException("expected " + expected + ", saw " + Arrays.toString(sink.toArray()));
        }
    }
}

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

const EXCHANGE = "topic_logs";

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

async function bind(connection: ChannelModel, want: number, ...patterns: string[]): Promise<Set<string>> {
  const ch = await connection.createChannel();
  const q = await ch.assertQueue("", { exclusive: true });
  for (const p of patterns) await ch.bindQueue(q.queue, EXCHANGE, p);
  const seen = new Set<string>();
  await ch.consume(q.queue, (msg) => {
    if (msg) seen.add(msg.content.toString());
  }, { noAck: true });
  // Returns the live set; the caller waits until it reaches `want`.
  return Object.assign(seen, { want });
}

async function main(): Promise<void> {
  const connection = await amqp.connect(url());
  const channel = await connection.createChannel();
  await channel.assertExchange(EXCHANGE, "topic", { durable: false });

  const q1 = await bind(connection, 3, "*.orange.*");
  const q2 = await bind(connection, 4, "*.*.rabbit", "lazy.#");

  const keys = [
    "quick.orange.rabbit",  // q1 + q2
    "lazy.orange.elephant", // q1 + q2
    "quick.orange.fox",     // q1 only
    "lazy.brown.fox",       // q2 only
    "lazy.pink.rabbit",     // q2 only — matches BOTH q2 bindings, ONE copy
    "quick.brown.fox",      // nobody → silent drop
  ];
  for (const key of keys) channel.publish(EXCHANGE, key, Buffer.from(key));
  console.log(`[x] Published ${keys.length} keys`);

  await waitFor(q1, 3);
  await waitFor(q2, 4);
  console.log(`[q1] matched: ${[...q1].sort()}`);
  console.log(`[q2] matched: ${[...q2].sort()}`);
  console.log("[x] lazy.pink.rabbit matched both q2 bindings → one copy; quick.brown.fox → dropped");
  await connection.close();
}

function waitFor(set: Set<string>, n: number): Promise<void> {
  return new Promise((resolve, reject) => {
    const deadline = Date.now() + 30_000;
    const tick = setInterval(() => {
      if (set.size >= n) {
        clearInterval(tick);
        resolve();
      } else if (Date.now() > deadline) {
        clearInterval(tick);
        reject(new Error(`expected ${n}, saw ${set.size}`));
      }
    }, 50);
  });
}

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 exchange = "topic_logs";

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("topics");
await using var channel = await connection.CreateChannelAsync();
await channel.ExchangeDeclareAsync(exchange, ExchangeType.Topic, durable: false);

async Task<ConcurrentDictionary<string, byte>> Bind(params string[] patterns)
{
    var ch = await connection.CreateChannelAsync();
    var queue = (await ch.QueueDeclareAsync("", durable: false, exclusive: true, autoDelete: true)).QueueName;
    foreach (var p in patterns) await ch.QueueBindAsync(queue, exchange, p);
    var seen = new ConcurrentDictionary<string, byte>();
    var consumer = new AsyncEventingBasicConsumer(ch);
    consumer.ReceivedAsync += (_, ea) =>
    {
        seen.TryAdd(Encoding.UTF8.GetString(ea.Body.Span), 0);
        return Task.CompletedTask;
    };
    await ch.BasicConsumeAsync(queue, autoAck: true, consumer: consumer);
    return seen;
}

var q1 = await Bind("*.orange.*");
var q2 = await Bind("*.*.rabbit", "lazy.#");

string[] keys =
{
    "quick.orange.rabbit",  // q1 + q2
    "lazy.orange.elephant", // q1 + q2
    "quick.orange.fox",     // q1 only
    "lazy.brown.fox",       // q2 only
    "lazy.pink.rabbit",     // q2 only — matches BOTH q2 bindings, ONE copy
    "quick.brown.fox",      // nobody → silent drop
};
foreach (var key in keys)
    await channel.BasicPublishAsync(exchange, key, body: Encoding.UTF8.GetBytes(key));
Console.WriteLine($"[x] Published {keys.Length} keys");

await WaitFor(q1, 3);
await WaitFor(q2, 4);
Console.WriteLine($"[q1] matched: {string.Join(", ", q1.Keys.Order())}");
Console.WriteLine($"[q2] matched: {string.Join(", ", q2.Keys.Order())}");
Console.WriteLine("[v] lazy.pink.rabbit matched both q2 bindings → one copy; quick.brown.fox → dropped");

static async Task WaitFor(ConcurrentDictionary<string, byte> set, int n)
{
    var deadline = DateTime.UtcNow.AddSeconds(30);
    while (set.Count < n && DateTime.UtcNow < deadline) await Task.Delay(50);
    if (set.Count < n) throw new Exception($"expected {n}, saw {set.Count}");
}

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

EXCHANGE = "topic_logs"

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

ch = conn.create_channel
exchange = ch.topic(EXCHANGE, durable: false)

bind = lambda do |*patterns|
  sub_ch = conn.create_channel
  queue = sub_ch.queue("", exclusive: true)
  patterns.each { |p| queue.bind(exchange, routing_key: p) }
  seen = []
  mutex = Mutex.new
  queue.subscribe(manual_ack: false, block: false) { |_di, _props, body| mutex.synchronize { seen << body } }
  { seen: seen, mutex: mutex }
end

q1 = bind.call("*.orange.*")            # orange animals
q2 = bind.call("*.*.rabbit", "lazy.#")  # rabbits and everything lazy

keys = %w[
  quick.orange.rabbit lazy.orange.elephant quick.orange.fox
  lazy.brown.fox lazy.pink.rabbit quick.brown.fox
]
keys.each { |k| exchange.publish(k, routing_key: k) }
puts " [x] Published #{keys.size} keys"

wait_for = lambda do |entry, n|
  deadline = Time.now + 30
  loop do
    break if entry[:mutex].synchronize { entry[:seen].uniq.size } >= n || Time.now > deadline

    sleep 0.05
  end
end

wait_for.call(q1, 3)
wait_for.call(q2, 4)
puts " [q1] matched: #{q1[:seen].uniq.sort}"
puts " [q2] matched: #{q2[:seen].uniq.sort}"
puts " [x] lazy.pink.rabbit matched both q2 bindings → one copy; quick.brown.fox → dropped"

conn.close

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

const EXCHANGE: &str = "topic_logs";

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 bind(conn: &Connection, patterns: &[&str]) -> Result<lapin::Consumer, Box<dyn std::error::Error>> {
    let ch = conn.create_channel().await?;
    let queue = ch
        .queue_declare("", QueueDeclareOptions { exclusive: true, ..Default::default() }, FieldTable::default())
        .await?;
    for p in patterns {
        ch.queue_bind(queue.name().as_str(), EXCHANGE, p, QueueBindOptions::default(), FieldTable::default())
            .await?;
    }
    Ok(ch
        .basic_consume(queue.name().as_str(), "", BasicConsumeOptions { no_ack: true, ..Default::default() }, FieldTable::default())
        .await?)
}

async fn collect(label: &str, consumer: &mut lapin::Consumer, want: usize) -> Result<(), Box<dyn std::error::Error>> {
    let mut seen = BTreeSet::new();
    while seen.len() < want {
        let delivery = consumer.next().await.ok_or("consumer closed early")??;
        seen.insert(String::from_utf8_lossy(&delivery.data).into_owned());
    }
    println!("[{label}] matched: {seen:?}");
    Ok(())
}

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

    let mut q1 = bind(&conn, &["*.orange.*"]).await?;
    let mut q2 = bind(&conn, &["*.*.rabbit", "lazy.#"]).await?;

    let keys = [
        "quick.orange.rabbit",  // q1 + q2
        "lazy.orange.elephant", // q1 + q2
        "quick.orange.fox",     // q1 only
        "lazy.brown.fox",       // q2 only
        "lazy.pink.rabbit",     // q2 only — matches BOTH q2 bindings, ONE copy
        "quick.brown.fox",      // nobody → silent drop
    ];
    for key in keys {
        ch.basic_publish(EXCHANGE, key, BasicPublishOptions::default(), key.as_bytes(), BasicProperties::default())
            .await?;
    }
    println!("[x] Published {} keys", keys.len());

    collect("q1", &mut q1, 3).await?;
    collect("q2", &mut q2, 4).await?;
    println!("[x] lazy.pink.rabbit matched both q2 bindings → one copy; quick.brown.fox → dropped");

    conn.close(0, "done").await?;
    Ok(())
}