Queries

package main

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

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

const (
	replyDest = "/reply/q1"     // connection-local reply inbox
	queryDest = "/query/lookup" // Queries pattern → channel lookup
)

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/q1 FIRST (connection-local).
	reply, err := conn.Subscribe(replyDest, stomp.AckAuto)
	if err != nil {
		log.Fatalf("subscribe reply: %v", err)
	}

	// Step 2: SEND /query/lookup with reply-to + correlation-id + timeout(ms).
	if err := conn.Send(queryDest, "application/json", []byte(`{"key":"user-42"}`),
		stomp.SendOpt.Header("reply-to", replyDest),
		stomp.SendOpt.Header("correlation-id", "abc123"),
		stomp.SendOpt.Header("timeout", "5000"),
	); err != nil {
		log.Fatalf("send query: %v", err)
	}

	// Step 3: receive the response body on /reply/q1.
	select {
	case msg := <-reply.C:
		if se := msg.Header.Get("stomp-error"); se != "" {
			log.Fatalf("query failed: stomp-error=%q (connection stays open)", se)
		}
		fmt.Printf("query answered: %s (content-type=%s)\n",
			string(msg.Body), msg.Header.Get("content-type"))
	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/q1"     # connection-local reply inbox
QUERY_DEST = "/query/lookup"  # Queries pattern → channel lookup


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/q1 FIRST (connection-local).
    conn.subscribe(REPLY_DEST, id="q1", ack="auto")

    # Step 2: SEND /query/lookup with reply-to + correlation-id + timeout(ms).
    conn.send(QUERY_DEST, '{"key":"user-42"}', content_type="application/json",
              headers={"reply-to": REPLY_DEST, "correlation-id": "abc123", "timeout": "5000"})

    # Step 3: receive the response body on /reply/q1.
    frame = replies.inbox.get(timeout=10)
    if frame.headers.get("stomp-error"):
        raise SystemExit(f"query failed: {frame.headers['stomp-error']!r}")
    print(f"query answered: {frame.body} (content-type={frame.headers.get('content-type')})")
    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/q1";    // connection-local reply inbox
    private static final String QUERY_DEST = "/query/lookup"; // Queries pattern → channel lookup

    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<Object[]> inbox = new ArrayBlockingQueue<>(1);
        StompSession conn = client.connectAsync(new StompSessionHandlerAdapter() {}).get(10, TimeUnit.SECONDS);

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

        // Step 2: SEND /query/lookup with reply-to + correlation-id + timeout(ms).
        StompHeaders q = new StompHeaders();
        q.setDestination(QUERY_DEST);
        q.add("content-type", "application/json");
        q.add("reply-to", REPLY_DEST);
        q.add("correlation-id", "abc123");
        q.add("timeout", "5000");
        conn.send(q, "{\"key\":\"user-42\"}".getBytes());

        // Step 3: receive the response body on /reply/q1.
        Object[] reply = inbox.poll(10, TimeUnit.SECONDS);
        if (reply == null) throw new IllegalStateException("timed out waiting for the reply");
        StompHeaders headers = (StompHeaders) reply[0];
        if (headers.getFirst("stomp-error") != null) {
            throw new IllegalStateException("query failed: " + headers.getFirst("stomp-error"));
        }
        System.out.printf("query answered: %s (content-type=%s)%n",
                reply[1], headers.getFirst("content-type"));
        conn.disconnect();
        client.stop();
    }
}

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

const REPLY_DEST = "/reply/q1";     // connection-local reply inbox
const QUERY_DEST = "/query/lookup"; // Queries pattern → channel lookup

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/q1 FIRST (connection-local).
  const reply = new Promise<{ headers: Record<string, string>; body: 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, body) => {
        clearTimeout(timer);
        if (readErr) return reject(readErr);
        resolve({ headers: message.headers as Record<string, string>, body: body ?? "" });
      });
    });
  });

  // Step 2: SEND /query/lookup with reply-to + correlation-id + timeout(ms).
  const frame = conn.send({
    destination: QUERY_DEST,
    "content-type": "application/json",
    "reply-to": REPLY_DEST,
    "correlation-id": "abc123",
    timeout: "5000",
  });
  frame.write(JSON.stringify({ key: "user-42" }));
  frame.end();

  // Step 3: inspect the response body on /reply/q1.
  const { headers, body } = await reply;
  if (headers["stomp-error"]) throw new Error(`query failed: ${headers["stomp-error"]}`);
  console.log(`query answered: ${body} (content-type=${headers["content-type"]})`);
  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/q1";     // connection-local reply inbox
const string queryDest = "/query/lookup"; // Queries pattern → channel lookup

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/q1 FIRST (connection-local).
using var replyConsumer = session.CreateConsumer(session.GetQueue(replyDest));

// Step 2: SEND /query/lookup with reply-to + correlation-id + timeout(ms).
using var producer = session.CreateProducer(session.GetQueue(queryDest));
var q = producer.CreateBytesMessage(Encoding.UTF8.GetBytes("{\"key\":\"user-42\"}"));
q.StompType = "application/json";
q.Headers.SetValue("reply-to", replyDest);
q.Headers.SetValue("correlation-id", "abc123");
q.Headers.SetValue("timeout", "5000");
producer.Send(q);

// Step 3: receive the response body on /reply/q1.
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($"query failed: {stompError}");
Console.WriteLine($"query answered: {Encoding.UTF8.GetString(reply.Content)} " +
                  $"(content-type={reply.Headers.GetValue("content-type")})");

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/q1"     # connection-local reply inbox
QUERY_DEST = "/query/lookup" # Queries pattern → channel lookup

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

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

# Step 2: SEND /query/lookup with reply-to + correlation-id + timeout(ms).
conn.publish(QUERY_DEST, '{"key":"user-42"}',
             "content-type" => "application/json",
             "reply-to" => REPLY_DEST,
             "correlation-id" => "abc123",
             "timeout" => "5000")

# Step 3: receive the response body on /reply/q1.
reply = Timeout.timeout(10) { inbox.pop }
raise "query failed: #{reply.headers['stomp-error']}" if reply.headers["stomp-error"]

puts "query answered: #{reply.body} (content-type=#{reply.headers['content-type']})"
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/q1";     // connection-local reply inbox
const QUERY_DEST: &str = "/query/lookup"; // Queries pattern → channel lookup

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/q1 FIRST (connection-local).
    conn.send(ToServer::Subscribe {
        destination: REPLY_DEST.into(),
        id: "q1".into(),
        ack: Some(AckMode::Auto),
    }.into())
    .await?;

    // Step 2: SEND /query/lookup with reply-to + correlation-id + timeout(ms).
    conn.send(ToServer::Send {
        destination: QUERY_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#"{"key":"user-42"}"#.to_vec()),
    }.into())
    .await?;

    // Step 3: receive the response body on /reply/q1.
    let frame = tokio::time::timeout(Duration::from_secs(10), conn.next())
        .await?
        .ok_or("stream closed")??;
    if let FromServer::Message { headers, body, .. } = frame.content {
        if let Some((_, err)) = headers.iter().find(|(k, _)| k == "stomp-error") {
            return Err(format!("query failed: {err}").into());
        }
        let payload = String::from_utf8_lossy(&body.unwrap_or_default()).into_owned();
        println!("query answered: {payload}");
    }
    Ok(())
}