Commands

package main

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

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

const (
	replyDest   = "/reply/r1"     // connection-local reply inbox
	commandDest = "/command/exec" // Commands pattern → channel exec
)

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()
	conn, err := stomp.Dial(network, host)
	if err != nil {
		log.Fatalf("dial: %v", err)
	}
	defer func() { _ = conn.Disconnect() }()

	// Step 1: SUBSCRIBE /reply/r1 FIRST (connection-local).
	reply, err := conn.Subscribe(replyDest, stomp.AckAuto)
	if err != nil {
		log.Fatalf("subscribe reply: %v", err)
	}

	// Step 2: SEND /command/exec with reply-to + correlation-id + timeout(ms).
	corrID := "abc123"
	if err := conn.Send(commandDest, "application/json", []byte(`{"op":"do-work"}`),
		stomp.SendOpt.Header("reply-to", replyDest),
		stomp.SendOpt.Header("correlation-id", corrID),
		stomp.SendOpt.Header("timeout", "5000"),
	); err != nil {
		log.Fatalf("send command: %v", err)
	}

	// Step 3: receive the execution result on /reply/r1.
	select {
	case msg := <-reply.C:
		if se := msg.Header.Get("stomp-error"); se != "" {
			log.Fatalf("command failed: stomp-error=%q (connection stays open)", se)
		}
		fmt.Printf("command executed (correlation-id=%s)\n", msg.Header.Get("correlation-id"))
	case <-time.After(10 * time.Second):
		log.Fatal("timed out waiting for the reply")
	}
}

import os
import queue
from urllib.parse import urlparse

import stomp

REPLY_DEST = "/reply/r1"      # connection-local reply inbox
COMMAND_DEST = "/command/exec"  # Commands pattern → channel exec


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 Replies(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()
    replies = Replies()
    conn = stomp.Connection12([(host, port)], heartbeats=(10000, 10000))
    conn.set_listener("r", replies)
    conn.connect(wait=True)

    # Step 1: SUBSCRIBE /reply/r1 FIRST (connection-local).
    conn.subscribe(REPLY_DEST, id="r1", ack="auto")

    # Step 2: SEND /command/exec with reply-to + correlation-id + timeout(ms).
    conn.send(COMMAND_DEST, '{"op":"do-work"}', content_type="application/json",
              headers={"reply-to": REPLY_DEST, "correlation-id": "abc123", "timeout": "5000"})

    # Step 3: receive the execution result on /reply/r1.
    frame = replies.inbox.get(timeout=10)
    if frame.headers.get("stomp-error"):
        raise SystemExit(f"command failed: {frame.headers['stomp-error']!r}")
    print(f"command executed (correlation-id={frame.headers.get('correlation-id')})")
    conn.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 REPLY_DEST = "/reply/r1";       // connection-local reply inbox
    private static final String COMMAND_DEST = "/command/exec"; // Commands pattern → channel exec

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

        BlockingQueue<StompHeaders> inbox = new ArrayBlockingQueue<>(1);
        StompSession conn = client.connectAsync(new StompSessionHandlerAdapter() {}).get(10, TimeUnit.SECONDS);

        // Step 1: SUBSCRIBE /reply/r1 FIRST (connection-local).
        StompHeaders replyHeaders = new StompHeaders();
        replyHeaders.setDestination(REPLY_DEST);
        replyHeaders.setId("r1");
        replyHeaders.setAck("auto");
        conn.subscribe(replyHeaders, new StompSessionHandlerAdapter() {
            @Override public Type getPayloadType(StompHeaders headers) { return byte[].class; }
            @Override public void handleFrame(StompHeaders headers, Object payload) { inbox.add(headers); }
        });

        // Step 2: SEND /command/exec with reply-to + correlation-id + timeout(ms).
        StompHeaders cmd = new StompHeaders();
        cmd.setDestination(COMMAND_DEST);
        cmd.add("content-type", "application/json");
        cmd.add("reply-to", REPLY_DEST);
        cmd.add("correlation-id", "abc123");
        cmd.add("timeout", "5000");
        conn.send(cmd, "{\"op\":\"do-work\"}".getBytes());

        // Step 3: receive the execution result on /reply/r1.
        StompHeaders reply = inbox.poll(10, TimeUnit.SECONDS);
        if (reply == null) throw new IllegalStateException("timed out waiting for the reply");
        if (reply.getFirst("stomp-error") != null) {
            throw new IllegalStateException("command failed: " + reply.getFirst("stomp-error"));
        }
        System.out.printf("command executed (correlation-id=%s)%n", reply.getFirst("correlation-id"));
        conn.disconnect();
        client.stop();
    }
}

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

