Events Store

package main

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

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

const channel = "amqp10.examples.durable"

// The durable identity = (containerID, linkName). Both MUST be stable to resume.
const (
	containerID = "amqp10-examples-durable-container"
	linkName    = "durable-sub"
)
const standingCredit = 100

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(), 90*time.Second)
	defer cancel()
	addr := "events-store/" + channel // events-store/ prefix → KubeMQ Events Store

	// Producer — a plain connection (no stable id needed) that publishes throughout.
	prodConn, _ := amqp.Dial(ctx, amqpURL(), nil)
	defer func() { _ = prodConn.Close() }()
	prodSess, _ := prodConn.NewSession(ctx, nil)
	sender, _ := prodSess.NewSender(ctx, addr, nil)

	// 1. DURABLE SUBSCRIBE (first attach): stable container-id + link name +
	//    non-expiring source + start=new-only.
	durRcv, durConn := attachDurable(ctx, "first attach")
	publish(ctx, sender, 0, 3)
	first := drain(ctx, durRcv, 3, 30*time.Second)
	fmt.Printf("first attach received %d events: %v\n", len(first), first)

	// 2. DISCONNECT — the connector preserves the durable cursor.
	_ = durConn.Close()
	time.Sleep(time.Second)

	// 3. PUBLISH 5 more while away.
	publish(ctx, sender, 3, 8)

	// 4. RE-ATTACH with the SAME identity → resumes and delivers the 5 missed events.
	durRcv2, durConn2 := attachDurable(ctx, "re-attach")
	defer func() { _ = durConn2.Close() }()
	resumed := drain(ctx, durRcv2, 5, 30*time.Second)
	fmt.Printf("re-attach resumed and received the %d events published while away: %v\n", len(resumed), resumed)
}

func attachDurable(ctx context.Context, phase string) (*amqp.Receiver, *amqp.Conn) {
	conn, err := amqp.Dial(ctx, amqpURL(), &amqp.ConnOptions{ContainerID: containerID})
	if err != nil {
		log.Fatalf("[%s] dial durable: %v", phase, err)
	}
	session, _ := conn.NewSession(ctx, nil)
	rcv, err := session.NewReceiver(ctx, "events-store/"+channel, &amqp.ReceiverOptions{
		Credit:             standingCredit,
		SourceExpiryPolicy: amqp.ExpiryPolicyNever,                           // durable signal
		Name:               linkName,                                         // stable link name
		Properties:         map[string]any{"x-opt-kubemq-start": "new-only"}, // start cursor
	})
	if err != nil {
		log.Fatalf("[%s] attach durable receiver: %v", phase, err)
	}
	time.Sleep(750 * time.Millisecond) // let the subscription pump go live
	return rcv, conn
}

func publish(ctx context.Context, sender *amqp.Sender, lo, hi int) {
	for i := lo; i < hi; i++ {
		if err := sender.Send(ctx, amqp.NewMessage([]byte(fmt.Sprintf("es-%03d", i))), nil); err != nil {
			log.Fatalf("publish: %v", err)
		}
	}
}

func drain(ctx context.Context, rcv *amqp.Receiver, max int, window time.Duration) []string {
	out := make([]string, 0, max)
	deadline := time.Now().Add(window)
	for len(out) < max && time.Now().Before(deadline) {
		rcvCtx, cancel := context.WithTimeout(ctx, time.Until(deadline))
		msg, err := rcv.Receive(rcvCtx, nil)
		cancel()
		if err != nil {
			break
		}
		_ = rcv.AcceptMessage(ctx, msg)
		out = append(out, string(msg.GetData()))
	}
	return out
}

import os
import time

from proton import Message, Terminus, symbol
from proton.reactor import Container, ReceiverOption
from proton.utils import BlockingConnection

CHANNEL = "amqp10.examples.durable"
# The durable identity = (container-id, link-name). Both MUST be stable to resume.
CONTAINER_ID = "amqp10-examples-durable-container"
LINK_NAME = "durable-sub"
START_PROP = "x-opt-kubemq-start"
STANDING_CREDIT = 100


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


class DurableSource(ReceiverOption):
    """Make the source durable + non-expiring and stamp x-opt-kubemq-start."""

    def __init__(self, start: str) -> None:
        self.start = start

    def apply(self, receiver) -> None:
        receiver.source.durability = Terminus.DELIVERIES
        receiver.source.expiry_policy = Terminus.EXPIRE_NEVER  # durable signal
        receiver.properties = {symbol(START_PROP): self.start}  # start cursor (link prop)


