Events Store

package main

import (
	"context"
	"fmt"
	"log"
	"net"
	"os"
	"strings"
	"time"

	"github.com/eclipse/paho.golang/paho"
	"github.com/google/uuid"
)

const topic = "store/demo/s" // store/ prefix -> Events Store; channel demo.s

func brokerURL() string {
	if u := os.Getenv("KUBEMQ_MQTT_URL"); u != "" {
		return u
	}
	return "tcp://localhost:1883"
}

func tcpAddr(raw string) string {
	for _, pfx := range []string{"tcp://", "ws://", "tls://"} {
		if strings.HasPrefix(raw, pfx) {
			return raw[len(pfx):]
		}
	}
	return raw
}

func dial(ctx context.Context, addr, id string, onMsg func(paho.PublishReceived) (bool, error)) *paho.Client {
	conn, err := net.Dial("tcp", addr)
	if err != nil {
		log.Fatalf("dial: %v", err)
	}
	cfg := paho.ClientConfig{Conn: conn}
	if onMsg != nil {
		cfg.OnPublishReceived = []func(paho.PublishReceived) (bool, error){onMsg}
	}
	c := paho.NewClient(cfg)
	ack, err := c.Connect(ctx, &paho.Connect{ClientID: id, KeepAlive: 30, CleanStart: true})
	if err != nil {
		log.Fatalf("connect: %v", err)
	}
	if ack.ReasonCode != 0 {
		log.Fatalf("CONNACK reason=0x%02X", ack.ReasonCode)
	}
	return c
}

func main() {
	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
	defer cancel()
	addr := tcpAddr(brokerURL())
	sfx := uuid.NewString()[:8]

	// 1. PUBLISH before subscribing — this is stored but never replayed over MQTT.
	pub := dial(ctx, addr, "go-store-pub-"+sfx, nil)
	if _, err := pub.Publish(ctx, &paho.Publish{Topic: topic, QoS: 1, Payload: []byte("pre-subscribe")}); err != nil {
		log.Fatalf("pre-publish: %v", err)
	}
	time.Sleep(300 * time.Millisecond)

	// 2. SUBSCRIBE — the connector registers a StartNewOnly subscription.
	got := make(chan string, 1)
	sub := dial(ctx, addr, "go-store-sub-"+sfx, func(pr paho.PublishReceived) (bool, error) {
		got <- string(pr.Packet.Payload)
		return true, nil
	})
	subAck, err := sub.Subscribe(ctx, &paho.Subscribe{
		Subscriptions: []paho.SubscribeOptions{{Topic: topic, QoS: 1}},
	})
	if err != nil {
		log.Fatalf("subscribe: %v", err)
	}
	// Wildcards on store/ are rejected with 0xA2.
	if subAck.Reasons[0] > 2 {
		log.Fatalf("SUBACK reason=0x%02X", subAck.Reasons[0])
	}
	time.Sleep(700 * time.Millisecond) // let the StartNewOnly subscription go live

	// 3. PUBLISH after subscribing — only this message is delivered.
	if _, err := pub.Publish(ctx, &paho.Publish{Topic: topic, QoS: 1, Payload: []byte("post-subscribe")}); err != nil {
		log.Fatalf("post-publish: %v", err)
	}

	select {
	case msg := <-got:
		fmt.Printf("received: %s (StartNewOnly — the pre-subscribe message was not replayed)\n", msg)
	case <-ctx.Done():
		log.Fatal("timed out waiting for the post-subscribe message")
	}
	_ = pub.Disconnect(&paho.Disconnect{ReasonCode: 0})
	_ = sub.Disconnect(&paho.Disconnect{ReasonCode: 0})
}

import os
import threading
import time

import paho.mqtt.client as mqtt
from paho.mqtt.enums import CallbackAPIVersion

TOPIC = "store/demo/s"  # store/ prefix -> Events Store; channel demo.s


