Events

package main

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

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

const destination = "/topic/demo" // Events pattern → channel demo

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. SUBSCRIBE FIRST — Events have no replay.
	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)
	if err != nil {
		log.Fatalf("subscribe: %v", err)
	}
	time.Sleep(300 * time.Millisecond) // let the subscription register

	// 2. PUBLISH — SEND one event with a receipt (= KubeMQ accepted the SEND).
	pub, err := stomp.Dial(network, host)
	if err != nil {
		log.Fatalf("dial: %v", err)
	}
	if err := pub.Send(destination, "application/json",
		[]byte(`{"message":"hello"}`), stomp.SendOpt.Receipt); err != nil {
		log.Fatalf("send: %v", err)
	}
	_ = pub.Disconnect()

	// 3. RECEIVE.
	select {
	case msg := <-subscription.C:
		fmt.Printf("received: %s (destination=%s)\n", string(msg.Body), msg.Destination)
	case <-time.After(10 * time.Second):
		log.Fatal("timed out waiting for the event")
	}
}

import os
import queue
from urllib.parse import urlparse

import stomp

DESTINATION = "/topic/demo"  # Events pattern → channel demo


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. SUBSCRIBE FIRST — Events have no replay.
    listener = Listener()
    sub = stomp.Connection12([(host, port)], heartbeats=(10000, 10000))
    sub.set_listener("l", listener)
    sub.connect(wait=True)
    sub.subscribe(DESTINATION, id="demo", ack="auto")

    # 2. PUBLISH — SEND one event with a receipt (= KubeMQ accepted the SEND).
    pub = stomp.Connection12([(host, port)], heartbeats=(10000, 10000))
    pub.connect(wait=True)
    pub.send(DESTINATION, '{"message":"hello"}', content_type="application/json")
    pub.disconnect()

    # 3. RECEIVE.
    frame = listener.inbox.get(timeout=10)
    print(f"received: {frame.body} (destination={frame.headers['destination']})")
    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/demo"; // Events pattern → channel demo

    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. SUBSCRIBE FIRST — Events have no replay.
        BlockingQueue<String> inbox = new ArrayBlockingQueue<>(1);
        StompSession sub = client.connectAsync(new StompSessionHandlerAdapter() {}).get(10, TimeUnit.SECONDS);
        StompHeaders subHeaders = new StompHeaders();
        subHeaders.setDestination(DESTINATION);
        subHeaders.setId("demo");
        subHeaders.setAck("auto");
        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());
            }
        });
        Thread.sleep(300); // let the subscription register

        // 2. PUBLISH — SEND one event.
        StompSession pub = client.connectAsync(new StompSessionHandlerAdapter() {}).get(10, TimeUnit.SECONDS);
        StompHeaders pubHeaders = new StompHeaders();
        pubHeaders.setDestination(DESTINATION);
        pubHeaders.add("content-type", "application/json");
        pub.send(pubHeaders, "{\"message\":\"hello\"}".getBytes());
        Thread.sleep(300);
        pub.disconnect();

        // 3. RECEIVE.
        String body = inbox.poll(10, TimeUnit.SECONDS);
        if (body == null) throw new IllegalStateException("timed out waiting for the event");
        System.out.printf("received: %s%n", body);
        sub.disconnect();
        client.stop();
    }
}

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

const DESTINATION = "/topic/demo"; // Events pattern → channel demo

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. SUBSCRIBE FIRST — Events have no replay.
  const sub = await open();
  const received = new Promise<string>((resolve, reject) => {
    sub.subscribe({ destination: DESTINATION, ack: "auto" }, (err, message) => {
      if (err) return reject(err);
      message.readString("utf-8", (readErr, body) => (readErr ? reject(readErr) : resolve(body ?? "")));
    });
  });
  await new Promise<void>((r) => setTimeout(r, 300)); // let the subscription register

  // 2. PUBLISH — SEND one event.
  const pub = await open();
  const frame = pub.send({ destination: DESTINATION, "content-type": "application/json" });
  frame.write(JSON.stringify({ message: "hello" }));
  frame.end();
  await new Promise<void>((r) => pub.disconnect(() => r()));

  // 3. RECEIVE.
  console.log("received:", await received);
  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/demo"; // Events pattern → channel demo

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