def attach_durable(phase: str):
    container = Container()
    container.container_id = CONTAINER_ID  # half the durable identity
    conn = BlockingConnection(amqp_url(), container=container)
    receiver = conn.create_receiver(
        "events-store/" + CHANNEL,
        credit=STANDING_CREDIT,
        name=LINK_NAME,  # the other half of the identity
        options=DurableSource("new-only"),
    )
    time.sleep(0.75)  # let the subscription pump go live
    return conn, receiver


def publish(sender, lo: int, hi: int) -> None:
    for i in range(lo, hi):
        sender.send(Message(body=f"es-{i:03d}"))


def drain(receiver, want: int, window: float) -> list[str]:
    out: list[str] = []
    deadline = time.monotonic() + window
    while len(out) < want and time.monotonic() < deadline:
        try:
            msg = receiver.receive(timeout=max(0.0, deadline - time.monotonic()))
        except Exception:
            break
        if receiver.fetcher.unsettled:
            receiver.accept()
        out.append(str(msg.body))
    return out


def main() -> None:
    addr = "events-store/" + CHANNEL  # events-store/ prefix → KubeMQ Events Store
    prod_conn = BlockingConnection(amqp_url())
    sender = prod_conn.create_sender(addr)

    # 1. DURABLE SUBSCRIBE (first attach).
    dur_conn, dur_rcv = attach_durable("first attach")
    publish(sender, 0, 3)
    first = drain(dur_rcv, 3, 30.0)
    print(f"first attach received {len(first)} events: {first}")

    # 2. DISCONNECT — the connector preserves the durable cursor.
    dur_conn.close()
    time.sleep(1.0)

    # 3. PUBLISH 5 more while away.
    publish(sender, 3, 8)

    # 4. RE-ATTACH with the SAME identity → resumes and delivers the 5 missed events.
    dur_conn2, dur_rcv2 = attach_durable("re-attach")
    resumed = drain(dur_rcv2, 5, 30.0)
    print(f"re-attach resumed and received the {len(resumed)} events published while away: {resumed}")

    dur_conn2.close()
    sender.close()
    prod_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.Session;
import javax.jms.Topic;

import org.apache.qpid.jms.JmsConnectionFactory;

public final class Main {
    private static final String CHANNEL = "amqp10.examples.durable";
    // The durable identity = (JMS clientID, subscription name). Both MUST be stable.
    private static final String CLIENT_ID = "amqp10-examples-durable-container";
    private static final String SUB_NAME = "durable-sub";

    public static void main(String[] args) throws Exception {
        String url = System.getenv().getOrDefault("KUBEMQ_AMQP_URL", "amqp://localhost:5672");
        String address = "events-store/" + CHANNEL; // events-store/ → KubeMQ Events Store
        JmsConnectionFactory factory = new JmsConnectionFactory(url);

        // Producer — a separate connection (no stable clientID needed).
        try (Connection prodConn = factory.createConnection();
                Session prodSession = prodConn.createSession(false, Session.AUTO_ACKNOWLEDGE)) {
            prodConn.start();
            Topic topic = prodSession.createTopic(address);
            try (MessageProducer producer = prodSession.createProducer(topic)) {

                // 1. DURABLE SUBSCRIBE (first attach): clientID + durable consumer.
                try (Connection durConn = factory.createConnection()) {
                    durConn.setClientID(CLIENT_ID);
                    durConn.start();
                    try (Session durSession = durConn.createSession(false, Session.CLIENT_ACKNOWLEDGE)) {
                        Topic durTopic = durSession.createTopic(address);
                        MessageConsumer durable = durSession.createDurableConsumer(durTopic, SUB_NAME);
                        Thread.sleep(750);
                        publish(prodSession, producer, 0, 3);
                        Set<String> first = drain(durable, 3, 30_000);
                        System.out.printf("first attach received %d events: %s%n", first.size(), first);
                        durable.close(); // detach but KEEP the durable
                    }
                }
                Thread.sleep(1_000);

                // 3. PUBLISH 5 more while away.
                publish(prodSession, producer, 3, 8);

                // 4. RE-ATTACH with the SAME identity → resumes the subscription.
                try (Connection durConn2 = factory.createConnection()) {
                    durConn2.setClientID(CLIENT_ID);
                    durConn2.start();
                    try (Session durSession2 = durConn2.createSession(false, Session.CLIENT_ACKNOWLEDGE)) {
                        Topic durTopic2 = durSession2.createTopic(address);
                        MessageConsumer durable2 = durSession2.createDurableConsumer(durTopic2, SUB_NAME);
                        Set<String> resumed = drain(durable2, 5, 30_000);
                        System.out.printf(
                                "re-attach resumed and received the %d events published while away: %s%n",
                                resumed.size(), resumed);
                        durable2.close();
                        durSession2.unsubscribe(SUB_NAME); // clean teardown of the durable
                    }
                }
            }
        }
    }

