# SSE Behavior (/connectors/cloudevents/how-to/sse-behavior)



The CloudEvents connector streams every subscription over **Server-Sent Events (SSE)** — a standard, long-lived HTTP mechanism for server-to-client delivery. This guide covers the wire format your client must parse, the keepalive and idle-timeout behavior, the connection limit, and how `Last-Event-ID` resumes an events-store stream after a disconnect.

## Overview [#overview]

When a client opens any `GET /ce/subscribe/*` endpoint, the connector holds the connection open and responds with `Content-Type: text/event-stream`. Messages arriving on the subscribed channel are pushed to the client as SSE frames as they happen — there is no polling.

All four subscription endpoints (`events`, `events-store`, `commands`, `queries`) share the same SSE wire format and lifecycle. The only behavioral difference is **replay**: events-store streams carry an `id:` per frame and support `Last-Event-ID` reconnection, while plain `events` streams do not (non-persistent events cannot be replayed).

<Callout type="info">
  SSE subscription endpoints are long-lived `GET` requests and are **not** subject to the `TimeoutSeconds` (default 60s) that applies to synchronous `POST` endpoints. They stay open until idle timeout, the connection limit, a client disconnect, or a server shutdown.
</Callout>

## SSE wire format [#sse-wire-format]

Each frame is a set of optional `id:`, `event:`, and `data:` lines terminated by a blank line. The `event:` field tells the client which kind of payload the `data:` line carries.

```text title="SSE frame — CloudEvent message"
id: 42
event: cloudevent
data: {"specversion":"1.0","type":"com.example.order","source":"svc","id":"abc","subject":"orders","data":{"amount":99}}

```

```text title="SSE frame — non-CloudEvent message"
event: message
data: {"channel":"orders","metadata":"","tags":{"key":"val"},"data":{"amount":99}}

```

```text title="SSE frame — error event"
event: error
data: {"is_error":true,"message":"stream idle timeout"}

```

```text title="Keepalive comment (not a named event)"
: keepalive

```

### Event types [#event-types]

The `event:` field distinguishes message kinds on a single stream:

| `event:` value | Description                                                                                                        |
| -------------- | ------------------------------------------------------------------------------------------------------------------ |
| `cloudevent`   | The message carries a `ce_specversion` tag; `data:` is a reconstructed CloudEvent JSON object.                     |
| `message`      | A non-CloudEvent message; `data:` is plain JSON with native KubeMQ fields (`channel`, `metadata`, `tags`, `data`). |
| `error`        | A terminal error (for example, idle timeout); `data:` is `{"is_error":true,"message":"..."}`.                      |
| *(absent)*     | A keepalive comment line (`: keepalive`) — ignored by standard `EventSource` clients.                              |

## How reconnection works [#how-reconnection-works]

The diagram below shows an events-store subscription: the client reads several frames, records the last `id:` it saw, disconnects, then reconnects with `Last-Event-ID` to resume from the next sequence — no duplicates, no gaps.

<Mermaid
  chart="`
sequenceDiagram
participant Client
participant CE as CloudEvents connector
participant Store as Events Store
Client->>CE: GET /ce/subscribe/events-store?events_store_type=2
CE->>Store: read from first stored message
Store-->>CE: messages
CE-->>Client: id: 1 / event: cloudevent
CE-->>Client: id: 2 / event: cloudevent
CE-->>Client: id: 3 / event: cloudevent
Note over Client: connection drops<br/>(Last-Event-ID = 3 recorded)
Client->>CE: GET /ce/subscribe/events-store (Last-Event-ID: 3)
CE->>Store: resume from sequence 3 + 1
Store-->>CE: messages
CE-->>Client: id: 4 / event: cloudevent
CE-->>Client: id: 5 / event: cloudevent
`"
/>

*An events-store stream resumes from `sequence + 1` when the client reconnects with `Last-Event-ID`.*

## Keepalive [#keepalive]

To stop proxies and load balancers from closing an idle connection, the connector sends a keepalive comment every **30 seconds**:

```text
: keepalive

```

Keepalive frames are SSE comments (lines starting with `:`). Standard `EventSource` clients ignore them automatically; a manual parser should skip any line that starts with `:`.

## Idle timeout [#idle-timeout]

If no message arrives for `MaxSSEIdleSeconds` (default **300 seconds**), the connector emits an `error` event and closes the connection:

```text
event: error
data: {"is_error":true,"message":"stream idle timeout"}

