AMQP 1.0

package main

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

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

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(), 30*time.Second)
	defer cancel()

	addr := "queues/amqp10.examples.basic"

	// OPEN: SASL ANONYMOUS by default (no userinfo in the URL).
	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)
	}

	// ATTACH a sender (server-receiver link) and send one unsettled message.
	sender, err := session.NewSender(ctx, addr, nil)
	if err != nil {
		log.Fatalf("new sender: %v", err)
	}
	if err := sender.Send(ctx, amqp.NewMessage([]byte("hello from AMQP 1.0")), nil); err != nil {
		log.Fatalf("send: %v", err)
	}
	_ = sender.Close(ctx)
	fmt.Printf("sent 1 message to %s (accepted)\n", addr)
}

import os

from proton import Message
from proton.utils import BlockingConnection


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


def main() -> None:
    addr = "queues/amqp10.examples.basic"

    # OPEN: SASL ANONYMOUS by default (no userinfo in the URL).
    conn = BlockingConnection(amqp_url())
    try:
        # ATTACH a sender; each send blocks for the accepted disposition.
        sender = conn.create_sender(addr)
        sender.send(Message(body="hello from AMQP 1.0"))
        sender.close()
        print(f"sent 1 message to {addr} (accepted)")
    finally:
        conn.close()


if __name__ == "__main__":
    main()

import javax.jms.Connection;
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 {
    public static void main(String[] args) throws Exception {
        String url = System.getenv().getOrDefault("KUBEMQ_AMQP_URL", "amqp://localhost:5672");
        String address = "queues/amqp10.examples.basic";

        // OPEN: SASL ANONYMOUS by default. The JMS destination name IS the
        // connector 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);
                try (MessageProducer producer = session.createProducer(queue)) {
                    TextMessage msg = session.createTextMessage("hello from AMQP 1.0");
                    producer.send(msg); // blocks until the accepted DISPOSITION
                }
                System.out.printf("sent 1 message to %s (accepted)%n", address);
            }
        }
    }
}

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

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

var addr = "queues/amqp10.examples.basic";

// OPEN: SASL ANONYMOUS by default (no userinfo in the URL).
var connection = await Connection.Factory.CreateAsync(new Address(AmqpUrl()));
try
{
    var session = new Session(connection);

    // ATTACH a sender (server-receiver link) and send one unsettled message.
    var sender = new SenderLink(session, "basic-sender", addr);
    var message = new Message
    {
        BodySection = new Data { Binary = Encoding.UTF8.GetBytes("hello from AMQP 1.0") },
    };
    sender.Send(message, TimeSpan.FromSeconds(15)); // blocks for the accepted DISPOSITION
    await sender.CloseAsync();
    await session.CloseAsync();
    Console.WriteLine($"sent 1 message to {addr} (accepted)");
}
finally
{
    await connection.CloseAsync();
}

import { Connection, type ConnectionOptions } from "rhea-promise";

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,
    // The connector requires a non-empty container-id; rhea sends one by default.
    container_id: `kubemq-amqp10-js-${process.pid}`,
    reconnect: false,
  };
}

async function main(): Promise<void> {
  const address = "queues/amqp10.examples.basic";

  // OPEN: SASL ANONYMOUS by default (no username/password).
  const connection = new Connection(connectionOptions());
  await connection.open();
  try {
    // Attach an AwaitableSender; send() resolves on the accepted disposition.
    const sender = await connection.createAwaitableSender({ target: { address } });
    await sender.send({ body: "hello from AMQP 1.0" }, { timeoutInSeconds: 15 });
    await sender.close();
    console.log(`sent 1 message to ${address} (accepted)`);
  } finally {
    await connection.close();
  }
}

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

use fe2o3_amqp::{Connection, Sender, Session};
use fe2o3_amqp_types::definitions::SenderSettleMode;

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

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let addr = "queues/amqp10.examples.basic";

    // OPEN: SASL ANONYMOUS by default; a non-empty container-id is required.
    let mut connection = Connection::open("kubemq-amqp10-rust", amqp_url().as_str()).await?;
    let mut session = Session::begin(&mut connection).await?;

    // ATTACH a sender pinned to Unsettled (at-least-once); the connector
    // rejects the AMQP default `mixed`.
    let mut sender = Sender::builder()
        .name("basic-sender")
        .target(addr)
        .sender_settle_mode(SenderSettleMode::Unsettled)
        .attach(&mut session)
        .await?;

    let outcome = sender.send("hello from AMQP 1.0").await?;
    if !outcome.is_accepted() {
        return Err(format!("unexpected outcome {outcome:?}").into());
    }
    sender.close().await?;
    println!("sent 1 message to {addr} (accepted)");

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