    private static void publish(Session session, MessageProducer producer, int lo, int hi) throws Exception {
        for (int i = lo; i < hi; i++) {
            producer.send(session.createTextMessage(String.format("es-%03d", i)));
        }
    }

    private static Set<String> drain(MessageConsumer consumer, int max, long timeoutMillis) throws Exception {
        Set<String> out = new HashSet<>();
        long deadline = System.currentTimeMillis() + timeoutMillis;
        while (out.size() < max) {
            long remaining = deadline - System.currentTimeMillis();
            if (remaining <= 0) break;
            Message msg = consumer.receive(remaining);
            if (msg == null) break;
            msg.acknowledge();
            out.add(msg.getBody(String.class));
        }
        return out;
    }
}

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

const string channel = "amqp10.examples.durable";
// The durable identity = (containerID, linkName). Both MUST be stable to resume.
const string containerId = "amqp10-examples-durable-container";
const string linkName = "durable-sub";
const int standingCredit = 100;
var expiryNever = new Symbol("never"); // terminus-expiry-policy = never (durable signal)

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

var addr = "events-store/" + channel; // events-store/ prefix → KubeMQ Events Store

// Producer — a plain connection (no stable id needed) that publishes throughout.
var prodConnection = await Connection.Factory.CreateAsync(new Address(AmqpUrl()));
var prodSession = new Session(prodConnection);
var sender = new SenderLink(prodSession, "durable-producer", addr);

void Publish(int lo, int hi)
{
    for (var i = lo; i < hi; i++)
    {
        var message = new Message { BodySection = new Data { Binary = Encoding.UTF8.GetBytes($"es-{i:D3}") } };
        sender.Send(message, TimeSpan.FromSeconds(15));
    }
}

(ReceiverLink Receiver, Connection Connection) AttachDurable(string phase)
{
    var factory = new ConnectionFactory();
    factory.AMQP.ContainerId = containerId; // half the durable identity
    var connection = factory.CreateAsync(new Address(AmqpUrl())).GetAwaiter().GetResult();
    var session = new Session(connection);
    var attach = new Attach
    {
        Source = new Source { Address = addr, ExpiryPolicy = expiryNever },
        Target = new Target(),
        LinkName = linkName, // the other half of the durable identity
        Properties = new Fields { { new Symbol("x-opt-kubemq-start"), "new-only" } },
    };
    var receiver = new ReceiverLink(session, linkName, attach, null);
    receiver.SetCredit(standingCredit, autoRestore: true);
    Thread.Sleep(750); // let the subscription pump go live
    return (receiver, connection);
}

List<string> Drain(ReceiverLink receiver, int max, TimeSpan window)
{
    var outp = new List<string>(max);
    var deadline = DateTime.UtcNow + window;
    while (outp.Count < max && DateTime.UtcNow < deadline)
    {
        var message = receiver.Receive(TimeSpan.FromSeconds(2));
        if (message is null) continue;
        receiver.Accept(message);
        outp.Add(BodyString(message));
    }
    return outp;
}

