Routing (Direct)

package main

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

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

const exchange = "direct_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, key string) <-chan amqp.Delivery {
	ch, err := conn.Channel()
	if err != nil {
		log.Fatalf("channel for %s: %v", key, err)
	}
	q, err := ch.QueueDeclare("", false, false, true, false, nil) // server-named, exclusive
	if err != nil {
		log.Fatalf("declare %s: %v", key, err)
	}
	if err := ch.QueueBind(q.Name, key, exchange, false, nil); err != nil { // bind on the exact key
		log.Fatalf("bind %s: %v", key, err)
	}
	msgs, err := ch.Consume(q.Name, "", true, false, false, false, nil)
	if err != nil {
		log.Fatalf("consume %s: %v", key, err)
	}
	return msgs
}

func main() {
	ctx, cancel := context.WithTimeout(context.Background(), 15*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, "direct", false, false, false, false, nil); err != nil {
		log.Fatalf("declare exchange: %v", err)
	}

	errorMsgs := bindConsumer(conn, "error")
	infoMsgs := bindConsumer(conn, "info")

	for key, body := range map[string]string{
		"error": "an error happened",
		"info":  "all is well",
		"debug": "nobody is bound to debug", // silently dropped (no binding)
	} {
		if err := ch.PublishWithContext(ctx, exchange, key, false, false, amqp.Publishing{
			ContentType: "text/plain",
			Body:        []byte(body),
		}); err != nil {
			log.Fatalf("publish key=%s: %v", key, err)
		}
	}
	log.Printf(" [x] Published keys: error, info, debug")

	select {
	case d := <-errorMsgs:
		log.Printf(" [error] received %q (key=%q)", d.Body, d.RoutingKey)
	case <-ctx.Done():
		log.Fatal("error consumer timed out")
	}
	select {
	case d := <-infoMsgs:
		log.Printf(" [info]  received %q (key=%q)", d.Body, d.RoutingKey)
	case <-ctx.Done():
		log.Fatal("info consumer timed out")
	}
	log.Printf(" [✓] debug (unbound) → silently dropped (no consumer)")
}

import os

import pika

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


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

    consumers = {}
    for key in ("error", "info"):
        queue = ch.queue_declare(queue="", exclusive=True).method.queue
        ch.queue_bind(exchange=EXCHANGE, queue=queue, routing_key=key)  # bind on the exact key
        consumers[key] = queue

    for key, body in {
        "error": "an error happened",
        "info": "all is well",
        "debug": "nobody is bound to debug",  # silently dropped (no binding)
    }.items():
        ch.basic_publish(exchange=EXCHANGE, routing_key=key, body=body.encode())
    print(" [x] Published keys: error, info, debug")

    for key, queue in consumers.items():
        for method, _props, body in ch.consume(queue, inactivity_timeout=10, auto_ack=True):
            if method is None:
                raise SystemExit(f"{key} consumer timed out")
            print(f" [{key}] received {body.decode()!r} (key={method.routing_key})")
            break
        ch.cancel()

    print(" [✓] debug (unbound) → silently dropped (no consumer)")
    ch.close()
    conn.close()


if __name__ == "__main__":
    main()

import java.nio.charset.StandardCharsets;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
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 = "direct_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("/");
        }

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

            Map<String, BlockingQueue<String>> received = new LinkedHashMap<>();
            for (String key : new String[] {"error", "info"}) {
                Channel ch = connection.createChannel();
                String queue = ch.queueDeclare("", false, true, true, null).getQueue();
                ch.queueBind(queue, EXCHANGE, key); // bind on the exact key
                BlockingQueue<String> sink = new ArrayBlockingQueue<>(4);
                ch.basicConsume(queue, true,
                        (tag, d) -> sink.offer(new String(d.getBody(), StandardCharsets.UTF_8)),
                        tag -> { });
                received.put(key, sink);
            }

            Map<String, String> publishes = new LinkedHashMap<>();
            publishes.put("error", "an error happened");
            publishes.put("info", "all is well");
            publishes.put("debug", "nobody is bound to debug"); // silently dropped
            for (Map.Entry<String, String> e : publishes.entrySet()) {
                channel.basicPublish(EXCHANGE, e.getKey(), null,
                        e.getValue().getBytes(StandardCharsets.UTF_8));
            }
            System.out.println("[x] Published keys: error, info, debug");

            for (Map.Entry<String, BlockingQueue<String>> e : received.entrySet()) {
                String body = e.getValue().poll(10, TimeUnit.SECONDS);
                if (body == null) throw new IllegalStateException(e.getKey() + " consumer timed out");
                System.out.println("[" + e.getKey() + "] received " + body);
            }
            System.out.println("[v] debug (unbound) → silently dropped (no consumer)");
        }
    }
}

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

const EXCHANGE = "direct_logs";

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