def parse_url(url: str) -> tuple[str, int]:
    scheme, rest = url.split("://", 1)
    host, _, port = rest.rstrip("/").partition(":")
    return host, int(port) if port else {"tcp": 1883, "tls": 8883, "ws": 8083}[scheme]


def make_client(client_id: str) -> mqtt.Client:
    return mqtt.Client(CallbackAPIVersion.VERSION2, client_id=client_id, protocol=mqtt.MQTTv5)


def main() -> None:
    host, port = parse_url(os.environ.get("KUBEMQ_MQTT_URL", "tcp://localhost:1883"))

    # 1. PUBLISH before subscribing — stored but never replayed over MQTT.
    pub = make_client("py-store-pub")
    pub.connect(host, port, keepalive=30)
    pub.loop_start()
    time.sleep(0.3)
    pub.publish(TOPIC, payload=b"pre-subscribe", qos=1).wait_for_publish(10)
    time.sleep(0.3)

    # 2. SUBSCRIBE — the connector registers a StartNewOnly subscription.
    received = threading.Event()
    payload: list[bytes] = []
    sub = make_client("py-store-sub")
    sub.on_connect = lambda c, *_: c.subscribe(TOPIC, qos=1)
    sub.on_message = lambda c, u, m: (payload.append(m.payload), received.set())
    sub.connect(host, port, keepalive=30)
    sub.loop_start()
    time.sleep(0.7)  # let the StartNewOnly subscription go live

    # 3. PUBLISH after subscribing — only this message is delivered.
    pub.publish(TOPIC, payload=b"post-subscribe", qos=1).wait_for_publish(10)

    if not received.wait(timeout=10):
        raise TimeoutError("timed out waiting for the post-subscribe message")
    print(f"received: {payload[0].decode()!r} "
          "(StartNewOnly — the pre-subscribe message was not replayed)")

    pub.loop_stop(); pub.disconnect()
    sub.loop_stop(); sub.disconnect()


if __name__ == "__main__":
    main()

import java.nio.charset.StandardCharsets;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

import org.eclipse.paho.mqttv5.client.MqttAsyncClient;
import org.eclipse.paho.mqttv5.client.MqttCallback;
import org.eclipse.paho.mqttv5.client.MqttConnectionOptions;
import org.eclipse.paho.mqttv5.client.MqttDisconnectResponse;
import org.eclipse.paho.mqttv5.client.IMqttToken;
import org.eclipse.paho.mqttv5.common.MqttException;
import org.eclipse.paho.mqttv5.common.MqttMessage;
import org.eclipse.paho.mqttv5.common.packet.MqttProperties;

public final class Main {
    // store/ prefix -> Events Store; KubeMQ channel demo.s ('/' -> '.').
    private static final String TOPIC = "store/demo/s";

