Resume After Disconnect

package main

import (
    "context"
    "fmt"
    "log"

    "github.com/kubemq-io/kubemq-go/v2"
)

func main() {
    ctx := context.Background()
    client, err := kubemq.NewClient(ctx,
        kubemq.WithAddress("localhost", 50000),
    )
    if err != nil {
        log.Fatal(err)
    }
    defer client.Close()

    sub, err := client.SubscribeToEventsStore(ctx,
        "orders.events",
        "order-processor",
        kubemq.StartFromFirst(),
        kubemq.WithOnEvent(func(event *kubemq.Event) {
            fmt.Printf("Processing seq=%d: %s\n",
                event.Sequence, string(event.Body))
        }),
        kubemq.WithOnError(func(err error) {
            log.Println("Error:", err)
        }),
    )
    if err != nil {
        log.Fatal(err)
    }
    defer sub.Unsubscribe()

    <-ctx.Done()
}

import time
from kubemq import (
    PubSubClient, EventsStoreSubscription,
    EventStoreStartPosition, CancellationToken,
)

def on_event(event):
    print(f"Processing seq={event.sequence}: {event.body.decode('utf-8')}")

with PubSubClient(address="localhost:50000") as client:
    client.subscribe_to_events_store(
        subscription=EventsStoreSubscription(
            channel="orders.events",
            group="order-processor",
            start_position=EventStoreStartPosition.StartFromFirst,
            on_receive_event_callback=on_event,
            on_error_callback=lambda e: print(f"Error: {e}"),
        ),
        cancel=CancellationToken(),
    )
    time.sleep(300)

import { KubeMQClient, EventStoreStartPosition } from 'kubemq-js';

const client = await KubeMQClient.create({ address: 'localhost:50000' });

client.subscribeToEventsStore({
  channel: 'orders.events',
  group: 'order-processor',
  startPosition: EventStoreStartPosition.StartFromFirst,
  onEvent: (msg) =>
    console.log(`Processing seq=${msg.sequence}: ${new TextDecoder().decode(msg.body)}`),
  onError: (err) => console.error('Error:', err.message),
});

PubSubClient client = PubSubClient.builder()
    .address("localhost:50000")
    .clientId("order-processor")
    .build();

client.subscribeToEventsStore(EventsStoreSubscription.builder()
    .channel("orders.events")
    .group("order-processor")
    .startPosition(EventStoreStartPosition.StartFromFirst)
    .onReceiveEventCallback(event ->
        System.out.printf("Processing seq=%d: %s%n",
            event.getSequence(), new String(event.getBody())))
    .onErrorCallback(err -> System.err.println("Error: " + err.getMessage()))
    .build());

Thread.sleep(300_000);
client.close();

await using var client = new KubeMQClient(new KubeMQClientOptions());
await client.ConnectAsync();

await foreach (var msg in client.SubscribeToEventsStoreAsync(
    new EventsStoreSubscription
    {
        Channel = "orders.events",
        Group = "order-processor",
        StartPosition = EventStoreStartPosition.StartFromFirst,
    }))
{
    Console.WriteLine($"Processing seq={msg.Sequence}: "
        + $"{Encoding.UTF8.GetString(msg.Body.Span)}");
}

val client = KubeMQClient.pubSub {
    address = "localhost:50000"
    clientId = "order-processor"
}

client.use {
    client.subscribeToEventsStore {
        channel = "orders.events"
        group = "order-processor"
        startPosition = StartPosition.StartFromFirst
    }.collect { msg ->
        println("Processing seq=${msg.sequence}: ${String(msg.body)}")
    }
}

kubemq::ClientOptions options;
options.set_address("localhost", 50000);
options.set_client_id("order-processor");
auto client = kubemq::Client::Create(options).value();

client->SubscribeToEventsStore("orders.events", "order-processor",
    kubemq::StartPosition::StartFromFirst,
    [](const kubemq::EventStoreReceived& msg) {
        std::cout << "Processing seq=" << msg.sequence()
                  << ": " << msg.body() << std::endl;
    },
    [](const std::string& err) {
        std::cerr << "Error: " << err << std::endl;
    });

use kubemq::prelude::*;
use kubemq::EventsStoreSubscription;
use std::time::Duration;

#[tokio::main]
async fn main() -> kubemq::Result<()> {
    let client = KubemqClient::builder()
        .host("localhost")
        .port(50000)
        .build()
        .await?;

    // Named group enables durable position tracking.
    let sub = client
        .subscribe_to_events_store(
            "orders.events",
            "order-processor",
            EventsStoreSubscription::StartFromFirst,
            |event| {
                Box::pin(async move {
                    println!(
                        "Processing seq={}: {}",
                        event.sequence,
                        String::from_utf8_lossy(&event.body)
                    );
                })
            },
            None,
        )
        .await?;

    tokio::time::sleep(Duration::from_secs(300)).await;
    sub.unsubscribe().await;
    client.close().await?;
    Ok(())
}

require 'kubemq'

client = KubeMQ::PubSubClient.new(address: 'localhost:50000', client_id: 'order-processor')
cancel = KubeMQ::CancellationToken.new

# Named group enables durable position tracking.
subscription = KubeMQ::PubSub::EventsStoreSubscription.new(
  channel: 'orders.events',
  group: 'order-processor',
  start_position: KubeMQ::PubSub::EventStoreStartPosition::START_FROM_FIRST
)

client.subscribe_to_events_store(subscription, cancellation_token: cancel, on_error: lambda { |e|
  puts "Error: #{e.message}"
}) do |event|
  puts "Processing seq=#{event.sequence}: #{event.body}"
end

sleep 300
cancel.cancel
client.close

{:ok, client} =
  KubeMQ.Client.start_link(address: "localhost:50000", client_id: "order-processor")

# Named group enables durable position tracking.
{:ok, _sub} =
  KubeMQ.Client.subscribe_to_events_store(client, "orders.events",
    start_at: :start_from_first,
    group: "order-processor",
    on_event: fn event ->
      IO.puts("Processing seq=#{event.sequence}: #{event.body}")
    end,
    on_error: fn err -> IO.puts("Error: #{inspect(err)}") end
  )

Process.sleep(300_000)
KubeMQ.Client.close(client)