RabbitMQ (AMQP 0-9-1)

package main

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

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

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

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

	ch, err := conn.Channel()
	if err != nil {
		log.Fatalf("channel: %v", err)
	}
	defer func() { _ = ch.Close() }()

	// Declare "hello" → KubeMQ Queue channel amqp.default.hello.
	q, err := ch.QueueDeclare("hello", false, false, false, false, nil)
	if err != nil {
		log.Fatalf("declare queue: %v", err)
	}

	ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
	defer cancel()

	// Publish on the default exchange — routing key = queue name.
	if err := ch.PublishWithContext(ctx, "", q.Name, false, false, amqp.Publishing{
		ContentType: "text/plain",
		Body:        []byte("Hello World!"),
	}); err != nil {
		log.Fatalf("publish: %v", err)
	}
	log.Printf(" [x] Sent %q", "Hello World!")

	// Consume with auto-ack and print the exact body.
	msgs, err := ch.Consume(q.Name, "", true, false, false, false, nil)
	if err != nil {
		log.Fatalf("consume: %v", err)
	}
	select {
	case d := <-msgs:
		log.Printf(" [x] Received %q", string(d.Body))
	case <-ctx.Done():
		log.Fatalf("timed out: %v", ctx.Err())
	}
}

import os

import pika

URL = os.environ.get("KUBEMQ_AMQP_URL", "amqp://guest:guest@localhost:5672/")
QUEUE = "hello"  # → KubeMQ Queue channel amqp.default.hello


def main() -> None:
    connection = pika.BlockingConnection(pika.URLParameters(URL))
    channel = connection.channel()

    # Default (nameless) exchange: routing key == queue name.
    channel.queue_declare(queue=QUEUE, durable=False, exclusive=False, auto_delete=False)

    channel.basic_publish(
        exchange="",
        routing_key=QUEUE,
        body=b"Hello World!",
        properties=pika.BasicProperties(content_type="text/plain"),
    )
    print(" [x] Sent 'Hello World!'")

    method, _props, body = channel.basic_get(queue=QUEUE, auto_ack=True)
    if method is None:
        raise SystemExit("no message received from queue")
    print(f" [x] Received {body.decode()!r}")

    channel.close()
    connection.close()


if __name__ == "__main__":
    main()

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;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;

public final class Main {

    private static final String QUEUE = "hello"; // → amqp.default.hello

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

        try (Connection connection = factory.newConnection();
                Channel channel = connection.createChannel()) {

            // queue.declare: non-durable, non-exclusive, no auto-delete.
            channel.queueDeclare(QUEUE, false, false, false, null);

            // Publish via the default exchange ("") with routing key = queue name.
            AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
                    .contentType("text/plain")
                    .build();
            channel.basicPublish("", QUEUE, props, "Hello World!".getBytes("UTF-8"));
            System.out.println("[x] Sent 'Hello World!'");

            BlockingQueue<String> received = new ArrayBlockingQueue<>(1);
            DeliverCallback onDeliver = (tag, delivery) ->
                    received.offer(new String(delivery.getBody(), "UTF-8"));
            channel.basicConsume(QUEUE, true, onDeliver, tag -> { });

            String got = received.poll(15, TimeUnit.SECONDS);
            if (got == null) {
                throw new IllegalStateException("timed out waiting for the message");
            }
            System.out.println("[x] Received '" + got + "'");
        }
    }
}

import amqp from "amqplib";

// amqplib reads the URL path as the vhost; the bare trailing "/" in the dev URL
// resolves to the default vhost (KubeMQ segment "default").
const URL = process.env.KUBEMQ_AMQP_URL ?? "amqp://guest:guest@localhost:5672/";
const QUEUE = "hello"; // → KubeMQ Queue channel amqp.default.hello

async function main() {
  const conn = await amqp.connect(URL);
  const ch = await conn.createChannel();

  await ch.assertQueue(QUEUE, { durable: false, autoDelete: false, exclusive: false });

  const received = new Promise((resolve) => {
    ch.consume(
      QUEUE,
      (msg) => {
        if (msg === null) return;
        console.log(`[x] Received: ${msg.content.toString()}`);
        resolve();
      },
      { noAck: true },
    );
  });

  // Default exchange ("") routes by queue name.
  ch.publish("", QUEUE, Buffer.from("Hello World!"), { contentType: "text/plain" });
  console.log("[x] Sent:     Hello World!");

  await received;
  await ch.close();
  await conn.close();
}