try
{
    // 1. DURABLE SUBSCRIBE (first attach).
    var (durRcv, durConn) = AttachDurable("first attach");
    Publish(0, 3);
    var first = Drain(durRcv, 3, TimeSpan.FromSeconds(30));
    Console.WriteLine($"first attach received {first.Count} events: [{string.Join(" ", first)}]");

    // 2. DISCONNECT — the connector preserves the durable cursor.
    await durConn.CloseAsync();
    await Task.Delay(1000);

    // 3. PUBLISH 5 more while away.
    Publish(3, 8);

    // 4. RE-ATTACH with the SAME identity → resumes the subscription.
    var (durRcv2, durConn2) = AttachDurable("re-attach");
    try
    {
        var resumed = Drain(durRcv2, 5, TimeSpan.FromSeconds(30));
        Console.WriteLine($"re-attach resumed and received the {resumed.Count} events published while away: [{string.Join(" ", resumed)}]");
        await durRcv2.CloseAsync();
    }
    finally
    {
        await durConn2.CloseAsync();
    }
}
finally
{
    await sender.CloseAsync();
    await prodSession.CloseAsync();
    await prodConnection.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.durable";
// The durable identity = (containerID, linkName). Both MUST be stable to resume.
const containerId = "amqp10-examples-durable-container";
const linkName = "durable-sub";
const standingCredit = 100;

function brokerEndpoint(): { host: string; port: number } {
  const url = new URL(process.env["KUBEMQ_AMQP_URL"] ?? "amqp://localhost:5672");
  return { host: url.hostname, port: url.port ? Number(url.port) : 5672 };
}

function durableOptions(): ConnectionOptions {
  const { host, port } = brokerEndpoint();
  return { host, port, container_id: containerId, reconnect: false }; // STABLE id
}

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

function sleep(ms: number): Promise<void> {
  return new Promise((resolve) => setTimeout(resolve, ms));
}

interface DurableAttach {
  connection: Connection;
  receiver: Receiver;
}

async function attachDurable(): Promise<DurableAttach> {
  const connection = new Connection(durableOptions());
  await connection.open();
  const receiver = await connection.createReceiver({
    name: linkName, // half the durable identity
    source: {
      address: `events-store/${channel}`,
      expiry_policy: "never", // durable signal
    },
    properties: { "x-opt-kubemq-start": "new-only" }, // start cursor
    credit_window: 0,
    autoaccept: false,
    autosettle: false,
  });
  await sleep(750); // let the subscription pump go live
  return { connection, receiver };
}

function drain(receiver: Receiver, max: number, windowMs: number): Promise<string[]> {
  return new Promise((resolve, reject) => {
    const out: string[] = [];
    const timer = setTimeout(() => {
      receiver.removeListener(ReceiverEvents.message, handler);
      resolve(out);
    }, windowMs);
    const handler = (ctx: EventContext): void => {
      try {
        ctx.delivery?.accept(); // accept advances the durable cursor
        out.push(bodyToString(ctx.message?.body));
        if (out.length >= max) {
          clearTimeout(timer);
          receiver.removeListener(ReceiverEvents.message, handler);
          resolve(out);
          return;
        }
        receiver.addCredit(1);
      } catch (err) {
        clearTimeout(timer);
        receiver.removeListener(ReceiverEvents.message, handler);
        reject(err instanceof Error ? err : new Error(String(err)));
      }
    };
    receiver.on(ReceiverEvents.message, handler);
    receiver.addCredit(standingCredit);
  });
}

async function main(): Promise<void> {
  const { host, port } = brokerEndpoint();
  const address = `events-store/${channel}`; // events-store/ → KubeMQ Events Store

  // Producer — a plain connection (no stable id needed) that publishes throughout.
  const prodConnection = new Connection({
    host, port, container_id: `kubemq-amqp10-js-durable-prod-${process.pid}`, reconnect: false,
  });
  await prodConnection.open();
  const sender = await prodConnection.createAwaitableSender({ target: { address } });
  const publish = async (lo: number, hi: number): Promise<void> => {
    for (let i = lo; i < hi; i++) {
      await sender.send({ body: `es-${String(i).padStart(3, "0")}` }, { timeoutInSeconds: 15 });
    }
  };

  try {
    // 1. DURABLE SUBSCRIBE (first attach).
    const first = await attachDurable();
    await publish(0, 3);
    const firstBodies = await drain(first.receiver, 3, 30_000);
    console.log(`first attach received ${firstBodies.length} events: [${firstBodies.join(" ")}]`);

    // 2. DISCONNECT — the connector preserves the durable cursor.
    await first.connection.close();
    await sleep(1_000);

    // 3. PUBLISH 5 more while away.
    await publish(3, 8);

    // 4. RE-ATTACH with the SAME identity → resumes the subscription.
    const second = await attachDurable();
    try {
      const resumed = await drain(second.receiver, 5, 30_000);
      console.log(`re-attach resumed and received the ${resumed.length} events published while away: [${resumed.join(" ")}]`);
    } finally {
      await second.connection.close();
    }
  } finally {
    await sender.close();
    await prodConnection.close();
  }
}

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

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

use fe2o3_amqp::connection::ConnectionHandle;
use fe2o3_amqp::link::receiver::CreditMode;
use fe2o3_amqp::session::SessionHandle;
use fe2o3_amqp::{Connection, Receiver, Sender, Session};
use fe2o3_amqp_types::definitions::SenderSettleMode;
use fe2o3_amqp_types::messaging::{Body, Source, TerminusExpiryPolicy};
use fe2o3_amqp_types::primitives::{OrderedMap, Symbol, Value};

const CHANNEL: &str = "amqp10.examples.durable";
// The durable identity = (container-id, link name). Both MUST be stable to resume.
const CONTAINER_ID: &str = "amqp10-examples-durable-container";
const LINK_NAME: &str = "durable-sub";
const STANDING_CREDIT: u32 = 100;
const START_PROP: &str = "x-opt-kubemq-start";

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

fn body_string(body: &Body<Value>) -> String {
    let bytes = match 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(),
    };
    String::from_utf8_lossy(&bytes).into_owned()
}