    public static void main(String[] args) throws Exception {
        String broker = System.getenv().getOrDefault("KUBEMQ_MQTT_URL", "tcp://localhost:1883");
        MqttConnectionOptions opts = new MqttConnectionOptions();
        opts.setCleanStart(true);
        opts.setKeepAliveInterval(30);

        // 1. PUBLISH before subscribing — stored but never replayed over MQTT.
        MqttAsyncClient pub = new MqttAsyncClient(broker, "java-store-pub-" + UUID.randomUUID().toString().substring(0, 8));
        pub.connect(opts).waitForCompletion(10_000);
        pub.publish(TOPIC, msg("pre-subscribe")).waitForCompletion(5_000);
        Thread.sleep(300);

        // 2. SUBSCRIBE — the connector registers a StartNewOnly subscription.
        CountDownLatch received = new CountDownLatch(1);
        String[] body = {null};
        MqttAsyncClient sub = new MqttAsyncClient(broker, "java-store-sub-" + UUID.randomUUID().toString().substring(0, 8));
        sub.setCallback(new MqttCallback() {
            public void messageArrived(String t, MqttMessage m) {
                body[0] = new String(m.getPayload(), StandardCharsets.UTF_8);
                received.countDown();
            }
            public void disconnected(MqttDisconnectResponse r) {}
            public void mqttErrorOccurred(MqttException e) {}
            public void deliveryComplete(IMqttToken t) {}
            public void connectComplete(boolean reconnect, String uri) {}
            public void authPacketArrived(int code, MqttProperties props) {}
        });
        sub.connect(opts).waitForCompletion(10_000);
        IMqttToken subToken = sub.subscribe(TOPIC, 1);
        subToken.waitForCompletion(10_000);
        // Wildcards on store/ are rejected with 0xA2.
        if (subToken.getGrantedQos()[0] >= 0x80) {
            throw new IllegalStateException("subscription rejected");
        }
        Thread.sleep(700); // let the StartNewOnly subscription go live

        // 3. PUBLISH after subscribing — only this message is delivered.
        pub.publish(TOPIC, msg("post-subscribe")).waitForCompletion(5_000);

        if (!received.await(10, TimeUnit.SECONDS)) {
            throw new IllegalStateException("timed out waiting for the post-subscribe message");
        }
        System.out.printf("received: %s (StartNewOnly — the pre-subscribe message was not replayed)%n", body[0]);

        pub.disconnect().waitForCompletion(5_000); pub.close();
        sub.disconnect().waitForCompletion(5_000); sub.close();
    }

    private static MqttMessage msg(String body) {
        MqttMessage m = new MqttMessage(body.getBytes(StandardCharsets.UTF_8));
        m.setQos(1);
        m.setRetained(false); // retain is silently dropped by the connector
        return m;
    }
}

import mqtt, { type MqttClient } from "mqtt";

const TOPIC = "store/demo/s"; // store/ prefix -> Events Store; channel demo.s

function connect(url: string, clientId: string): Promise<MqttClient> {
  return new Promise((resolve, reject) => {
    const client = mqtt.connect(url, { clientId, protocolVersion: 5, clean: true, keepalive: 30 });
    client.once("connect", () => resolve(client));
    client.once("error", reject);
  });
}

function publish(client: MqttClient, payload: string): Promise<void> {
  return new Promise((resolve, reject) => {
    client.publish(TOPIC, payload, { qos: 1 }, (err) => (err ? reject(err) : resolve()));
  });
}

async function main(): Promise<void> {
  const url = process.env["KUBEMQ_MQTT_URL"] ?? "tcp://localhost:1883";

  // 1. PUBLISH before subscribing — stored but never replayed over MQTT.
  const publisher = await connect(url, "js-store-pub");
  await publish(publisher, "pre-subscribe");
  await new Promise<void>((r) => setTimeout(r, 300));

  // 2. SUBSCRIBE — the connector registers a StartNewOnly subscription.
  const subscriber = await connect(url, "js-store-sub");
  const received = new Promise<string>((resolve) => {
    subscriber.on("message", (_topic, payload) => resolve(payload.toString()));
  });
  await new Promise<void>((resolve, reject) => {
    subscriber.subscribe(TOPIC, { qos: 1 }, (err) => (err ? reject(err) : resolve()));
  });
  await new Promise<void>((r) => setTimeout(r, 700)); // let the subscription go live

  // 3. PUBLISH after subscribing — only this message is delivered.
  await publish(publisher, "post-subscribe");

  console.log(`received: ${await received} `
    + "(StartNewOnly — the pre-subscribe message was not replayed)");

  await publisher.endAsync();
  await subscriber.endAsync();
}

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

using System.Text;
using MQTTnet;
using MQTTnet.Client;
using MQTTnet.Protocol;