main().catch((err) => {
  console.error("publish failed:", err);
  process.exit(1);
});

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

const string queueName = "hello"; // → KubeMQ Queue channel amqp.default.hello

var factory = new ConnectionFactory
{
    Uri = new Uri(Environment.GetEnvironmentVariable("KUBEMQ_AMQP_URL")
                  ?? "amqp://guest:guest@localhost:5672/"),
};

using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30));
var ct = cts.Token;

await using var connection = await factory.CreateConnectionAsync(ct);
await using var channel = await connection.CreateChannelAsync(cancellationToken: ct);

// Non-durable, not exclusive, not auto-delete — a plain shared queue.
await channel.QueueDeclareAsync(queueName, durable: false, exclusive: false,
    autoDelete: false, arguments: null, cancellationToken: ct);

var received = new TaskCompletionSource<string>(TaskCreationOptions.RunContinuationsAsynchronously);
var consumer = new AsyncEventingBasicConsumer(channel);
consumer.ReceivedAsync += (_, ea) =>
{
    received.TrySetResult(Encoding.UTF8.GetString(ea.Body.Span));
    return Task.CompletedTask;
};
await channel.BasicConsumeAsync(queueName, autoAck: true, consumer: consumer, cancellationToken: ct);

// Publish to the default exchange with routing key = queue name.
var props = new BasicProperties { ContentType = "text/plain" };
await channel.BasicPublishAsync(exchange: "", routingKey: queueName, mandatory: false,
    basicProperties: props, body: Encoding.UTF8.GetBytes("Hello World!"), cancellationToken: ct);
Console.WriteLine("[x] Sent 'Hello World!'");

var message = await received.Task.WaitAsync(ct);
Console.WriteLine($"[x] Received '{message}'");

# frozen_string_literal: true
require "bunny"

URL = ENV.fetch("KUBEMQ_AMQP_URL", "amqp://guest:guest@localhost:5672/")
QUEUE = "hello" # → KubeMQ Queue channel amqp.default.hello

conn = Bunny.new(URL)
conn.start
ch = conn.create_channel

# Publisher confirms: wait for the broker to accept and route the publish before
# reading it back, so a fire-and-forget publish is never lost in flight.
ch.confirm_select

queue = ch.queue(QUEUE, durable: false, auto_delete: false, exclusive: false)

# Default ("") exchange routes by queue name.
ch.default_exchange.publish("Hello World!", routing_key: queue.name, content_type: "text/plain")
ch.wait_for_confirms
puts " [x] Sent 'Hello World!'"

body = nil
deadline = Time.now + 5
while body.nil? && Time.now < deadline
  _info, _props, body = queue.pop(manual_ack: false)
  sleep 0.2 if body.nil?
end
puts " [x] Received '#{body}'"

conn.close

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

const QUEUE: &str = "hello"; // → KubeMQ Queue channel amqp.default.hello

// lapin reads the URL path as the vhost and treats a bare trailing "/" as an
// empty vhost, which the connector rejects. The default "/" vhost must be
// percent-encoded as "%2f".
fn amqp_url() -> String {
    let url = std::env::var("KUBEMQ_AMQP_URL")
        .unwrap_or_else(|_| "amqp://guest:guest@localhost:5672/".into());
    let host = url.rsplit_once('@').map_or(url.as_str(), |(_, h)| h);
    if host.ends_with('/') && !host.ends_with("/%2f") {
        format!("{url}%2f")
    } else {
        url
    }
}

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

    channel
        .queue_declare(QUEUE, QueueDeclareOptions::default(), FieldTable::default())
        .await?;

    // Publish to the default exchange, routing key = queue name.
    channel
        .basic_publish(
            "",
            QUEUE,
            BasicPublishOptions::default(),
            b"Hello World!",
            BasicProperties::default().with_content_type("text/plain".into()),
        )
        .await?
        .await?;
    println!("[x] Sent 'Hello World!'");

    let mut consumer = channel
        .basic_consume(
            QUEUE,
            "hello-consumer",
            BasicConsumeOptions { no_ack: true, ..Default::default() },
            FieldTable::default(),
        )
        .await?;

    if let Some(delivery) = consumer.next().await {
        let delivery = delivery?;
        println!("[x] Received '{}'", String::from_utf8_lossy(&delivery.data));
    }

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