const REPLY_DEST = "/reply/r1";       // connection-local reply inbox
const COMMAND_DEST = "/command/exec"; // Commands pattern → channel exec

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> {
  const conn = await open();

  // Step 1: SUBSCRIBE /reply/r1 FIRST (connection-local).
  const reply = new Promise<Record<string, string>>((resolve, reject) => {
    const timer = setTimeout(() => reject(new Error("timed out waiting for the reply")), 15_000);
    conn.subscribe({ destination: REPLY_DEST, ack: "auto" }, (err, message) => {
      if (err) return reject(err);
      message.readString("utf-8", (readErr) => {
        clearTimeout(timer);
        if (readErr) return reject(readErr);
        resolve(message.headers as Record<string, string>);
      });
    });
  });

  // Step 2: SEND /command/exec with reply-to + correlation-id + timeout(ms).
  const frame = conn.send({
    destination: COMMAND_DEST,
    "content-type": "application/json",
    "reply-to": REPLY_DEST,
    "correlation-id": "abc123",
    timeout: "5000",
  });
  frame.write(JSON.stringify({ op: "do-work" }));
  frame.end();

  // Step 3: inspect the execution result on /reply/r1.
  const headers = await reply;
  if (headers["stomp-error"]) throw new Error(`command failed: ${headers["stomp-error"]}`);
  console.log(`command executed (correlation-id=${headers["correlation-id"]})`);
  await new Promise<void>((r) => conn.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 replyDest = "/reply/r1";       // connection-local reply inbox
const string commandDest = "/command/exec"; // Commands pattern → channel exec

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

using var conn = factory.CreateConnection();
conn.Start();
using var session = conn.CreateSession(AcknowledgementMode.AutoAcknowledge);

// Step 1: SUBSCRIBE /reply/r1 FIRST (connection-local).
using var replyConsumer = session.CreateConsumer(session.GetQueue(replyDest));

// Step 2: SEND /command/exec with reply-to + correlation-id + timeout(ms).
using var producer = session.CreateProducer(session.GetQueue(commandDest));
var cmd = producer.CreateBytesMessage(Encoding.UTF8.GetBytes("{\"op\":\"do-work\"}"));
cmd.StompType = "application/json";
cmd.Headers.SetValue("reply-to", replyDest);
cmd.Headers.SetValue("correlation-id", "abc123");
cmd.Headers.SetValue("timeout", "5000");
producer.Send(cmd);

// Step 3: receive the execution result on /reply/r1.
var reply = replyConsumer.Receive(TimeSpan.FromSeconds(10))
    ?? throw new InvalidOperationException("timed out waiting for the reply");
var stompError = reply.Headers.GetValue("stomp-error");
if (!string.IsNullOrEmpty(stompError))
    throw new InvalidOperationException($"command failed: {stompError}");
Console.WriteLine($"command executed (correlation-id={reply.Headers.GetValue("correlation-id")})");

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: "" }]
REPLY_DEST = "/reply/r1"       # connection-local reply inbox
COMMAND_DEST = "/command/exec" # Commands pattern → channel exec

conn = Stomp::Client.new(hosts: hosts, connect_headers: { "accept-version" => "1.2", "host" => "/" })

# Step 1: SUBSCRIBE /reply/r1 FIRST (connection-local).
inbox = Thread::Queue.new
conn.subscribe(REPLY_DEST, id: "r1", ack: "auto") { |msg| inbox << msg }

# Step 2: SEND /command/exec with reply-to + correlation-id + timeout(ms).
conn.publish(COMMAND_DEST, '{"op":"do-work"}',
             "content-type" => "application/json",
             "reply-to" => REPLY_DEST,
             "correlation-id" => "abc123",
             "timeout" => "5000")

# Step 3: receive the execution result on /reply/r1.
reply = Timeout.timeout(10) { inbox.pop }
raise "command failed: #{reply.headers['stomp-error']}" if reply.headers["stomp-error"]

puts "command executed (correlation-id=#{reply.headers['correlation-id']})"
conn.close

use std::time::Duration;

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

const REPLY_DEST: &str = "/reply/r1";       // connection-local reply inbox
const COMMAND_DEST: &str = "/command/exec"; // Commands pattern → channel exec

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();
    let mut conn = Connector::builder()
        .server(format!("{host}:{port}"))
        .virtualhost(&host)
        .connect()
        .await?;

    // Step 1: SUBSCRIBE /reply/r1 FIRST (connection-local).
    conn.send(ToServer::Subscribe {
        destination: REPLY_DEST.into(),
        id: "r1".into(),
        ack: Some(AckMode::Auto),
    }.into())
    .await?;

    // Step 2: SEND /command/exec with reply-to + correlation-id + timeout(ms).
    conn.send(ToServer::Send {
        destination: COMMAND_DEST.into(),
        transaction: None,
        headers: Some(vec![
            ("content-type".into(), "application/json".into()),
            ("reply-to".into(), REPLY_DEST.into()),
            ("correlation-id".into(), "abc123".into()),
            ("timeout".into(), "5000".into()),
        ]),
        body: Some(br#"{"op":"do-work"}"#.to_vec()),
    }.into())
    .await?;

    // Step 3: receive the execution result on /reply/r1.
    let frame = tokio::time::timeout(Duration::from_secs(10), conn.next())
        .await?
        .ok_or("stream closed")??;
    if let FromServer::Message { headers, .. } = frame.content {
        if let Some((_, err)) = headers.iter().find(|(k, _)| k == "stomp-error") {
            return Err(format!("command failed: {err}").into());
        }
        let corr = headers.iter().find(|(k, _)| k == "correlation-id").map(|(_, v)| v.as_str());
        println!("command executed (correlation-id={})", corr.unwrap_or(""));
    }
    Ok(())
}