Getting Started

package main

import (
	"fmt"
	"log"
	"os"
	"time"

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

func stompAddr() string {
	url := os.Getenv("KUBEMQ_STOMP_URL")
	if url == "" {
		url = "tcp://localhost:61613"
	}
	return url[len("tcp://"):]
}

func dial(login string) (*stomp.Conn, error) {
	return stomp.Dial("tcp", stompAddr(),
		stomp.ConnOpt.AcceptVersion(stomp.V12),
		stomp.ConnOpt.Login(login, ""))
}

func main() {
	const destination = "/topic/demo" // Events pattern, channel "demo"

	// SUBSCRIBER — connect and subscribe before publishing (Events is at-most-once).
	sub, err := dial("go-sub")
	if err != nil {
		log.Fatalf("sub connect: %v", err)
	}
	subscription, err := sub.Subscribe(destination, stomp.AckAuto)
	if err != nil {
		log.Fatalf("subscribe: %v", err)
	}
	fmt.Printf("[sub] subscribed to %s\n", destination)
	time.Sleep(300 * time.Millisecond) // let the subscription register

	// PUBLISHER — connect and SEND one message.
	pub, err := dial("go-pub")
	if err != nil {
		log.Fatalf("pub connect: %v", err)
	}
	if err := pub.Send(destination, "application/json",
		[]byte(`{"message":"hello"}`), stomp.SendOpt.Receipt); err != nil {
		log.Fatalf("send: %v", err)
	}
	fmt.Printf("[pub] sent to %s\n", destination)

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

	_ = pub.Disconnect()
	_ = sub.Disconnect()
}

import os
import queue
import time

import stomp


def endpoint() -> tuple[str, int]:
    url = os.environ.get("KUBEMQ_STOMP_URL", "tcp://localhost:61613")
    host, port = url.split("://", 1)[1].split(":", 1)
    return host, int(port)


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

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


def connect(login: str, listener: stomp.ConnectionListener | None = None) -> stomp.Connection:
    conn = stomp.Connection([endpoint()], heartbeats=(10000, 10000))
    if listener is not None:
        conn.set_listener("", listener)
    conn.connect(login=login, passcode="", wait=True)
    return conn


def main() -> None:
    destination = "/topic/demo"  # Events pattern, channel "demo"

    # SUBSCRIBER — connect and subscribe first.
    listener = Listener()
    sub = connect("py-sub", listener)
    sub.subscribe(destination=destination, id="sub-1", ack="auto")
    print(f"[sub] subscribed to {destination}")
    time.sleep(0.3)

    # PUBLISHER — connect and SEND.
    pub = connect("py-pub")
    pub.send(destination=destination, body='{"message":"hello"}',
             content_type="application/json")
    print(f"[pub] sent to {destination}")

    frame = listener.messages.get(timeout=10)
    body = frame.body if isinstance(frame.body, str) else frame.body.decode()
    print(f"[sub] received: {body} (destination={frame.headers['destination']})")

    pub.disconnect()
    sub.disconnect()


if __name__ == "__main__":
    main()

import java.net.URI;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.lang.reflect.Type;

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 {
    public static void main(String[] args) throws Exception {
        String url = System.getenv().getOrDefault("KUBEMQ_STOMP_URL", "tcp://localhost:61613");
        URI uri = URI.create(url);
        String destination = "/topic/demo"; // Events pattern, channel "demo"

        ReactorNettyTcpStompClient client =
                new ReactorNettyTcpStompClient(uri.getHost(), uri.getPort());
        LinkedBlockingQueue<String> inbox = new LinkedBlockingQueue<>();

        // SUBSCRIBER
        StompSession sub = client.connect(new StompSessionHandlerAdapter() {}).get();
        sub.subscribe(destination, new StompSessionHandlerAdapter() {
            @Override public Type getPayloadType(StompHeaders headers) { return byte[].class; }
            @Override public void handleFrame(StompHeaders headers, Object payload) {
                inbox.add(new String((byte[]) payload));
            }
        });
        System.out.printf("[sub] subscribed to %s%n", destination);
        Thread.sleep(300);

        // PUBLISHER
        StompSession pub = client.connect(new StompSessionHandlerAdapter() {}).get();
        StompHeaders headers = new StompHeaders();
        headers.setDestination(destination);
        headers.add("content-type", "application/json");
        pub.send(headers, "{\"message\":\"hello\"}".getBytes());
        System.out.printf("[pub] sent to %s%n", destination);

        String body = inbox.poll(10, TimeUnit.SECONDS);
        System.out.printf("[sub] received: %s%n", body);

        pub.disconnect();
        sub.disconnect();
        client.shutdown();
    }
}

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

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 dial(login: string): Promise<Client> {
  const { host, port } = endpoint();
  return new Promise((resolve, reject) => {
    connect({ host, port, connectHeaders: { "accept-version": "1.2",
      "heart-beat": "10000,10000", login } },
      (err, c) => (err ? reject(err) : resolve(c)));
  });
}

async function main(): Promise<void> {
  const destination = "/topic/demo"; // Events pattern, channel "demo"

  // SUBSCRIBER
  const sub = await dial("js-sub");
  const received = new Promise<string>((resolve, reject) => {
    sub.subscribe({ destination, ack: "auto" }, (err, message) => {
      if (err) return reject(err);
      message.readString("utf-8", (e, body) => (e ? reject(e) : resolve(body ?? "")));
    });
  });
  console.log(`[sub] subscribed to ${destination}`);
  await new Promise((r) => setTimeout(r, 300));

  // PUBLISHER
  const pub = await dial("js-pub");
  const frame = pub.send({ destination, "content-type": "application/json" });
  frame.write('{"message":"hello"}');
  frame.end();
  console.log(`[pub] sent to ${destination}`);

  console.log(`[sub] received: ${await received}`);
  pub.disconnect();
  sub.disconnect();
}

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

using System.Text;
using Apache.NMS;
using Stomp.Net;

static string BrokerUri()
{
    var url = Environment.GetEnvironmentVariable("KUBEMQ_STOMP_URL") ?? "tcp://localhost:61613";
    var u = new Uri(url);
    var transport = u.Scheme == "tls" ? "ssl" : "tcp";
    return $"{transport}://{u.Host}:{(u.Port > 0 ? u.Port : 61613)}";
}

const string destination = "/topic/demo"; // Events pattern, channel "demo"

var factory = new ConnectionFactory(BrokerUri(), new StompConnectionSettings());
var received = new TaskCompletionSource<string>(TaskCreationOptions.RunContinuationsAsynchronously);

// SUBSCRIBER
using var subConn = factory.CreateConnection();
subConn.Start();
using var subSession = subConn.CreateSession(AcknowledgementMode.AutoAcknowledge);
using var consumer = subSession.CreateConsumer(subSession.GetTopic(destination));
consumer.Listener += msg =>
    received.TrySetResult(Encoding.UTF8.GetString(((IBytesMessage)msg).Content));
Console.WriteLine($"[sub] subscribed to {destination}");
await Task.Delay(300);

// PUBLISHER
using var pubConn = factory.CreateConnection();
pubConn.Start();
using var pubSession = pubConn.CreateSession(AcknowledgementMode.AutoAcknowledge);
using var producer = pubSession.CreateProducer(pubSession.GetTopic(destination));
var message = pubSession.CreateBytesMessage(Encoding.UTF8.GetBytes("{\"message\":\"hello\"}"));
message.StompType = "application/json";
producer.Send(message);
Console.WriteLine($"[pub] sent to {destination}");

using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10));
Console.WriteLine($"[sub] received: {await received.Task.WaitAsync(cts.Token)}");

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