static (string host, int port) Endpoint()
{
    var url = Environment.GetEnvironmentVariable("KUBEMQ_MQTT_URL") ?? "tcp://localhost:1883";
    foreach (var pfx in new[] { "tcp://", "tls://", "ws://" })
        if (url.StartsWith(pfx, StringComparison.OrdinalIgnoreCase)) url = url[pfx.Length..];
    var parts = url.TrimEnd('/').Split(':');
    return (parts[0], parts.Length > 1 ? int.Parse(parts[1]) : 1883);
}

var (host, port) = Endpoint();
const string topic = "store/demo/s"; // store/ prefix -> Events Store; channel demo.s

var factory = new MqttFactory();

MqttClientOptions Options(string id) => new MqttClientOptionsBuilder()
    .WithTcpServer(host, port)
    .WithClientId($"{id}-{Guid.NewGuid():N}"[..26])
    .WithProtocolVersion(MQTTnet.Formatter.MqttProtocolVersion.V500)
    .WithCleanSession(true)
    .Build();

async Task Publish(IMqttClient client, string body)
{
    var result = await client.PublishAsync(new MqttApplicationMessageBuilder()
        .WithTopic(topic)
        .WithPayload(Encoding.UTF8.GetBytes(body))
        .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce)
        .Build());
    if (result.ReasonCode != MqttClientPublishReasonCode.Success)
        throw new Exception($"PUBACK reason: {result.ReasonCode}");
}

// 1. PUBLISH before subscribing — stored but never replayed over MQTT.
using var pub = factory.CreateMqttClient();
await pub.ConnectAsync(Options("csharp-store-pub"));
await Publish(pub, "pre-subscribe");
await Task.Delay(300);

// 2. SUBSCRIBE — the connector registers a StartNewOnly subscription.
var received = new TaskCompletionSource<string>(TaskCreationOptions.RunContinuationsAsynchronously);
using var sub = factory.CreateMqttClient();
sub.ApplicationMessageReceivedAsync += e =>
{
    received.TrySetResult(Encoding.UTF8.GetString(e.ApplicationMessage.PayloadSegment));
    return Task.CompletedTask;
};
await sub.ConnectAsync(Options("csharp-store-sub"));
var subResult = await sub.SubscribeAsync(new MqttClientSubscribeOptionsBuilder()
    .WithTopicFilter(f => f.WithTopic(topic).WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce))
    .Build());
// Wildcards on store/ are rejected with 0xA2.
if (subResult.Items.First().ResultCode > MqttClientSubscribeResultCode.GrantedQoS2)
    throw new Exception("subscription rejected");
await Task.Delay(700); // let the StartNewOnly subscription go live

// 3. PUBLISH after subscribing — only this message is delivered.
await Publish(pub, "post-subscribe");

using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10));
Console.WriteLine($"received: {await received.Task.WaitAsync(cts.Token)} "
    + "(StartNewOnly — the pre-subscribe message was not replayed)");

await pub.DisconnectAsync();
await sub.DisconnectAsync();

# The mqtt gem speaks MQTT 3.1.1. Events Store publish/subscribe works on any
# version; over MQTT the start position is always StartNewOnly regardless of client.
require "mqtt"
require "uri"
require "securerandom"
require "timeout"

uri = URI.parse(ENV.fetch("KUBEMQ_MQTT_URL", "tcp://localhost:1883"))
conn = { host: uri.host, port: uri.port, ssl: uri.scheme == "tls" }
topic = "store/demo/s" # store/ prefix -> Events Store; channel demo.s

# 1. PUBLISH before subscribing — stored but never replayed over MQTT.
MQTT::Client.connect(host: conn[:host], port: conn[:port], ssl: conn[:ssl],
                     client_id: "ruby-store-pub-#{SecureRandom.hex(4)}",
                     clean_session: true, keep_alive: 30) do |client|
  client.publish(topic, "pre-subscribe", false, 1)
end
sleep 0.3

