Queues

package main

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

	amqp "github.com/Azure/go-amqp"
)

const channel = "amqp10.examples.basic"
const total = 10

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

func main() {
	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
	defer cancel()
	addr := "queues/" + channel // queues/ prefix → KubeMQ Queues pattern

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

	// 1. Produce — each Send is unsettled and blocks for the accepted DISPOSITION.
	sender, _ := session.NewSender(ctx, addr, nil)
	for i := 0; i < total; i++ {
		if err := sender.Send(ctx, amqp.NewMessage([]byte(fmt.Sprintf("msg-%03d", i))), nil); err != nil {
			log.Fatalf("send: %v", err)
		}
	}
	_ = sender.Close(ctx)

	// 2. Consume — grant credit, Receive, AcceptMessage (⇒ AckRange, removed).
	receiver, _ := session.NewReceiver(ctx, addr, &amqp.ReceiverOptions{Credit: 10})
	seen := make(map[string]struct{}, total)
	for len(seen) < total {
		msg, err := receiver.Receive(ctx, nil)
		if err != nil {
			log.Fatalf("receive: %v", err)
		}
		if err := receiver.AcceptMessage(ctx, msg); err != nil {
			log.Fatalf("accept: %v", err)
		}
		seen[string(msg.GetData())] = struct{}{}
	}
	fmt.Printf("consumed and accepted %d messages\n", len(seen))

	// 3. Assert the queue is empty — a further Receive must time out.
	emptyCtx, emptyCancel := context.WithTimeout(ctx, 2*time.Second)
	if _, err := receiver.Receive(emptyCtx, nil); err == nil {
		log.Fatal("expected an empty queue")
	}
	emptyCancel()
	_ = receiver.Close(ctx)
}

import os

from proton import Message
from proton.utils import BlockingConnection

CHANNEL = "amqp10.examples.basic"
TOTAL = 10


def amqp_url() -> str:
    return os.environ.get("KUBEMQ_AMQP_URL", "amqp://localhost:5672")


def main() -> None:
    addr = "queues/" + CHANNEL  # queues/ prefix → KubeMQ Queues pattern
    conn = BlockingConnection(amqp_url())
    try:
        # 1. Produce — each send is unsettled and blocks for the accepted DISPOSITION.
        sender = conn.create_sender(addr)
        for i in range(TOTAL):
            sender.send(Message(body=f"msg-{i:03d}"))
        sender.close()

        # 2. Consume — grant credit, receive, accept (⇒ AckRange, removed).
        receiver = conn.create_receiver(addr, credit=10)
        seen: set[str] = set()
        while len(seen) < TOTAL:
            msg = receiver.receive(timeout=30.0)
            receiver.accept()
            seen.add(str(msg.body))
        print(f"consumed and accepted {len(seen)} messages")

        # 3. Assert the queue is empty — a further receive must time out.
        try:
            receiver.receive(timeout=2.0)
        except Exception:
            pass  # the EXPECTED idle timeout on an empty queue
        else:
            raise SystemExit("expected an empty queue")
        receiver.close()
    finally:
        conn.close()


if __name__ == "__main__":
    main()

import java.util.HashSet;
import java.util.Set;

import javax.jms.Connection;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.qpid.jms.JmsConnectionFactory;

public final class Main {
    private static final String CHANNEL = "amqp10.examples.basic";
    private static final int TOTAL = 10;

    public static void main(String[] args) throws Exception {
        String url = System.getenv().getOrDefault("KUBEMQ_AMQP_URL", "amqp://localhost:5672");
        String address = "queues/" + CHANNEL; // queues/ prefix → KubeMQ Queues pattern

        JmsConnectionFactory factory = new JmsConnectionFactory(url);
        try (Connection connection = factory.createConnection()) {
            connection.start();
            try (Session session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE)) {
                Queue queue = session.createQueue(address);

                // 1. Produce — a MessageProducer is a server-receiver link; each send
                //    blocks for the accepted DISPOSITION (unsettled, at-least-once).
                try (MessageProducer producer = session.createProducer(queue)) {
                    for (int i = 0; i < TOTAL; i++) {
                        producer.send(session.createTextMessage(String.format("msg-%03d", i)));
                    }
                }

                // 2. Consume — a CLIENT_ACKNOWLEDGE consumer; acknowledge() settles
                //    `accepted` ⇒ AckRange (removed from the queue).
                try (MessageConsumer consumer = session.createConsumer(queue)) {
                    Set<String> seen = new HashSet<>();
                    while (seen.size() < TOTAL) {
                        Message msg = consumer.receive(30_000);
                        if (msg == null) throw new IllegalStateException("timed out");
                        String body = msg.getBody(String.class);
                        msg.acknowledge();
                        seen.add(body);
                    }
                    System.out.printf("consumed and accepted %d messages%n", seen.size());

                    // 3. Assert the queue is empty — a further receive times out.
                    if (consumer.receive(2_000) != null) {
                        throw new IllegalStateException("expected an empty queue");
                    }
                }
            }
        }
    }
}