fn durable_source(addr: &str) -> Source {
    Source::builder()
        .address(addr)
        .expiry_policy(TerminusExpiryPolicy::Never) // durable signal
        .build()
}

fn start_props(start: &str) -> OrderedMap<Symbol, Value> {
    let mut props = OrderedMap::new();
    props.insert(Symbol::from(START_PROP), Value::String(start.to_string()));
    props
}

#[allow(clippy::type_complexity)]
async fn attach_durable(
    url: &str,
    addr: &str,
) -> Result<(Receiver, SessionHandle<()>, ConnectionHandle<()>), Box<dyn std::error::Error>> {
    let mut connection = Connection::builder()
        .container_id(CONTAINER_ID) // half the durable identity
        .open(url)
        .await?;
    let mut session = Session::begin(&mut connection).await?;
    let receiver = Receiver::builder()
        .name(LINK_NAME) // the other half of the identity
        .source(durable_source(addr))
        .properties(start_props("new-only")) // start cursor
        .credit_mode(CreditMode::Auto(STANDING_CREDIT))
        .attach(&mut session)
        .await?;
    tokio::time::sleep(Duration::from_millis(750)).await; // let the pump go live
    Ok((receiver, session, connection))
}

async fn publish(sender: &mut Sender, lo: usize, hi: usize) -> Result<(), Box<dyn std::error::Error>> {
    for i in lo..hi {
        let outcome = sender.send(format!("es-{i:03}")).await?;
        if !outcome.is_accepted() {
            return Err(format!("publish es-{i:03}: unexpected outcome {outcome:?}").into());
        }
    }
    Ok(())
}

async fn drain(receiver: &mut Receiver, max: usize, window: Duration) -> Result<Vec<String>, Box<dyn std::error::Error>> {
    let mut out = Vec::with_capacity(max);
    let deadline = tokio::time::Instant::now() + window;
    while out.len() < max {
        let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
        if remaining.is_zero() {
            break;
        }
        match tokio::time::timeout(remaining, receiver.recv::<Body<Value>>()).await {
            Ok(Ok(delivery)) => {
                receiver.accept(&delivery).await?;
                out.push(body_string(&delivery.message().body));
            }
            _ => break,
        }
    }
    Ok(out)
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let url = amqp_url();
    let addr = format!("events-store/{CHANNEL}"); // events-store/ → KubeMQ Events Store

    // Producer — a plain connection (no stable id needed) that publishes throughout.
    let mut prod_conn = Connection::open("amqp10-examples-durable-producer", url.as_str()).await?;
    let mut prod_sess = Session::begin(&mut prod_conn).await?;
    let mut sender = Sender::builder()
        .name("durable-replay-producer")
        .target(addr.as_str())
        .sender_settle_mode(SenderSettleMode::Unsettled)
        .attach(&mut prod_sess)
        .await?;

    // 1. DURABLE SUBSCRIBE (first attach).
    let (mut dur_rcv, mut dur_sess, mut dur_conn) = attach_durable(url.as_str(), addr.as_str()).await?;
    publish(&mut sender, 0, 3).await?;
    let first = drain(&mut dur_rcv, 3, Duration::from_secs(30)).await?;
    println!("first attach received {} events: {first:?}", first.len());

    // 2. DISCONNECT — the connector preserves the durable cursor.
    dur_rcv.close().await?;
    dur_sess.end().await?;
    dur_conn.close().await?;
    tokio::time::sleep(Duration::from_secs(1)).await;

    // 3. PUBLISH 5 more while away.
    publish(&mut sender, 3, 8).await?;

    // 4. RE-ATTACH with the SAME identity → resumes the subscription.
    let (mut dur_rcv2, mut dur_sess2, mut dur_conn2) = attach_durable(url.as_str(), addr.as_str()).await?;
    let resumed = drain(&mut dur_rcv2, 5, Duration::from_secs(30)).await?;
    let resumed_set: HashSet<String> = resumed.iter().cloned().collect();
    println!("re-attach resumed and received the {} events published while away: {resumed:?}", resumed_set.len());

    dur_rcv2.close().await?;
    dur_sess2.end().await?;
    dur_conn2.close().await?;
    sender.close().await?;
    prod_sess.end().await?;
    prod_conn.close().await?;
    Ok(())
}