```

The idle timer resets on every received message — keepalive comments do **not** reset it. Treat the idle-timeout error as a normal lifecycle event and reconnect if you still need the stream. See [Configuration](/connectors/cloudevents/concepts/configuration-model) to tune `MaxSSEIdleSeconds`.

## Connection limits [#connection-limits]

When `MaxSSEConnections` is greater than `0`, the connector caps the number of concurrent SSE connections across **all** subscription endpoints. A new connection that exceeds the limit is rejected with **HTTP 429 Too Many Requests**. The default of `0` disables the limit (unlimited connections).

## Last-Event-ID reconnection [#last-event-id-reconnection]

Replay-on-reconnect is available for **events-store subscriptions only**. Each frame on an events-store stream carries an `id:` line set to the message sequence number. When a client reconnects with the `Last-Event-ID` header, the connector resumes from `sequence + 1`, automatically replaying anything missed during the disconnect.

This follows the standard SSE reconnection protocol — browsers and `EventSource` clients send `Last-Event-ID` on reconnect for you.

<Callout type="warn">
  **Omit `events_store_type` when reconnecting with `Last-Event-ID`.** If both are present, the `events_store_type` query parameter takes precedence and the `Last-Event-ID` resume is ignored. Drop `events_store_type` from the reconnect URL.
</Callout>

Plain `events` subscriptions do not emit `id:` fields and cannot be replayed. For command/query subscriptions, CloudEvent messages carry an `id:` set to the CloudEvent `id` attribute, while non-CE command/query messages have no `id:`.

## Mixed CE and non-CE messages [#mixed-ce-and-non-ce-messages]

A single channel can carry both CloudEvents and native KubeMQ messages, and the connector delivers both on the same SSE stream. The `event:` field is how you tell them apart:

* `event: cloudevent` — `data:` is a reconstructed CloudEvent JSON object.
* `event: message` — `data:` is plain JSON with native KubeMQ fields.

Always branch on the `event:` value so your consumer handles mixed-protocol channels correctly.

## Usage [#usage]

The example below demonstrates the full reconnection lifecycle against `events-store`: subscribe with `events_store_type=2` (start from first), read a batch of frames while recording the last `id:`, disconnect, then reconnect with the `Last-Event-ID` header to resume. The curl tab shows the two raw requests; each language tab is the verbatim KubeMQ example that parses the SSE stream and performs the resume.

<Tabs groupId="language" items="['curl','C#','Go','Java','JavaScript','Python','Ruby','Rust']">
  <Tab value="curl">
    ```bash
    # 1. Subscribe from the first stored message; note the id: on each frame.
    curl -N \
      -H "Accept: text/event-stream" \
      "http://localhost:9090/ce/subscribe/events-store?client_id=my-client&channel=audit-log&events_store_type=2"

    # Frames look like:
    #   id: 1
    #   event: cloudevent
    #   data: {"specversion":"1.0",...}
    #
    #   : keepalive

    # 2. Reconnect after a disconnect — resume from sequence + 1.
    #    Omit events_store_type so Last-Event-ID takes precedence.
    curl -N \
      -H "Accept: text/event-stream" \
      -H "Last-Event-ID: 3" \
      "http://localhost:9090/ce/subscribe/events-store?client_id=my-client&channel=audit-log"
    ```
  </Tab>

  <Tab value="C#">
    ```csharp
    // events-store/ReconnectResume — Last-Event-ID reconnect.
    using CloudNative.CloudEvents;
    using CloudNative.CloudEvents.SystemTextJson;
    using System.Net.Http.Headers;
    using System.Text;
    using System.Text.Json;

    static string ServerUrl() => Environment.GetEnvironmentVariable("KUBEMQ_CE_URL") ?? "http://localhost:9090";

    var base_ = ServerUrl();
    var channel = "csharp-ce-events-store.reconnect-resume";
    var formatter = new JsonEventFormatter();
    using var httpClient = new HttpClient();

    // Publish 4 events.
    Console.WriteLine("Publishing 4 events...");
    for (int i = 1; i <= 4; i++)
    {
        var ev = new CloudEvent {
            Id = Guid.NewGuid().ToString(),
            Type = "com.kubemq.examples.eventsstore.stored",
            Source = new Uri("urn:kubemq-ce-csharp-example"),
            Subject = channel,
            DataContentType = "application/json",
            Data = new { seq = i },
        };
        var bytes = formatter.EncodeStructuredModeMessage(ev, out var ct);
        using var c = new ByteArrayContent(bytes.ToArray());
        c.Headers.ContentType = MediaTypeHeaderValue.Parse(ct.ToString());
        await httpClient.PostAsync($"{base_}/ce/send/event-store", c);
    }

    // Helper: subscribe and collect up to maxEvents, return (events, lastId).
    async Task<(List<JsonElement> events, string lastId)> Subscribe(
        string clientId, string? lastEventId, int maxEvents)
    {
        var events = new List<JsonElement>();
        var lastId = "";
        using var http = new HttpClient { Timeout = Timeout.InfiniteTimeSpan };
        // Omit events_store_type when reconnecting with Last-Event-ID.
        var url = lastEventId == null
            ? $"{base_}/ce/subscribe/events-store?client_id={clientId}&channel={Uri.EscapeDataString(channel)}&events_store_type=2"
            : $"{base_}/ce/subscribe/events-store?client_id={clientId}&channel={Uri.EscapeDataString(channel)}";
        using var req = new HttpRequestMessage(HttpMethod.Get, url);
        req.Headers.Accept.Add(new MediaTypeWithQualityHeaderValue("text/event-stream"));
        if (lastEventId != null) req.Headers.Add("Last-Event-ID", lastEventId);
        using var resp = await http.SendAsync(req, HttpCompletionOption.ResponseHeadersRead);
        using var reader = new StreamReader(await resp.Content.ReadAsStreamAsync(), Encoding.UTF8);
        string? evType = null, data = null, id = null, line;
        using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10));
        while ((line = await reader.ReadLineAsync().WaitAsync(cts.Token)) != null)
        {
            if (line == "") {
                if (evType == "cloudevent" && data != null) {
                    if (id != null) lastId = id;
                    events.Add(JsonSerializer.Deserialize<JsonElement>(data));
                    if (events.Count >= maxEvents) break;
                }
                evType = null; data = null; id = null;
            }
            else if (line.StartsWith(":")) { }
            else if (line.StartsWith("id:")) id = line[3..].Trim();
            else if (line.StartsWith("event:")) evType = line[6..].Trim();
            else if (line.StartsWith("data:")) data = line[5..].Trim();
        }
        return (events, lastId);
    }

    // First connection: receive 2 events.
    Console.WriteLine("First connection — receiving 2 events:");
    var (first, lastId) = await Subscribe("csharp-es-reconnect-1", null, 2);
    foreach (var ce in first)
        Console.WriteLine($"  Received: seq={ce.GetProperty("data").GetProperty("seq")}");
    Console.WriteLine($"  Last-Event-ID recorded: {lastId}");

    // Reconnect using Last-Event-ID (new client_id; broker requires a unique active client_id).
    await Task.Delay(500);
    Console.WriteLine($"Reconnecting with Last-Event-ID={lastId}...");
    var (second, _) = await Subscribe("csharp-es-reconnect-2", lastId, 2);
    foreach (var ce in second)
        Console.WriteLine($"  Resumed: seq={ce.GetProperty("data").GetProperty("seq")}");
    Console.WriteLine("Reconnect-resume demonstration complete.");
    ```
  </Tab>

  <Tab value="Go">
    ```go
    // events-store/reconnect-resume — SSE reconnection with Last-Event-ID.
    package main

    import (
        "bufio"
        "encoding/json"
        "fmt"
        "io"
        "log"
        "net/http"
        "os"
        "strings"
        "time"

        cloudevents "github.com/cloudevents/sdk-go/v2"
    )

    func serverURL() string {
        if u := os.Getenv("KUBEMQ_CE_URL"); u != "" {
            return u
        }
        return "http://localhost:9090"
    }

    func publishAll(base, channel string, count int) {
        for i := 1; i <= count; i++ {
            event := cloudevents.NewEvent()
            event.SetType("com.kubemq.examples.eventsstore.reconnect")
            event.SetSource("kubemq-ce-go-example")
            event.SetSubject(channel)
            _ = event.SetData(cloudevents.ApplicationJSON, map[string]int{"n": i})

            body, _ := json.Marshal(event)
            req, _ := http.NewRequest("POST", base+"/ce/send/event-store",
                strings.NewReader(string(body)))
            req.Header.Set("Content-Type", "application/cloudevents+json")
            resp, err := http.DefaultClient.Do(req)
            if err != nil {
                log.Fatal("send:", err)
            }
            resp.Body.Close()
        }
        fmt.Printf("Published %d events to events-store.\n", count)
    }

    // readN reads exactly n cloudevents from an SSE stream, returning the last SSE id.
    func readN(body io.ReadCloser, n int) (lastID string) {
        scanner := bufio.NewScanner(body)
        var evType, data, sseID string
        received := 0
        for scanner.Scan() {
            line := scanner.Text()
            if line == "" {
                if evType == "cloudevent" && data != "" {
                    received++
                    lastID = sseID
                    var ce map[string]interface{}
                    _ = json.Unmarshal([]byte(data), &ce)
                    fmt.Printf("  [%d] id=%s data=%v\n", received, sseID, ce["data"])
                    if received == n {
                        return lastID
                    }
                }
                evType, data, sseID = "", "", ""
                continue
            }
            if strings.HasPrefix(line, ":") {
                continue
            }
            if strings.HasPrefix(line, "id:") {
                sseID = strings.TrimSpace(strings.TrimPrefix(line, "id:"))
            } else if strings.HasPrefix(line, "event:") {
                evType = strings.TrimSpace(strings.TrimPrefix(line, "event:"))
            } else if strings.HasPrefix(line, "data:") {
                data = strings.TrimSpace(strings.TrimPrefix(line, "data:"))
            }
        }
        return lastID
    }

    func openSSE(base, channel, clientID, lastEventID string) (*http.Response, error) {
        // When reconnecting, omit events_store_type so Last-Event-ID takes precedence.
        var sseURL string
        if lastEventID == "" {
            sseURL = fmt.Sprintf(
                "%s/ce/subscribe/events-store?client_id=%s&channel=%s&events_store_type=2",
                base, clientID, channel)
        } else {
            sseURL = fmt.Sprintf(
                "%s/ce/subscribe/events-store?client_id=%s&channel=%s",
                base, clientID, channel)
        }

        req, _ := http.NewRequest("GET", sseURL, nil)
        req.Header.Set("Accept", "text/event-stream")
        req.Header.Set("Cache-Control", "no-cache")
        if lastEventID != "" {
            req.Header.Set("Last-Event-ID", lastEventID)
        }

        client := &http.Client{Timeout: 0}
        return client.Do(req)
    }

    func main() {
        base := serverURL()
        channel := "go-ce-events-store.reconnect-resume"
        const totalEvents = 6
        const firstBatch = 3

        publishAll(base, channel, totalEvents)
        time.Sleep(200 * time.Millisecond)

        // First connection: receive first batch.
        fmt.Printf("\nFirst connection (StartFromFirst, reading first %d events):\n", firstBatch)
        resp1, err := openSSE(base, channel, "go-reconnect-sub", "")
        if err != nil {
            log.Fatal("first connect:", err)
        }
        lastID := readN(resp1.Body, firstBatch)
        resp1.Body.Close()
        fmt.Printf("Disconnected. Last-Event-ID captured: %s\n", lastID)

        // Second connection: resume from lastID.
        fmt.Printf("\nReconnecting with Last-Event-ID: %s\n", lastID)
        resp2, err := openSSE(base, channel, "go-reconnect-sub", lastID)
        if err != nil {
            log.Fatal("reconnect:", err)
        }
        defer resp2.Body.Close()

        remaining := totalEvents - firstBatch
        fmt.Printf("Receiving remaining %d events:\n", remaining)
        readN(resp2.Body, remaining)

        fmt.Println("\nReconnect-resume demonstration complete.")
    }
    ```
  </Tab>

  <Tab value="Java">
    ```java
    // events-store/reconnect-resume — Last-Event-ID reconnect.
    package io.kubemq.examples.eventsstore.reconnectresume;

    import com.fasterxml.jackson.databind.ObjectMapper;
    import io.cloudevents.CloudEvent;
    import io.cloudevents.core.builder.CloudEventBuilder;
    import io.cloudevents.core.format.EventFormat;
    import io.cloudevents.core.provider.EventFormatProvider;
    import io.cloudevents.jackson.JsonFormat;

    import java.io.BufferedReader;
    import java.io.InputStreamReader;
    import java.net.HttpURLConnection;
    import java.net.URI;
    import java.net.URL;
    import java.net.http.HttpClient;
    import java.net.http.HttpRequest;
    import java.net.http.HttpResponse;
    import java.nio.charset.StandardCharsets;
    import java.time.OffsetDateTime;
    import java.util.Map;
    import java.util.UUID;
    import java.util.concurrent.ArrayBlockingQueue;
    import java.util.concurrent.BlockingQueue;
    import java.util.concurrent.TimeUnit;
    import java.util.concurrent.atomic.AtomicReference;

    public class Main {
        static String serverUrl() {
            String u = System.getenv("KUBEMQ_CE_URL");
            return (u != null && !u.isEmpty()) ? u : "http://localhost:9090";
        }
        static final ObjectMapper MAPPER = new ObjectMapper();

        /** Opens one SSE connection, reads up to maxEvents, returns lastEventID seen. */
        static String subscribe(String base, String channel, String clientId,
                                String lastEventId, int maxEvents,
                                BlockingQueue<String> out) throws Exception {
            // Build URL — omit events_store_type when reconnecting with Last-Event-ID.
            String sseUrl = base + "/ce/subscribe/events-store?client_id=" + clientId + "&channel=" + channel;
            if (lastEventId == null) {
                sseUrl += "&events_store_type=2";
            }
            final String finalLastEventId = lastEventId;
            final String finalUrl = sseUrl;
            AtomicReference<String> lastId = new AtomicReference<>("");
            BlockingQueue<Boolean> done = new ArrayBlockingQueue<>(1);

            Thread.ofVirtual().start(() -> {
                try {
                    HttpURLConnection conn = (HttpURLConnection) new URL(finalUrl).openConnection();
                    conn.setRequestProperty("Accept", "text/event-stream");
                    conn.setReadTimeout(10_000);
                    if (finalLastEventId != null) {
                        conn.setRequestProperty("Last-Event-ID", finalLastEventId);
                    }
                    int[] count = {0};
                    try (BufferedReader reader = new BufferedReader(
                            new InputStreamReader(conn.getInputStream(), StandardCharsets.UTF_8))) {
                        String line; String evType = null, data = null, id = null;
                        while ((line = reader.readLine()) != null && count[0] < maxEvents) {
                            if (line.isEmpty()) {
                                if ("cloudevent".equals(evType) && data != null) {
                                    if (id != null) lastId.set(id);
                                    out.offer(data);
                                    count[0]++;
                                    if (count[0] >= maxEvents) break;
                                }
                                evType = null; data = null; id = null;
                            } else if (line.startsWith("id:")) id = line.substring(3).trim();
                            else if (line.startsWith("event:")) evType = line.substring(6).trim();
                            else if (line.startsWith("data:")) data = line.substring(5).trim();
                        }
                    }
                } catch (Exception e) { /* read ended */ }
                done.offer(true);
            });
            done.poll(12, TimeUnit.SECONDS);
            return lastId.get();
        }

        public static void main(String[] args) throws Exception {
            String base = serverUrl();
            String channel = "java-ce-events-store.reconnect-resume";

            EventFormatProvider.getInstance().registerFormat(new JsonFormat());
            EventFormat format = EventFormatProvider.getInstance().resolveFormat(JsonFormat.CONTENT_TYPE);
            HttpClient httpClient = HttpClient.newHttpClient();

            // Publish 4 events.
            System.out.println("Publishing 4 events...");
            for (int i = 1; i <= 4; i++) {
                CloudEvent ev = CloudEventBuilder.v1()
                        .withId(UUID.randomUUID().toString())
                        .withType("com.kubemq.examples.eventsstore.stored")
                        .withSource(URI.create("kubemq-ce-java-example"))
                        .withSubject(channel)
                        .withDataContentType("application/json")
                        .withTime(OffsetDateTime.now())
                        .withData("application/json", MAPPER.writeValueAsBytes(Map.of("seq", i)))
                        .build();
                httpClient.send(
                        HttpRequest.newBuilder().uri(URI.create(base + "/ce/send/event-store"))
                                .POST(HttpRequest.BodyPublishers.ofByteArray(format.serialize(ev)))
                                .header("Content-Type", "application/cloudevents+json").build(),
                        HttpResponse.BodyHandlers.ofString());
            }

            // First connection: receive 2 events, record lastEventID.
            BlockingQueue<String> received = new ArrayBlockingQueue<>(10);
            System.out.println("First connection — receiving 2 events:");
            String lastId = subscribe(base, channel, "java-es-reconnect", null, 2, received);
            for (int i = 0; i < 2; i++) {
                String data = received.poll(5, TimeUnit.SECONDS);
                if (data != null) {
                    Map<?, ?> ce = MAPPER.readValue(data, Map.class);
                    System.out.println("  Received: seq=" + ((Map<?, ?>) ce.get("data")).get("seq"));
                }
            }
            System.out.println("  Last-Event-ID recorded: " + lastId);

            // Reconnect using Last-Event-ID — resume from next sequence.
            System.out.println("Reconnecting with Last-Event-ID=" + lastId + "...");
            subscribe(base, channel, "java-es-reconnect", lastId, 2, received);
            for (int i = 0; i < 2; i++) {
                String data = received.poll(5, TimeUnit.SECONDS);
                if (data != null) {
                    Map<?, ?> ce = MAPPER.readValue(data, Map.class);
                    System.out.println("  Resumed: seq=" + ((Map<?, ?>) ce.get("data")).get("seq"));
                }
            }
            System.out.println("Reconnect-resume demonstration complete.");
        }
    }
    ```
  </Tab>

  <Tab value="JavaScript">
    ```typescript
    // events-store/reconnect-resume — Last-Event-ID reconnection.
    import { CloudEvent, HTTP } from 'cloudevents';
    import EventSource from 'eventsource';

    function serverUrl(): string {
      return process.env.KUBEMQ_CE_URL ?? 'http://localhost:9090';
    }

    function readNEvents(
      base: string, channel: string, clientId: string,
      n: number, lastEventId?: string,
    ): Promise<string> {
      return new Promise((resolve) => {
        const params = new URLSearchParams({ client_id: clientId, channel });
        if (lastEventId) {
          // omit events_store_type so Last-Event-ID takes precedence
        } else {
          params.set('events_store_type', '2');
        }
        const headers: Record<string, string> = {};
        if (lastEventId) headers['Last-Event-ID'] = lastEventId;

        // EventSource works for the initial connection; for the Last-Event-ID
        // reconnect we use fetch so we can set the header explicitly.
        if (!lastEventId) {
          const es = new EventSource(`${base}/ce/subscribe/events-store?${params}`);
          let count = 0;
          let lastId = '';
          es.addEventListener('error', (err) => {
            console.error('SSE error:', err);
            es.close();
            resolve('');
          });
          es.addEventListener('cloudevent', (evt: MessageEvent & { lastEventId: string }) => {
            const ce = JSON.parse(evt.data) as Record<string, unknown>;
            count++;
            lastId = evt.lastEventId;
            console.log(`  [${count}] id=${lastId} data=${JSON.stringify(ce.data)}`);
            if (count === n) {
              es.close();
              resolve(lastId);
            }
          });
        } else {
          const url = `${base}/ce/subscribe/events-store?${params}`;
          fetch(url, { headers }).then(async (resp) => {
            const reader = resp.body!.getReader();
            const decoder = new TextDecoder();
            let buffer = '';
            let count = 0;
            let lastId = lastEventId;
            let evType = '';
            let data = '';

            while (count < n) {
              const { value, done } = await reader.read();
              if (done) break;
              buffer += decoder.decode(value, { stream: true });
              const lines = buffer.split('\n');
              buffer = lines.pop() ?? '';
              for (const line of lines) {
                if (line === '') {
                  if (evType === 'cloudevent' && data) {
                    const ce = JSON.parse(data) as Record<string, unknown>;
                    count++;
                    console.log(`  [${count}] id=${lastId} data=${JSON.stringify(ce.data)}`);
                    if (count === n) { reader.cancel(); resolve(lastId); return; }
                  }
                  evType = data = '';
                } else if (line.startsWith('id:')) {
                  lastId = line.slice(3).trim();
                } else if (line.startsWith('event:')) {
                  evType = line.slice(6).trim();
                } else if (line.startsWith('data:')) {
                  data = line.slice(5).trim();
                }
              }
            }
            resolve(lastId);
          });
        }
      });
    }

    async function main(): Promise<void> {
      const base = serverUrl();
      const channel = 'js-ce-events-store.reconnect-resume';
      const total = 6;
      const firstBatch = 3;

      for (let i = 1; i <= total; i++) {
        const event = new CloudEvent({
          type: 'com.kubemq.examples.eventsstore.reconnect',
          source: 'kubemq-ce-js-example',
          subject: channel,
          datacontenttype: 'application/json',
          data: { n: i },
        });
        const msg = HTTP.structured(event);
        await fetch(`${base}/ce/send/event-store`, {
          method: 'POST',
          headers: msg.headers as Record<string, string>,
          body: msg.body as string,
        });
      }
      console.log(`Published ${total} events.`);
      await new Promise((r) => setTimeout(r, 200));

      console.log(`\nFirst connection (reading first ${firstBatch} events):`);
      const lastId = await readNEvents(base, channel, 'js-reconnect-sub', firstBatch);
      console.log(`Disconnected. Last-Event-ID: ${lastId}`);

      console.log(`\nReconnecting with Last-Event-ID=${lastId}:`);
      await readNEvents(base, channel, 'js-reconnect-sub', total - firstBatch, lastId);
      console.log('\nReconnect-resume complete.');
    }

    main().catch(console.error);
    ```
  </Tab>

  <Tab value="Python">
    ```python
    """events_store/reconnect_resume — SSE reconnect with Last-Event-ID."""

    from __future__ import annotations

    import json
    import os
    import time

    import requests
    from cloudevents.v1.conversion import to_structured
    from cloudevents.v1.http import CloudEvent


    def server_url() -> str:
        return os.environ.get("KUBEMQ_CE_URL", "http://localhost:9090")


    def read_n_events(base: str, channel: str, client_id: str,
                      n: int, last_event_id: str = "") -> str:
        """Open SSE, read n events, return last SSE id."""
        if last_event_id:
            sse_url = (f"{base}/ce/subscribe/events-store"
                       f"?client_id={client_id}&channel={channel}")
            extra_headers = {"Last-Event-ID": last_event_id}
        else:
            sse_url = (f"{base}/ce/subscribe/events-store"
                       f"?client_id={client_id}&channel={channel}&events_store_type=2")
            extra_headers = {}

        headers = {"Accept": "text/event-stream", **extra_headers}
        last_id = ""
        count = 0

        with requests.get(sse_url, stream=True, timeout=None, headers=headers) as resp:
            ev_type = data = sse_id = ""
            for line in resp.iter_lines(decode_unicode=True):
                if line == "":
                    if ev_type == "cloudevent" and data:
                        ce = json.loads(data)
                        count += 1
                        last_id = sse_id
                        print(f"  [{count}] id={sse_id} data={ce.get('data')}")
                        if count == n:
                            return last_id
                    ev_type = data = sse_id = ""
                    continue
                if line.startswith(":"):
                    continue
                if line.startswith("id:"):
                    sse_id = line[3:].strip()
                elif line.startswith("event:"):
                    ev_type = line[6:].strip()
                elif line.startswith("data:"):
                    data = line[5:].strip()
        return last_id


    def main() -> None:
        base = server_url()
        channel = "python-ce-events-store.reconnect-resume"
        total = 6
        first_batch = 3

        for i in range(1, total + 1):
            event = CloudEvent(
                attributes={
                    "type": "com.kubemq.examples.eventsstore.reconnect",
                    "source": "kubemq-ce-python-example",
                    "subject": channel,
                    "datacontenttype": "application/json",
                },
                data={"n": i},
            )
            headers, body = to_structured(event)
            requests.post(f"{base}/ce/send/event-store", data=body,
                          headers=dict(headers), timeout=10)
        print(f"Published {total} events.")
        time.sleep(0.2)

        print(f"\nFirst connection (reading first {first_batch} events):")
        last_id = read_n_events(base, channel, "python-reconnect-sub", first_batch)
        print(f"Disconnected. Last-Event-ID: {last_id}")

        print(f"\nReconnecting with Last-Event-ID={last_id}:")
        read_n_events(base, channel, "python-reconnect-sub", total - first_batch, last_id)
        print("\nReconnect-resume complete.")


    if __name__ == "__main__":
        main()
    ```
  </Tab>

  <Tab value="Ruby">
    ```ruby
    # events_store/reconnect_resume — Last-Event-ID reconnect.
    require "net/http"; require "uri"; require "json"; require "cloud_events"; require "securerandom"
    def server_url = ENV.fetch("KUBEMQ_CE_URL", "http://localhost:9090")

    base = server_url; channel = "ruby-ce-events-store.reconnect-resume"
    sdk = CloudEvents::HttpBinding.default

    # Publish 4 events.
    puts "Publishing 4 events..."
    (1..4).each do |i|
      ev = CloudEvents::Event::V1.new(
        id: SecureRandom.uuid, type: "com.kubemq.examples.eventsstore.stored",
        source: URI("urn:kubemq-ce-ruby-example"), subject: channel, spec_version: "1.0",
        data_content_type: CloudEvents::ContentType.new("application/json"),
        data: JSON.generate({ seq: i })
      )
      enc_h, enc_b = sdk.encode_event(ev, structured_format: "json")
      uri = URI("#{base}/ce/send/event-store")
      Net::HTTP.start(uri.host, uri.port) do |http|
        req = Net::HTTP::Post.new(uri); enc_h.each{|k,v|req[k]=v}; req.body=enc_b
        http.request(req)
      end
    end

    # Helper: open one SSE connection, collect up to max_events, return [events, last_id].
    def subscribe_es(base, channel, last_event_id, max_events)
      # Omit events_store_type when reconnecting with Last-Event-ID.
      query = last_event_id.nil? \
        ? "events_store_type=2" \
        : ""  # no events_store_type on reconnect
      uri = URI("#{base}/ce/subscribe/events-store?client_id=ruby-es-reconnect&channel=#{URI.encode_www_form_component(channel)}#{query.empty? ? '' : '&' + query}")
      events = []; last_id = nil
      Net::HTTP.start(uri.host, uri.port, read_timeout: 12) do |http|
        req = Net::HTTP::Get.new(uri); req["Accept"] = "text/event-stream"
        req["Last-Event-ID"] = last_event_id if last_event_id
        http.request(req) do |resp|
          ev_type = nil; data = nil; id = nil
          resp.read_body do |chunk|
            chunk.each_line do |line|
              line.chomp!
              if line.empty?
                if ev_type == "cloudevent" && data
                  last_id = id if id
                  ce = JSON.parse(data)
                  ce["data"] = JSON.parse(ce["data"]) if ce["data"].is_a?(String)
                  events << ce
                  return [events, last_id] if events.size >= max_events
                end
                ev_type = nil; data = nil; id = nil
              elsif line.start_with?("id:") then id = line.sub("id:","").strip
              elsif line.start_with?("event:") then ev_type = line.sub("event:","").strip
              elsif line.start_with?("data:") then data = line.sub("data:","").strip
              end
            end
          end
        end
      end
      [events, last_id]
    end

    # First connection: receive 2 events.
    puts "First connection — receiving 2 events:"
    first_events, last_id = subscribe_es(base, channel, nil, 2)
    first_events.each { |ce| puts "  Received: seq=#{ce.dig('data','seq')}" }
    puts "  Last-Event-ID recorded: #{last_id}"

    # Reconnect using Last-Event-ID.
    puts "Reconnecting with Last-Event-ID=#{last_id}..."
    second_events, _ = subscribe_es(base, channel, last_id, 2)
    second_events.each { |ce| puts "  Resumed: seq=#{ce.dig('data','seq')}" }
    puts "Reconnect-resume demonstration complete."
    ```
  </Tab>

  <Tab value="Rust">
    ```rust
    // events-store/reconnect-resume — Last-Event-ID reconnect.
    use bytes::Bytes;
    use cloudevents::{EventBuilder, EventBuilderV10};
    use futures_util::StreamExt;
    use reqwest::Client;
    use serde_json::{json, Value};
    use std::env;
    use uuid::Uuid;

    fn server_url() -> String {
        env::var("KUBEMQ_CE_URL").unwrap_or_else(|_| "http://localhost:9090".to_string())
    }

    /// Subscribe and collect up to `max` events. Returns (events, last_id_seen).
    async fn subscribe_and_collect(
        client: &Client,
        url: &str,
        last_event_id: Option<&str>,
        max: usize,
    ) -> (Vec<Value>, String) {
        let mut req = client.get(url).header("Accept", "text/event-stream");
        if let Some(id) = last_event_id {
            req = req.header("Last-Event-ID", id);
        }
        let stream = req.send().await.expect("SSE connect").bytes_stream();
        let mut stream = Box::pin(stream);
        let mut ev_type = String::new(); let mut data = String::new();
        let mut id_field = String::new(); let mut last_id = String::new();
        let mut buffer = String::new();
        let mut events = Vec::new();

        while let Some(chunk) = stream.next().await {
            let chunk: Bytes = chunk.unwrap_or_default();
            buffer.push_str(&String::from_utf8_lossy(&chunk));
            while let Some(pos) = buffer.find('\n') {
                let line = buffer[..pos].trim_end_matches('\r').to_string();
                buffer = buffer[pos + 1..].to_string();
                if line.is_empty() {
                    if ev_type == "cloudevent" && !data.is_empty() {
                        if !id_field.is_empty() { last_id = id_field.clone(); }
                        let ce: Value = serde_json::from_str(&data).unwrap_or(Value::Null);
                        events.push(ce);
                        if events.len() >= max { return (events, last_id); }
                    }
                    ev_type.clear(); data.clear(); id_field.clear();
                } else if line.starts_with(':') {
                } else if let Some(v) = line.strip_prefix("id:") { id_field = v.trim().to_string(); }
                else if let Some(v) = line.strip_prefix("event:") { ev_type = v.trim().to_string(); }
                else if let Some(v) = line.strip_prefix("data:") { data = v.trim().to_string(); }
            }
        }
        (events, last_id)
    }

    #[tokio::main]
    async fn main() -> Result<(), Box<dyn std::error::Error>> {
        let base = server_url();
        let channel = "rust-ce-events-store.reconnect-resume";
        let client = Client::new();

        // Publish 4 events.
        println!("Publishing 4 events...");
        for i in 1..=4u32 {
            let event = EventBuilderV10::new()
                .id(Uuid::new_v4().to_string())
                .ty("com.kubemq.examples.eventsstore.stored")
                .source("urn:kubemq-ce-rust-example")
                .subject(channel)
                .data("application/json", json!({"seq": i}))
                .build()?;
            let body = serde_json::to_string(&event)?;
            client.post(format!("{}/ce/send/event-store", base))
                .header("Content-Type", "application/cloudevents+json")
                .body(body).send().await?;
        }

        // First connection: StartFromFirst (events_store_type=2), receive 2 events.
        let first_url = format!(
            "{}/ce/subscribe/events-store?client_id=rust-es-reconnect&channel={}&events_store_type=2",
            base, channel
        );
        println!("First connection — receiving 2 events:");
        let (first_events, last_id) = subscribe_and_collect(&client, &first_url, None, 2).await;
        for ce in &first_events {
            println!("  Received: seq={}", ce["data"]["seq"]);
        }
        println!("  Last-Event-ID recorded: {}", last_id);

        // Reconnect using Last-Event-ID (no events_store_type param).
        let reconnect_url = format!(
            "{}/ce/subscribe/events-store?client_id=rust-es-reconnect&channel={}",
            base, channel
        );
        println!("Reconnecting with Last-Event-ID={}...", last_id);
        let (second_events, _) = subscribe_and_collect(&client, &reconnect_url, Some(&last_id), 2).await;
        for ce in &second_events {
            println!("  Resumed: seq={}", ce["data"]["seq"]);
        }
        println!("Reconnect-resume demonstration complete.");
        Ok(())
    }
    ```
  </Tab>
</Tabs>

## Client implementation tips [#client-implementation-tips]

* **Standard `EventSource` clients** (browsers, `eventsource` libraries) handle keepalive comments, `event:` dispatch, and `Last-Event-ID` resumption automatically. They cannot, however, set a custom `Last-Event-ID` on the first manual reconnect — use a raw HTTP read for that case, as the JavaScript example does.
* **Manual parsers** (the Go, Java, Python, Ruby, Rust, and C# tabs) read the body line by line: an empty line dispatches the current frame, a line starting with `:` is a keepalive to skip, and `id:` / `event:` / `data:` accumulate frame state. Branch on the `event:` value so `cloudevent`, `message`, and `error` frames are each handled.
* **Always handle the `error` event.** Treat `stream idle timeout` as a normal lifecycle signal and reconnect (with `Last-Event-ID` for events-store) rather than as a fatal failure.

## Related [#related]

<Cards>
  <Card title="Events Store" href="/connectors/cloudevents/how-to/events-store" description="The 6 start positions and replay semantics that drive Last-Event-ID reconnection." />

  <Card title="Channel resolution" href="/connectors/cloudevents/how-to/channel-resolution" description="How subject, ?channel=, and ClientID map a subscription to a KubeMQ channel." />

  <Card title="Content modes" href="/connectors/cloudevents/how-to/content-modes" description="Structured vs binary encoding for the CloudEvents delivered on the stream." />

  <Card title="Configuration" href="/connectors/cloudevents/concepts/configuration-model" description="Tune MaxSSEIdleSeconds, MaxSSEConnections, and SubBuffSize." />
</Cards>