SSE Behavior

# 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"

// 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.");

// 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.")
}

// 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.");
    }
}

// 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);

"""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()

# 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."

// 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(())
}