Events Store

package main

import (
	"fmt"
	"log"
	"net/url"
	"os"
	"time"

	"github.com/go-stomp/stomp/v3"
)

const destination = "/topic-store/audit" // Events Store pattern → channel audit
const total = 3

func addr() (network, host string) {
	u, _ := url.Parse(os.Getenv("KUBEMQ_STOMP_URL"))
	if u == nil || u.Host == "" {
		return "tcp", "localhost:61613"
	}
	return "tcp", u.Host
}

func main() {
	network, host := addr()

	// 1. Persist — SEND `total` messages to the store.
	pub, err := stomp.Dial(network, host)
	if err != nil {
		log.Fatalf("dial: %v", err)
	}
	for i := 1; i <= total; i++ {
		if err := pub.Send(destination, "text/plain",
			[]byte(fmt.Sprintf("evt-%d", i)), stomp.SendOpt.Receipt); err != nil {
			log.Fatalf("send: %v", err)
		}
	}
	_ = pub.Disconnect()

	// 2. Replay — SUBSCRIBE start-from:first to read the full history in order.
	sub, err := stomp.Dial(network, host)
	if err != nil {
		log.Fatalf("dial: %v", err)
	}
	defer func() { _ = sub.Disconnect() }()
	subscription, err := sub.Subscribe(destination, stomp.AckAuto,
		stomp.SubscribeOpt.Header("start-from", "first"))
	if err != nil {
		log.Fatalf("subscribe: %v", err)
	}

	for got := 0; got < total; got++ {
		select {
		case msg := <-subscription.C:
			fmt.Printf("replayed: %s\n", string(msg.Body))
		case <-time.After(10 * time.Second):
			log.Fatal("timed out replaying the store")
		}
	}
}

import os
import queue
from urllib.parse import urlparse

import stomp

DESTINATION = "/topic-store/audit"  # Events Store pattern → channel audit
TOTAL = 3


def endpoint() -> tuple[str, int]:
    parsed = urlparse(os.environ.get("KUBEMQ_STOMP_URL", "tcp://localhost:61613"))
    return parsed.hostname or "localhost", parsed.port or 61613


class Listener(stomp.ConnectionListener):
    def __init__(self) -> None:
        self.inbox: queue.Queue = queue.Queue()

    def on_message(self, frame) -> None:
        self.inbox.put(frame)


def main() -> None:
    host, port = endpoint()

    # 1. Persist — SEND `total` messages to the store.
    pub = stomp.Connection12([(host, port)], heartbeats=(10000, 10000))
    pub.connect(wait=True)
    for i in range(1, TOTAL + 1):
        pub.send(DESTINATION, f"evt-{i}", content_type="text/plain")
    pub.disconnect()

    # 2. Replay — SUBSCRIBE start-from:first to read the full history in order.
    listener = Listener()
    sub = stomp.Connection12([(host, port)], heartbeats=(10000, 10000))
    sub.set_listener("l", listener)
    sub.connect(wait=True)
    sub.subscribe(DESTINATION, id="audit", ack="auto", headers={"start-from": "first"})

    for _ in range(TOTAL):
        frame = listener.inbox.get(timeout=10)
        print(f"replayed: {frame.body}")
    sub.disconnect()


if __name__ == "__main__":
    main()

import java.lang.reflect.Type;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;

import org.springframework.messaging.simp.stomp.StompHeaders;
import org.springframework.messaging.simp.stomp.StompSession;
import org.springframework.messaging.simp.stomp.StompSessionHandlerAdapter;
import org.springframework.messaging.simp.stomp.ReactorNettyTcpStompClient;

public final class Main {
    private static final String DESTINATION = "/topic-store/audit"; // → channel audit
    private static final int TOTAL = 3;

