Getting Started

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

	// CONNECT: OPEN (SASL ANONYMOUS) + BEGIN one session.
	conn, err := amqp.Dial(ctx, amqpURL(), nil)
	if err != nil {
		log.Fatalf("dial: %v", err)
	}
	defer func() { _ = conn.Close() }()
	session, err := conn.NewSession(ctx, nil)
	if err != nil {
		log.Fatalf("new session: %v", err)
	}

	// SEND: attach a sender; each unsettled Send blocks for the accepted disposition.
	sender, err := session.NewSender(ctx, addr, nil)
	if err != nil {
		log.Fatalf("new sender: %v", err)
	}
	for i := 0; i < total; i++ {
		body := fmt.Sprintf("msg-%03d", i)
		if err := sender.Send(ctx, amqp.NewMessage([]byte(body)), nil); err != nil {
			log.Fatalf("send %s: %v", body, err)
		}
	}
	_ = sender.Close(ctx)
	fmt.Printf("[send] Produced %d messages to %s\n", total, addr)

	// RECEIVE: attach a receiver with credit; accept each => removed from the queue.
	receiver, err := session.NewReceiver(ctx, addr, &amqp.ReceiverOptions{Credit: 10})
	if err != nil {
		log.Fatalf("new receiver: %v", err)
	}
	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("[recv] Consumed and accepted %d messages (no loss)\n", len(seen))
	_ = 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

    # CONNECT: OPEN (SASL ANONYMOUS) — proton sends a non-empty container-id.
    conn = BlockingConnection(amqp_url())
    try:
        # SEND: attach a sender; each send 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()
        print(f"[send] Produced {TOTAL} messages to {addr}")

        # RECEIVE: attach a receiver with credit; accept each => removed from the queue.
        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"[recv] Consumed and accepted {len(seen)} messages (no loss)")
        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 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;

        // CONNECT: OPEN (SASL ANONYMOUS). The JMS destination name IS the node address.
        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);

                // SEND: a producer is a server-receiver link; send blocks for accepted.
                try (MessageProducer producer = session.createProducer(queue)) {
                    for (int i = 0; i < TOTAL; i++) {
                        producer.send(session.createTextMessage(String.format("msg-%03d", i)));
                    }
                }
                System.out.printf("[send] Produced %d messages to %s%n", TOTAL, address);

                // RECEIVE: a CLIENT_ACKNOWLEDGE consumer; acknowledge() settles accepted.
                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 before draining the queue");
                        }
                        String body = msg.getBody(String.class);
                        msg.acknowledge();
                        seen.add(body);
                    }
                    System.out.printf("[recv] Consumed and accepted %d messages (no loss)%n", seen.size());
                }
            }
        }
    }
}

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;

// CONNECT: OPEN (SASL ANONYMOUS) + one session.
var connection = await Connection.Factory.CreateAsync(new Address(AmqpUrl()));
try
{
    var session = new Session(connection);

    // SEND: a SenderLink is a server-receiver link; Send blocks for accepted.
    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));
    }
    Console.WriteLine($"[send] Produced {total} messages to {addr}");

    // RECEIVE: grant credit; Accept each => removed from the queue. Keep the
    // sender open through the consume phase (detaching it early can stall delivery).
    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($"[recv] Consumed and accepted {seen.Count} messages (no loss)");

    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 ConnectionOptions,
  type EventContext,
  type Receiver,
} from "rhea-promise";

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

function connectionOptions(): ConnectionOptions {
  const url = new URL(process.env["KUBEMQ_AMQP_URL"] ?? "amqp://localhost:5672");
  return {
    host: url.hostname,
    port: url.port ? Number(url.port) : 5672,
    container_id: `kubemq-amqp10-js-${process.pid}`,
    reconnect: false,
  };
}

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

async function main(): Promise<void> {
  const address = `queues/${channel}`;

  // CONNECT: OPEN (SASL ANONYMOUS).
  const connection = new Connection(connectionOptions());
  await connection.open();
  try {
    // SEND: 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();
    console.log(`[send] Produced ${total} messages to ${address}`);

    // RECEIVE: manual credit + manual settle; accept() removes from the queue.
    const receiver = await connection.createReceiver({
      source: { address },
      credit_window: 0,
      autoaccept: false,
      autosettle: false,
    });
    const seen = new Set<string>();
    await new Promise<void>((resolve, reject) => {
      receiver.on(ReceiverEvents.message, (ctx: EventContext) => {
        ctx.delivery?.accept();
        seen.add(bodyToString(ctx.message?.body));
        if (seen.size >= total) resolve();
      });
      receiver.on(ReceiverEvents.receiverError, (ctx: EventContext) =>
        reject(ctx.receiver?.error ?? new Error("receiver error")));
      (receiver as Receiver).addCredit(total);
    });
    console.log(`[recv] Consumed and accepted ${seen.size} messages (no loss)`);
    await receiver.close();
  } finally {
    await connection.close();
  }
}

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

use std::collections::HashSet;

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_string(msg: &Message<Body<Value>>) -> String {
    match &msg.body {
        Body::Data(batch) => batch.iter().flat_map(|d| d.0.iter().copied()).collect::<Vec<u8>>(),
        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(),
    }
    .into_iter()
    .map(|b| b as char)
    .collect()
}

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

    // CONNECT: OPEN (SASL ANONYMOUS) + BEGIN one session.
    let mut connection = Connection::open("amqp10-examples-basic", amqp_url().as_str()).await?;
    let mut session = Session::begin(&mut connection).await?;

    // SEND: a sender pinned to Unsettled (at-least-once); the default `mixed` is rejected.
    let mut sender = Sender::builder()
        .name("basic-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 {i}: unexpected outcome {outcome:?}").into());
        }
    }
    sender.close().await?;
    println!("[send] Produced {TOTAL} messages to {addr}");

    // RECEIVE: client-granted auto credit; accept each => removed from the queue.
    let mut receiver = Receiver::builder()
        .name("basic-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?;
        receiver.accept(&delivery).await?;
        seen.insert(body_string(delivery.message()));
    }
    println!("[recv] Consumed and accepted {} messages (no loss)", seen.len());

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