function once(ch: Channel, queue: string): Promise<string> {
  return new Promise((resolve) => {
    ch.consume(queue, (msg) => {
      if (msg) resolve(msg.content.toString());
    }, { noAck: true });
  });
}

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

  const consumers: Record<string, Promise<string>> = {};
  for (const key of ["error", "info"]) {
    const ch = await connection.createChannel();
    const q = await ch.assertQueue("", { exclusive: true });
    await ch.bindQueue(q.queue, EXCHANGE, key); // bind on the exact key
    consumers[key] = once(ch, q.queue);
  }

  const publishes: Record<string, string> = {
    error: "an error happened",
    info: "all is well",
    debug: "nobody is bound to debug", // silently dropped (no binding)
  };
  for (const [key, body] of Object.entries(publishes)) {
    channel.publish(EXCHANGE, key, Buffer.from(body));
  }
  console.log("[x] Published keys: error, info, debug");

  console.log(`[error] received ${await consumers["error"]}`);
  console.log(`[info]  received ${await consumers["info"]}`);
  console.log("[v] debug (unbound) → silently dropped (no consumer)");
  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 exchange = "direct_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("routing-direct");
await using var channel = await connection.CreateChannelAsync();
await channel.ExchangeDeclareAsync(exchange, ExchangeType.Direct, durable: false);

var received = new Dictionary<string, BlockingCollection<string>>();
foreach (var key in new[] { "error", "info" })
{
    var ch = await connection.CreateChannelAsync();
    var queue = (await ch.QueueDeclareAsync("", durable: false, exclusive: true, autoDelete: true)).QueueName;
    await ch.QueueBindAsync(queue, exchange, key); // bind on the exact key
    var sink = new BlockingCollection<string>();
    var consumer = new AsyncEventingBasicConsumer(ch);
    consumer.ReceivedAsync += (_, ea) =>
    {
        sink.Add(Encoding.UTF8.GetString(ea.Body.Span));
        return Task.CompletedTask;
    };
    await ch.BasicConsumeAsync(queue, autoAck: true, consumer: consumer);
    received[key] = sink;
}

var publishes = new Dictionary<string, string>
{
    ["error"] = "an error happened",
    ["info"] = "all is well",
    ["debug"] = "nobody is bound to debug", // silently dropped (no binding)
};
foreach (var (key, body) in publishes)
    await channel.BasicPublishAsync(exchange, key, body: Encoding.UTF8.GetBytes(body));
Console.WriteLine("[x] Published keys: error, info, debug");

foreach (var (key, sink) in received)
    Console.WriteLine($"[{key}] received {sink.Take()}");
Console.WriteLine("[v] debug (unbound) → silently dropped (no consumer)");

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

EXCHANGE = "direct_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.direct(EXCHANGE, durable: false)

consumers = {}
%w[error info].each do |key|
  sub_ch = conn.create_channel
  queue = sub_ch.queue("", exclusive: true)
  queue.bind(exchange, routing_key: key) # bind on the exact key
  sink = Queue.new
  queue.subscribe(manual_ack: false, block: false) { |_di, _props, body| sink.push(body) }
  consumers[key] = sink
end

{
  "error" => "an error happened",
  "info"  => "all is well",
  "debug" => "nobody is bound to debug" # silently dropped (no binding)
}.each { |key, body| exchange.publish(body, routing_key: key) }
puts " [x] Published keys: error, info, debug"

consumers.each { |key, sink| puts " [#{key}] received #{sink.pop.inspect}" }
puts " [x] debug (unbound) → silently dropped (no consumer)"

conn.close

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

const EXCHANGE: &str = "direct_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_consumer(conn: &Connection, key: &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?;
    ch.queue_bind(queue.name().as_str(), EXCHANGE, key, QueueBindOptions::default(), FieldTable::default())
        .await?; // bind on the exact key
    Ok(ch
        .basic_consume(queue.name().as_str(), "", BasicConsumeOptions { no_ack: true, ..Default::default() }, FieldTable::default())
        .await?)
}

#[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::Direct, ExchangeDeclareOptions::default(), FieldTable::default())
        .await?;

    let mut error_c = bind_consumer(&conn, "error").await?;
    let mut info_c = bind_consumer(&conn, "info").await?;

    for (key, body) in [
        ("error", "an error happened"),
        ("info", "all is well"),
        ("debug", "nobody is bound to debug"), // silently dropped (no binding)
    ] {
        ch.basic_publish(EXCHANGE, key, BasicPublishOptions::default(), body.as_bytes(), BasicProperties::default())
            .await?;
    }
    println!("[x] Published keys: error, info, debug");

    let err = error_c.next().await.ok_or("error consumer closed")??;
    println!("[error] received {}", String::from_utf8_lossy(&err.data));
    let info = info_c.next().await.ok_or("info consumer closed")??;
    println!("[info]  received {}", String::from_utf8_lossy(&info.data));
    println!("[x] debug (unbound) → silently dropped (no consumer)");

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