    public static void main(String[] args) throws Exception {
        String url = System.getenv().getOrDefault("KUBEMQ_STOMP_URL", "tcp://localhost:61613");
        java.net.URI u = java.net.URI.create(url);
        ReactorNettyTcpStompClient client = new ReactorNettyTcpStompClient(u.getHost(),
                u.getPort() > 0 ? u.getPort() : 61613);

        // 1. Persist — SEND `total` messages to the store.
        StompSession pub = client.connectAsync(new StompSessionHandlerAdapter() {}).get(10, TimeUnit.SECONDS);
        for (int i = 1; i <= TOTAL; i++) {
            StompHeaders h = new StompHeaders();
            h.setDestination(DESTINATION);
            h.add("content-type", "text/plain");
            pub.send(h, ("evt-" + i).getBytes());
        }
        Thread.sleep(500);
        pub.disconnect();

        // 2. Replay — SUBSCRIBE start-from:first to read the full history in order.
        BlockingQueue<String> inbox = new ArrayBlockingQueue<>(TOTAL);
        StompSession sub = client.connectAsync(new StompSessionHandlerAdapter() {}).get(10, TimeUnit.SECONDS);
        StompHeaders subHeaders = new StompHeaders();
        subHeaders.setDestination(DESTINATION);
        subHeaders.setId("audit");
        subHeaders.setAck("auto");
        subHeaders.add("start-from", "first");
        sub.subscribe(subHeaders, new StompSessionHandlerAdapter() {
            @Override public Type getPayloadType(StompHeaders headers) { return String.class; }
            @Override public void handleFrame(StompHeaders headers, Object payload) {
                inbox.add(payload == null ? "" : payload.toString());
            }
        });

        for (int i = 0; i < TOTAL; i++) {
            String body = inbox.poll(10, TimeUnit.SECONDS);
            if (body == null) throw new IllegalStateException("timed out replaying the store");
            System.out.printf("replayed: %s%n", body);
        }
        sub.disconnect();
        client.stop();
    }
}

import { connect, type Client } from "stompit";

const DESTINATION = "/topic-store/audit"; // Events Store pattern → channel audit
const TOTAL = 3;

function endpoint(): { host: string; port: number } {
  const url = new URL(process.env["KUBEMQ_STOMP_URL"] ?? "tcp://localhost:61613");
  return { host: url.hostname, port: Number(url.port) || 61613 };
}

function open(): Promise<Client> {
  const { host, port } = endpoint();
  return new Promise((resolve, reject) => {
    connect({ host, port, connectHeaders: { "accept-version": "1.2", "heart-beat": "10000,10000" } },
      (err, client) => (err ? reject(err) : resolve(client)));
  });
}

async function main(): Promise<void> {
  // 1. Persist — SEND `total` messages to the store.
  const pub = await open();
  for (let i = 1; i <= TOTAL; i++) {
    const frame = pub.send({ destination: DESTINATION, "content-type": "text/plain" });
    frame.write(`evt-${i}`);
    frame.end();
  }
  await new Promise<void>((r) => pub.disconnect(() => r()));

  // 2. Replay — SUBSCRIBE start-from:first to read the full history in order.
  const sub = await open();
  let got = 0;
  await new Promise<void>((resolve, reject) => {
    const timer = setTimeout(() => reject(new Error("timed out replaying the store")), 15_000);
    sub.subscribe({ destination: DESTINATION, ack: "auto", "start-from": "first" }, (err, message) => {
      if (err) return reject(err);
      message.readString("utf-8", (readErr, body) => {
        if (readErr) return reject(readErr);
        console.log(`replayed: ${body}`);
        if (++got === TOTAL) {
          clearTimeout(timer);
          resolve();
        }
      });
    });
  });
  await new Promise<void>((r) => sub.disconnect(() => r()));
}

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

using System.Text;
using Stomp.Net;

var url = Environment.GetEnvironmentVariable("KUBEMQ_STOMP_URL") ?? "tcp://localhost:61613";
var uri = new Uri(url);
const string destination = "/topic-store/audit"; // Events Store pattern → channel audit
const int total = 3;

string brokerUri = $"stomp:tcp://{uri.Host}:{(uri.Port > 0 ? uri.Port : 61613)}";
var factory = new ConnectionFactory(brokerUri) { UserName = "kubemq", Password = "" };

// 1. Persist — SEND `total` messages to the store.
using (var pubConn = factory.CreateConnection())
{
    pubConn.Start();
    using var session = pubConn.CreateSession(AcknowledgementMode.AutoAcknowledge);
    using var producer = session.CreateProducer(session.GetTopic(destination));
    for (var i = 1; i <= total; i++)
    {
        var msg = producer.CreateBytesMessage(Encoding.UTF8.GetBytes($"evt-{i}"));
        msg.StompType = "text/plain";
        producer.Send(msg);
    }
}