using System.Text;
using Amqp;
using Amqp.Framing;

const string channel = "amqp10.examples.basic";
const int total = 10;

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

var addr = "queues/" + channel; // queues/ prefix → KubeMQ Queues pattern

var connection = await Connection.Factory.CreateAsync(new Address(AmqpUrl()));
try
{
    var session = new Session(connection);

    // 1. Produce — each Send is unsettled and blocks for the accepted DISPOSITION.
    var sender = new SenderLink(session, "basic-sender", addr);
    for (var i = 0; i < total; i++)
    {
        var message = new Message { BodySection = new Data { Binary = Encoding.UTF8.GetBytes($"msg-{i:D3}") } };
        sender.Send(message, TimeSpan.FromSeconds(15));
    }

    // 2. Consume — grant credit, Receive, Accept (⇒ AckRange, removed).
    //    Keep the sender open through the consume phase (see the README gotcha).
    var receiver = new ReceiverLink(session, "basic-receiver", addr);
    receiver.SetCredit(10, autoRestore: true);
    var seen = new HashSet<string>();
    while (seen.Count < total)
    {
        var message = receiver.Receive(TimeSpan.FromSeconds(30))
            ?? throw new InvalidOperationException("receive timed out");
        receiver.Accept(message);
        seen.Add(BodyString(message));
    }
    Console.WriteLine($"consumed and accepted {seen.Count} messages");

    // 3. Assert the queue is empty — a further Receive must time out (null).
    if (receiver.Receive(TimeSpan.FromSeconds(2)) is not null)
        throw new InvalidOperationException("expected an empty queue");

    await receiver.CloseAsync();
    await sender.CloseAsync();
    await session.CloseAsync();
}
finally
{
    await connection.CloseAsync();
}

static string BodyString(Message message) => message.BodySection switch
{
    Data d => Encoding.UTF8.GetString(d.Binary),
    AmqpValue { Value: byte[] bytes } => Encoding.UTF8.GetString(bytes),
    AmqpValue { Value: string str } => str,
    AmqpValue v => v.Value?.ToString() ?? string.Empty,
    _ => string.Empty,
};

import {
  Connection,
  ReceiverEvents,
  type EventContext,
  type Receiver,
} from "rhea-promise";

const channel = "amqp10.examples.basic";
const total = 10;

function bodyToString(body: unknown): string {
  return Buffer.isBuffer(body) ? body.toString("utf8") : String(body);
}

async function main(): Promise<void> {
  const url = new URL(process.env["KUBEMQ_AMQP_URL"] ?? "amqp://localhost:5672");
  const address = `queues/${channel}`; // queues/ prefix → KubeMQ Queues pattern

  const connection = new Connection({
    host: url.hostname,
    port: url.port ? Number(url.port) : 5672,
    container_id: `kubemq-amqp10-js-basic-${process.pid}`,
    reconnect: false,
  });
  await connection.open();

  try {
    // 1. Produce — an AwaitableSender; each send() resolves on the accepted DISPOSITION.
    const sender = await connection.createAwaitableSender({ target: { address } });
    for (let i = 0; i < total; i++) {
      await sender.send({ body: `msg-${String(i).padStart(3, "0")}` }, { timeoutInSeconds: 15 });
    }
    await sender.close();

    // 2. Consume — grant credit manually, settle manually. Register the handler
    //    BEFORE addCredit so early deliveries are not missed.
    const receiver = await connection.createReceiver({
      source: { address },
      credit_window: 0,
      autoaccept: false,
      autosettle: false,
    });
    const seen = new Set<string>();
    await receiveUntil(receiver, (ctx) => {
      ctx.delivery?.accept(); // accept ⇒ AckRange ⇒ removed from the queue
      seen.add(bodyToString(ctx.message?.body));
      return seen.size >= total;
    }, total + 1, 30_000);
    console.log(`consumed and accepted ${seen.size} messages`);

    await receiver.close();
  } finally {
    await connection.close();
  }
}