# 2. SUBSCRIBE — the connector registers a StartNewOnly subscription.
received  = Queue.new
sub_ready = Queue.new
subscriber = Thread.new do
  MQTT::Client.connect(host: conn[:host], port: conn[:port], ssl: conn[:ssl],
                       client_id: "ruby-store-sub-#{SecureRandom.hex(4)}",
                       clean_session: true, keep_alive: 30) do |client|
    client.subscribe([topic, 1])
    sub_ready.push(:ready)
    _topic, payload = client.get
    received.push(payload)
  end
end
sub_ready.pop
sleep 0.7 # let the StartNewOnly subscription go live

# 3. PUBLISH after subscribing — only this message is delivered.
MQTT::Client.connect(host: conn[:host], port: conn[:port], ssl: conn[:ssl],
                     client_id: "ruby-store-pub2-#{SecureRandom.hex(4)}",
                     clean_session: true, keep_alive: 30) do |client|
  client.publish(topic, "post-subscribe", false, 1)
end

payload = Timeout.timeout(10) { received.pop }
puts "received: #{payload.inspect} (StartNewOnly — the pre-subscribe message was not replayed)"
subscriber.kill

use rumqttc::v5::mqttbytes::v5::Packet;
use rumqttc::v5::mqttbytes::QoS;
use rumqttc::v5::{AsyncClient, Event, MqttOptions};
use std::env;
use tokio::sync::oneshot;
use tokio::time::{timeout, Duration};
use uuid::Uuid;

const TOPIC: &str = "store/demo/s"; // store/ prefix -> Events Store; channel demo.s

fn parse_host_port(url: &str) -> (String, u16) {
    let stripped = url
        .trim_start_matches("tcp://")
        .trim_start_matches("tls://")
        .trim_start_matches("ws://");
    let host_port = stripped.split('/').next().unwrap_or(stripped);
    let mut parts = host_port.splitn(2, ':');
    let host = parts.next().unwrap_or("localhost").to_string();
    let port: u16 = parts.next().and_then(|p| p.parse().ok()).unwrap_or(1883);
    (host, port)
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let url = env::var("KUBEMQ_MQTT_URL").unwrap_or_else(|_| "tcp://localhost:1883".to_string());
    let (host, port) = parse_host_port(&url);
    let sfx = &Uuid::new_v4().to_string()[..8];

    // Single connection: publish-before-subscribe, subscribe, publish-after.
    let mut opts = MqttOptions::new(format!("rust-store-{sfx}"), &host, port);
    opts.set_keep_alive(Duration::from_secs(30));
    let (client, mut eventloop) = AsyncClient::new(opts, 10);

    let (msg_tx, msg_rx) = oneshot::channel::<String>();
    tokio::spawn(async move {
        let mut msg = Some(msg_tx);
        loop {
            match eventloop.poll().await {
                Ok(Event::Incoming(Packet::Publish(p))) => {
                    if let Some(tx) = msg.take() {
                        let _ = tx.send(String::from_utf8_lossy(&p.payload).to_string());
                    }
                    return;
                }
                Ok(_) => {}
                Err(e) => { eprintln!("event loop: {e}"); return; }
            }
        }
    });

    // 1. PUBLISH before subscribing — stored but never replayed over MQTT.
    client.publish(TOPIC, QoS::AtLeastOnce, false, b"pre-subscribe".as_ref()).await?;
    tokio::time::sleep(Duration::from_millis(400)).await;

    // 2. SUBSCRIBE — the connector registers a StartNewOnly subscription.
    client.subscribe(TOPIC, QoS::AtLeastOnce).await?;
    tokio::time::sleep(Duration::from_millis(700)).await; // let it go live

    // 3. PUBLISH after subscribing — only this message is delivered.
    client.publish(TOPIC, QoS::AtLeastOnce, false, b"post-subscribe".as_ref()).await?;

    let body = timeout(Duration::from_secs(10), msg_rx).await??;
    println!("received: {body} (StartNewOnly — the pre-subscribe message was not replayed)");
    Ok(())
}