// 2. Replay — SUBSCRIBE start-from:first to read the full history in order.
using var conn = factory.CreateConnection();
conn.Start();
using var consumeSession = conn.CreateSession(AcknowledgementMode.IndividualAcknowledge);
// Stomp.Net forwards extra consumer headers, including the store replay directive.
var topic = consumeSession.GetTopic(destination);
using var consumer = consumeSession.CreateConsumer(topic, null, false,
    new Dictionary<string, string> { ["start-from"] = "first" });
for (var i = 0; i < total; i++)
{
    var msg = consumer.Receive(TimeSpan.FromSeconds(10))
        ?? throw new InvalidOperationException("timed out replaying the store");
    Console.WriteLine($"replayed: {Encoding.UTF8.GetString(msg.Content)}");
}

require "stomp"
require "uri"
require "timeout"

uri = URI.parse(ENV.fetch("KUBEMQ_STOMP_URL", "tcp://localhost:61613"))
hosts = [{ host: uri.host, port: uri.port || 61613, login: "kubemq", passcode: "" }]
DESTINATION = "/topic-store/audit" # Events Store pattern → channel audit
TOTAL = 3

# 1. Persist — SEND `total` messages to the store.
pub = Stomp::Client.new(hosts: hosts, connect_headers: { "accept-version" => "1.2", "host" => "/" })
(1..TOTAL).each { |i| pub.publish(DESTINATION, "evt-#{i}", "content-type" => "text/plain") }
pub.close

# 2. Replay — SUBSCRIBE start-from:first to read the full history in order.
inbox = Thread::Queue.new
sub = Stomp::Client.new(hosts: hosts, connect_headers: { "accept-version" => "1.2", "host" => "/" })
sub.subscribe(DESTINATION, id: "audit", ack: "auto", "start-from" => "first") { |msg| inbox << msg }

TOTAL.times do
  msg = Timeout.timeout(10) { inbox.pop }
  puts "replayed: #{msg.body}"
end
sub.close

use std::time::Duration;

use async_stomp::client::Connector;
use async_stomp::{AckMode, FromServer, Message, ToServer};
use futures::{SinkExt, StreamExt};

const DESTINATION: &str = "/topic-store/audit"; // Events Store pattern → channel audit
const TOTAL: usize = 3;

fn host_port() -> (String, u16) {
    let url = std::env::var("KUBEMQ_STOMP_URL").unwrap_or_else(|_| "tcp://localhost:61613".into());
    let hp = url.trim_start_matches("tcp://").trim_start_matches("tls://");
    let mut parts = hp.splitn(2, ':');
    let host = parts.next().unwrap_or("localhost").to_string();
    let port = parts.next().and_then(|p| p.parse().ok()).unwrap_or(61613);
    (host, port)
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let (host, port) = host_port();

    // 1. Persist — SEND `total` messages to the store.
    let mut pub_conn = Connector::builder()
        .server(format!("{host}:{port}"))
        .virtualhost(&host)
        .connect()
        .await?;
    for i in 1..=TOTAL {
        pub_conn
            .send(ToServer::Send {
                destination: DESTINATION.into(),
                transaction: None,
                headers: Some(vec![("content-type".into(), "text/plain".into())]),
                body: Some(format!("evt-{i}").into_bytes()),
            }.into())
            .await?;
    }
    pub_conn.send(ToServer::Disconnect { receipt: None }.into()).await?;

    // 2. Replay — SUBSCRIBE start-from:first via an extra header.
    let mut sub = Connector::builder()
        .server(format!("{host}:{port}"))
        .virtualhost(&host)
        .connect()
        .await?;
    let mut subscribe: Message<ToServer> = ToServer::Subscribe {
        destination: DESTINATION.into(),
        id: "audit".into(),
        ack: Some(AckMode::Auto),
    }.into();
    subscribe.extra_headers = vec![(b"start-from".to_vec(), b"first".to_vec())];
    sub.send(subscribe).await?;

    for _ in 0..TOTAL {
        let frame = tokio::time::timeout(Duration::from_secs(10), sub.next())
            .await?
            .ok_or("stream closed")??;
        if let FromServer::Message { body, .. } = frame.content {
            println!("replayed: {}", String::from_utf8_lossy(&body.unwrap_or_default()));
        }
    }
    Ok(())
}