# Events Store (/connectors/amqp/concepts/events-store)



Events Store is **durable, replayable** pub/sub over the AMQP 1.0 connector. Attach a link to a node whose address begins with `events-store/<channel>` and the connector binds it to the KubeMQ **Events Store** pattern. Unlike plain [Events](/connectors/amqp/concepts/events) (fire-hose, no replay), an Events Store subscriber can **resume** where it left off after a disconnect and can **replay** history from a chosen start position.

## Overview [#overview]

The `events-store/` prefix selects the Events Store pattern (longest-prefix match, evaluated **before** `events/`). A producer attaches a **sender** to `events-store/<ch>`; each `TRANSFER` becomes a KubeMQ `SendEventsStore` with `Store=true`, so the message is persisted. A consumer subscribes **durably** by attaching a **receiver** with terminus `expiry-policy = never`, a **stable container-id**, and a &#x2A;*stable link `Name`** — on reconnect with the same identity, the subscription resumes.

| Operation         | AMQP action                                                                        | KubeMQ mapping                                  |
| ----------------- | ---------------------------------------------------------------------------------- | ----------------------------------------------- |
| Produce           | Attach a **sender** to `events-store/<ch>`, `TRANSFER`                             | `SendEventsStore` (`Store=true`) — persisted    |
| Subscribe durably | **Receiver** with `expiry-policy = never` + stable container-id + stable link name | Durable, resumable subscription                 |
| Set replay start  | Receiver link property `x-opt-kubemq-start`                                        | Start cursor (first / last / sequence / time …) |

### Durable identity [#durable-identity]

A durable subscription is identified by the pair &#x2A;*`(container-id, link name)`**. To resume, reconnect with the **same container-id AND the same link name** — change either and you get a *different* durable subscription that starts fresh. The connector derives a stable id from both:

```text
durableID = sanitize40(containerID) + "_" + sanitize40(linkName) + "_" + fnv1a32hex(containerID + "|" + linkName)
```

<Callout type="warn">
  **A stable container-id is mandatory.** Because the container-id is half the durable identity, a durable subscriber that lets its container-id drift across reconnects (for example, a randomly generated one) will never resume — it creates a brand-new subscription each time. Pin a stable container-id.
</Callout>

## Start positions [#start-positions]

A durable receiver's start position is set with the link property &#x2A;*`x-opt-kubemq-start`**. It applies only to `events-store` (it is ignored on plain Events and on RPC, which have no replay). The grammar:

| `x-opt-kubemq-start` value     | Meaning                                                                 |
| ------------------------------ | ----------------------------------------------------------------------- |
| `""` or `new-only`             | only messages published **after** the subscription starts (the default) |
| `first`                        | replay from the **beginning** of stored history                         |
| `last`                         | start from the **most recent** stored message                           |
| `sequence:<n>`                 | start at sequence number `<n>` (**1-based**, non-negative)              |
| `time:<RFC3339\|unix-seconds>` | start at a wall-clock time                                              |
| `time-delta:<seconds>`         | start `<seconds>` ago from now                                          |

Notes:

* **Time granularity is seconds; the store keeps nanoseconds.** You send `time:` as RFC3339 or whole seconds and the connector converts to the store's nanosecond resolution. `time-delta:` is whole seconds, used verbatim.
* **No "last N by count".** There is no way to ask for "the last N messages" — bound a replay with `sequence:`, `time:`, or `time-delta:`.
* **Malformed values are rejected at attach.** `sequence:abc`, `time:not-a-time`, or an unknown token returns `DETACH(amqp:invalid-field)` naming the offending token.

For **filtered replay** — replaying only the subset of stored messages matching a selector — combine a start position with a selector filter on the receiver; see [Address mapping](/connectors/amqp/reference/address-mapping).

## How it works [#how-it-works]

A durable subscriber attaches with a stable identity and a non-expiring source. On a clean disconnect the connector preserves the cursor; re-attaching with the same `(container-id, link name)` resumes and delivers every event published while the subscriber was away.

<Mermaid
  chart="`
graph LR
PROD[&#x22;Producer<br/>(sender → events-store/orders)&#x22;]
CONN[&#x22;AMQP 1.0 connector<br/>:5672&#x22;]
BROKER[&#x22;Message Broker&#x22;]
STORE[(&#x22;Persistence / Raft&#x22;)]
SUB[&#x22;Durable subscriber<br/>(stable container-id + link name)&#x22;]

PROD -- &#x22;TRANSFER&#x22; --> CONN
CONN -- &#x22;SendEventsStore (Store=true)&#x22; --> BROKER
BROKER --> STORE
STORE -. &#x22;replay from x-opt-kubemq-start&#x22; .-> CONN
CONN -. &#x22;resume on re-attach&#x22; .-> SUB

class PROD,SUB client
class CONN connector
class BROKER broker
class STORE store
`"
/>

*Persisted events are replayed to a durable subscriber from its chosen start position; the cursor survives disconnects so the subscription resumes exactly where it left off.*

<Callout type="warn">
  **Events-store stalled credit loses the buffered window.** A consume link fronts the subscription with a deliver-first ring buffer (≈1024 by default) that is auto-acked in the store *before* you take delivery. If the buffer fills while your credit stays at 0, the link detaches with `amqp:resource-limit-exceeded` ("credit stalled") and the entire buffered, already-acked window is lost — a durable re-attach resumes *after* it. Replenish credit aggressively so the buffer never fills at zero credit. The only signal is the metric `kubemq_amqp10_events_store_dropped_stalled_total`.
</Callout>

## Durable subscribe and resume [#durable-subscribe-and-resume]

Each example publishes 3 events to a live durable subscriber, disconnects, publishes 5 more while the subscriber is away, then re-attaches with the **same** durable identity and receives exactly the 5 missed events — no loss, no re-delivery of the already-consumed first 3. Every client reads the broker endpoint from `KUBEMQ_AMQP_URL` (default `amqp://localhost:5672`).

<Tabs groupId="language" items="['Go','Python','Java','C#','JavaScript','Rust']">
  <Tab value="Go">
    ```go
    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
    }
    ```
  </Tab>

  <Tab value="Python">
    ```python
    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()
    ```
  </Tab>

  <Tab value="Java">
    ```java
    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;
        }
    }
    ```
  </Tab>

  <Tab value="C#">
    ```csharp
    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,
    };
    ```
  </Tab>

  <Tab value="JavaScript">
    ```typescript
    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);
    });
    ```
  </Tab>

  <Tab value="Rust">
    ```rust
    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(())
    }
    ```
  </Tab>
</Tabs>

<Callout type="info">
  **Durable subscriptions are node-local.** A durable identity may have at most **one live attach per node**; a second live attach of the same identity returns `DETACH(amqp:not-allowed, "durable subscription in use")`. In a cluster the durable cursor lives on the node that owned the original attach, so a durable subscriber must reconnect to the **same node** to resume — use load-balancer session affinity or a sticky connection. (RPC replies travel the broker reply path and are cluster-safe; durable subscriptions are not.)
</Callout>

## Related [#related]

<Cards>
  <Card title="Events" href="/connectors/amqp/concepts/events" description="Fire-and-forget pub/sub with no persistence — the non-durable counterpart." />

  <Card title="Queues" href="/connectors/amqp/concepts/queues" description="Competing-consumer work queues with at-least-once delivery and settlement." />

  <Card title="Address mapping" href="/connectors/amqp/reference/address-mapping" description="The full address grammar, the events-store/ prefix, selectors, and start positions." />
</Cards>