function receiveUntil(
  receiver: Receiver,
  onMessage: (ctx: EventContext) => boolean,
  credit: number,
  timeoutMs: number,
): Promise<void> {
  return new Promise<void>((resolve, reject) => {
    const timer = setTimeout(() => {
      receiver.removeListener(ReceiverEvents.message, handler);
      reject(new Error("timed out waiting for messages"));
    }, timeoutMs);
    const handler = (ctx: EventContext): void => {
      if (onMessage(ctx)) {
        clearTimeout(timer);
        receiver.removeListener(ReceiverEvents.message, handler);
        resolve();
      }
    };
    receiver.on(ReceiverEvents.message, handler);
    receiver.addCredit(credit);
  });
}

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

use std::collections::HashSet;
use std::time::Duration;

use fe2o3_amqp::link::delivery::Delivery;
use fe2o3_amqp::link::receiver::CreditMode;
use fe2o3_amqp::{Connection, Receiver, Sender, Session};
use fe2o3_amqp_types::definitions::SenderSettleMode;
use fe2o3_amqp_types::messaging::{Body, Message};
use fe2o3_amqp_types::primitives::Value;

const CHANNEL: &str = "amqp10.examples.basic";
const TOTAL: usize = 10;

fn amqp_url() -> String {
    std::env::var("KUBEMQ_AMQP_URL").unwrap_or_else(|_| "amqp://localhost:5672".to_string())
}

fn body_bytes(msg: &Message<Body<Value>>) -> Vec<u8> {
    match &msg.body {
        Body::Data(batch) => batch.iter().flat_map(|d| d.0.to_vec()).collect(),
        Body::Value(v) => match &v.0 {
            Value::Binary(b) => b.to_vec(),
            Value::String(s) => s.clone().into_bytes(),
            other => format!("{other:?}").into_bytes(),
        },
        _ => Vec::new(),
    }
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let addr = format!("queues/{CHANNEL}"); // queues/ prefix → KubeMQ Queues pattern

    let mut connection = Connection::open("amqp10-examples-basic", amqp_url().as_str()).await?;
    let mut session = Session::begin(&mut connection).await?;

    // 1. Produce — pin SenderSettleMode::Unsettled (the connector rejects the AMQP
    //    default `mixed`); each send is at-least-once and blocks for the Accepted outcome.
    let mut sender = Sender::builder()
        .name("basic-send-receive-sender")
        .target(addr.as_str())
        .sender_settle_mode(SenderSettleMode::Unsettled)
        .attach(&mut session)
        .await?;
    for i in 0..TOTAL {
        let outcome = sender.send(format!("msg-{i:03}")).await?;
        if !outcome.is_accepted() {
            return Err(format!("send msg-{i:03}: unexpected outcome {outcome:?}").into());
        }
    }
    sender.close().await?;

    // 2. Consume — CreditMode::Auto(10) issues 10 credits and replenishes on settle;
    //    accept ⇒ AckRange ⇒ removed from the queue.
    let mut receiver = Receiver::builder()
        .name("basic-send-receive-receiver")
        .source(addr.as_str())
        .credit_mode(CreditMode::Auto(10))
        .attach(&mut session)
        .await?;
    let mut seen: HashSet<String> = HashSet::with_capacity(TOTAL);
    while seen.len() < TOTAL {
        let delivery: Delivery<Body<Value>> = receiver.recv().await?;
        let body = String::from_utf8_lossy(&body_bytes(delivery.message())).into_owned();
        receiver.accept(&delivery).await?;
        seen.insert(body);
    }
    println!("consumed and accepted {} messages", seen.len());

    // 3. Assert the queue is empty — a further recv must time out.
    if let Ok(Ok(_)) = tokio::time::timeout(Duration::from_secs(2), receiver.recv::<Body<Value>>()).await {
        return Err("expected an empty queue".into());
    }

    receiver.close().await?;
    session.end().await?;
    connection.close().await?;
    Ok(())
}