// 1. SUBSCRIBE FIRST — Events have no replay.
using var subConn = factory.CreateConnection();
subConn.Start();
using var subSession = subConn.CreateSession(AcknowledgementMode.IndividualAcknowledge);
using var consumer = subSession.CreateConsumer(subSession.GetTopic(destination));
await Task.Delay(300); // let the subscription register

// 2. PUBLISH — SEND one event.
using (var pubConn = factory.CreateConnection())
{
    pubConn.Start();
    using var pubSession = pubConn.CreateSession(AcknowledgementMode.AutoAcknowledge);
    using var producer = pubSession.CreateProducer(pubSession.GetTopic(destination));
    var msg = producer.CreateBytesMessage(Encoding.UTF8.GetBytes("{\"message\":\"hello\"}"));
    msg.StompType = "application/json";
    producer.Send(msg);
}

// 3. RECEIVE.
var received = consumer.Receive(TimeSpan.FromSeconds(10))
    ?? throw new InvalidOperationException("timed out waiting for the event");
Console.WriteLine($"received: {Encoding.UTF8.GetString(received.Content)} " +
                  $"(channel={received.StompDestination?.PhysicalName})");

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/demo" # Events pattern → channel demo

# 1. SUBSCRIBE FIRST — Events have no replay.
inbox = Thread::Queue.new
sub = Stomp::Client.new(hosts: hosts, connect_headers: { "accept-version" => "1.2", "host" => "/" })
sub.subscribe(DESTINATION, id: "demo", ack: "auto") { |msg| inbox << msg }
sleep 0.3 # let the subscription register

# 2. PUBLISH — SEND one event.
pub = Stomp::Client.new(hosts: hosts, connect_headers: { "accept-version" => "1.2", "host" => "/" })
pub.publish(DESTINATION, '{"message":"hello"}', "content-type" => "application/json")
pub.close

# 3. RECEIVE.
msg = Timeout.timeout(10) { inbox.pop }
puts "received: #{msg.body} (destination=#{msg.headers['destination']})"
sub.close

use std::time::Duration;

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

const DESTINATION: &str = "/topic/demo"; // Events pattern → channel demo

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. SUBSCRIBE FIRST — Events have no replay.
    let mut sub = Connector::builder()
        .server(format!("{host}:{port}"))
        .virtualhost(&host)
        .connect()
        .await?;
    sub.send(ToServer::Subscribe {
        destination: DESTINATION.into(),
        id: "demo".into(),
        ack: Some(AckMode::Auto),
    }.into())
    .await?;
    tokio::time::sleep(Duration::from_millis(300)).await; // let the subscription register

    // 2. PUBLISH — SEND one event.
    let mut publisher = Connector::builder()
        .server(format!("{host}:{port}"))
        .virtualhost(&host)
        .connect()
        .await?;
    publisher
        .send(ToServer::Send {
            destination: DESTINATION.into(),
            transaction: None,
            headers: Some(vec![("content-type".into(), "application/json".into())]),
            body: Some(br#"{"message":"hello"}"#.to_vec()),
        }.into())
        .await?;
    publisher.send(ToServer::Disconnect { receipt: None }.into()).await?;

    // 3. RECEIVE.
    let frame = tokio::time::timeout(Duration::from_secs(10), sub.next())
        .await?
        .ok_or("stream closed")??;
    if let FromServer::Message { body, destination, .. } = frame.content {
        let payload = String::from_utf8_lossy(&body.unwrap_or_default()).into_owned();
        println!("received: {payload} (destination={destination})");
    }
    Ok(())
}