uri = URI.parse(ENV.fetch("KUBEMQ_STOMP_URL", "tcp://localhost:61613"))
destination = "/topic/demo" # Events pattern, channel "demo"

def connect(uri, login)
  Stomp::Client.new(
    hosts: [{ host: uri.host, port: uri.port }],
    connect_headers: { "accept-version" => "1.2", "heart-beat" => "10000,10000",
                       "login" => login, "passcode" => "" },
  )
end

inbox = Queue.new

# SUBSCRIBER
sub = connect(uri, "rb-sub")
sub.subscribe(destination, id: "sub-1", ack: "auto") { |msg| inbox << msg.body }
puts "[sub] subscribed to #{destination}"
sleep 0.3

# PUBLISHER
pub = connect(uri, "rb-pub")
pub.publish(destination, '{"message":"hello"}', { "content-type" => "application/json" })
puts "[pub] sent to #{destination}"

body = Timeout.timeout(10) { inbox.pop }
puts "[sub] received: #{body}"

pub.close
sub.close

use async_stomp::client::Connector;
use async_stomp::{FromServer, ToServer};
use futures::{SinkExt, StreamExt};
use std::time::Duration;
use tokio::time::timeout;

async fn dial(host_port: &str, login: &str) -> Result<
    impl SinkExt<async_stomp::Message<ToServer>> + StreamExt, Box<dyn std::error::Error>> {
    Ok(Connector::builder()
        .server(host_port)
        .login(login.to_string())
        .passcode(String::new())
        .connect()
        .await?)
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let url = std::env::var("KUBEMQ_STOMP_URL").unwrap_or_else(|_| "tcp://localhost:61613".into());
    let host_port = url.split("://").nth(1).unwrap_or("localhost:61613").to_string();
    let destination = "/topic/demo"; // Events pattern, channel "demo"

    // SUBSCRIBER
    let mut sub = Connector::builder().server(&host_port)
        .login("rust-sub".into()).passcode(String::new()).connect().await?;
    sub.send(ToServer::Subscribe {
        destination: destination.to_string(),
        id: "sub-1".to_string(),
        ack: None,
    }).await?;
    println!("[sub] subscribed to {destination}");
    tokio::time::sleep(Duration::from_millis(300)).await;

    // PUBLISHER
    let mut pubc = Connector::builder().server(&host_port)
        .login("rust-pub".into()).passcode(String::new()).connect().await?;
    pubc.send(ToServer::Send {
        destination: destination.to_string(),
        transaction: None,
        headers: Some(vec![("content-type".into(), "application/json".into())]),
        body: Some(br#"{"message":"hello"}"#.to_vec()),
    }).await?;
    println!("[pub] sent to {destination}");

    // RECEIVE
    if let Ok(Some(Ok(msg))) = timeout(Duration::from_secs(10), sub.next()).await {
        if let FromServer::Message { body, .. } = msg.content {
            println!("[sub] received: {}", String::from_utf8_lossy(&body.unwrap_or_default()));
        }
    }

    pubc.send(ToServer::Disconnect { receipt: None }).await?;
    Ok(())
}