Queues

package main

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

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

const destination = "/queue/jobs/email" // Queues pattern → channel jobs.email
const count = 5

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. Produce — SEND `count` jobs to the queue.
	pub, err := stomp.Dial(network, host)
	if err != nil {
		log.Fatalf("dial: %v", err)
	}
	for i := 1; i <= count; i++ {
		body := fmt.Sprintf("job-%d", i)
		if err := pub.Send(destination, "text/plain", []byte(body)); err != nil {
			log.Fatalf("send: %v", err)
		}
	}
	_ = pub.Disconnect()

	// 2. Consume — SUBSCRIBE ack:client-individual; ACK each by its 1.2 token.
	sub, err := stomp.Dial(network, host)
	if err != nil {
		log.Fatalf("dial: %v", err)
	}
	defer func() { _ = sub.Disconnect() }()

	subscription, err := sub.Subscribe(destination, stomp.AckClientIndividual)
	if err != nil {
		log.Fatalf("subscribe: %v", err)
	}

	for got := 0; got < count; {
		select {
		case msg := <-subscription.C:
			if msg.Err != nil {
				log.Fatalf("receive: %v", msg.Err)
			}
			if msg.Header.Get("redelivered") == "true" {
				log.Fatalf("unexpected redelivery of %q", string(msg.Body))
			}
			if err := sub.Ack(msg); err != nil { // ACK by the 1.2 ack token
				log.Fatalf("ack: %v", err)
			}
			got++
		case <-time.After(10 * time.Second):
			log.Fatal("timed out draining the queue")
		}
	}
	fmt.Printf("drained and acked %d jobs\n", count)
}

import os
import queue
from urllib.parse import urlparse

import stomp

DESTINATION = "/queue/jobs/email"  # Queues pattern → channel jobs.email
COUNT = 5


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 Consumer(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. Produce — SEND `count` jobs to the queue.
    pub = stomp.Connection12([(host, port)], heartbeats=(10000, 10000))
    pub.connect(wait=True)
    for i in range(1, COUNT + 1):
        pub.send(DESTINATION, f"job-{i}", content_type="text/plain")
    pub.disconnect()

    # 2. Consume — SUBSCRIBE ack:client-individual; ACK each by its 1.2 token.
    consumer = Consumer()
    sub = stomp.Connection12([(host, port)], heartbeats=(10000, 10000))
    sub.set_listener("c", consumer)
    sub.connect(wait=True)
    sub.subscribe(DESTINATION, id="jobs", ack="client-individual")

    for _ in range(COUNT):
        frame = consumer.inbox.get(timeout=10)
        if frame.headers.get("redelivered") == "true":
            raise SystemExit(f"unexpected redelivery: {frame.body!r}")
        sub.ack(frame.headers["ack"])  # ACK by the 1.2 ack token
    sub.disconnect()
    print(f"drained and acked {COUNT} jobs")


if __name__ == "__main__":
    main()

import java.lang.reflect.Type;
import java.nio.charset.StandardCharsets;
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.tcp.reactor.ReactorNettyTcpClient;
import org.springframework.messaging.simp.stomp.ReactorNettyTcpStompClient;

public final class Main {
    private static final String DESTINATION = "/queue/jobs/email"; // → channel jobs.email
    private static final int COUNT = 5;

    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. Produce — SEND `count` jobs to the queue.
        StompSession pub = client.connectAsync(new StompSessionHandlerAdapter() {}).get(10, TimeUnit.SECONDS);
        for (int i = 1; i <= COUNT; i++) {
            StompHeaders h = new StompHeaders();
            h.setDestination(DESTINATION);
            h.add("content-type", "text/plain");
            pub.send(h, ("job-" + i).getBytes(StandardCharsets.UTF_8));
        }
        Thread.sleep(500);
        pub.disconnect();

        // 2. Consume — SUBSCRIBE ack:client-individual; ACK each delivery.
        BlockingQueue<StompHeaders> inbox = new ArrayBlockingQueue<>(COUNT);
        StompSession sub = client.connectAsync(new StompSessionHandlerAdapter() {}).get(10, TimeUnit.SECONDS);
        StompHeaders subHeaders = new StompHeaders();
        subHeaders.setDestination(DESTINATION);
        subHeaders.setId("jobs");
        subHeaders.setAck("client-individual");
        sub.subscribe(subHeaders, new StompSessionHandlerAdapter() {
            @Override public Type getPayloadType(StompHeaders headers) { return byte[].class; }
            @Override public void handleFrame(StompHeaders headers, Object payload) { inbox.add(headers); }
        });

        for (int i = 0; i < COUNT; i++) {
            StompHeaders headers = inbox.poll(10, TimeUnit.SECONDS);
            if (headers == null) throw new IllegalStateException("timed out draining the queue");
            if ("true".equals(headers.getFirst("redelivered"))) {
                throw new IllegalStateException("unexpected redelivery");
            }
            sub.acknowledge(headers.getAck(), true); // ACK by the 1.2 ack token
        }
        sub.disconnect();
        client.stop();
        System.out.printf("drained and acked %d jobs%n", COUNT);
    }
}

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

const DESTINATION = "/queue/jobs/email"; // Queues pattern → channel jobs.email
const COUNT = 5;

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. Produce — SEND `count` jobs to the queue.
  const pub = await open();
  for (let i = 1; i <= COUNT; i++) {
    const frame = pub.send({ destination: DESTINATION, "content-type": "text/plain" });
    frame.write(`job-${i}`);
    frame.end();
  }
  await new Promise<void>((r) => pub.disconnect(() => r()));

  // 2. Consume — SUBSCRIBE ack:client-individual; ACK each delivery.
  const sub = await open();
  let got = 0;
  await new Promise<void>((resolve, reject) => {
    const timer = setTimeout(() => reject(new Error("timed out draining the queue")), 15_000);
    sub.subscribe({ destination: DESTINATION, ack: "client-individual" }, (err, message) => {
      if (err) return reject(err);
      if (message.headers["redelivered"] === "true") return reject(new Error("unexpected redelivery"));
      message.readString("utf-8", (readErr) => {
        if (readErr) return reject(readErr);
        sub.ack(message); // ACK by the 1.2 ack token (stompit tracks it on the frame)
        if (++got === COUNT) {
          clearTimeout(timer);
          resolve();
        }
      });
    });
  });
  await new Promise<void>((r) => sub.disconnect(() => r()));
  console.log(`drained and acked ${COUNT} jobs`);
}

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 = "/queue/jobs/email"; // Queues pattern → channel jobs.email
const int count = 5;

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

// 1. Produce — SEND `count` jobs to the queue.
using (var pubConn = factory.CreateConnection())
{
    pubConn.Start();
    using var session = pubConn.CreateSession(AcknowledgementMode.AutoAcknowledge);
    using var producer = session.CreateProducer(session.GetQueue(destination));
    for (var i = 1; i <= count; i++)
    {
        var msg = producer.CreateBytesMessage(Encoding.UTF8.GetBytes($"job-{i}"));
        msg.StompType = "text/plain";
        producer.Send(msg);
    }
}

// 2. Consume — SUBSCRIBE client-individual; Acknowledge() each delivery.
using var conn = factory.CreateConnection();
conn.Start();
using var consumeSession = conn.CreateSession(AcknowledgementMode.IndividualAcknowledge);
using var consumer = consumeSession.CreateConsumer(consumeSession.GetQueue(destination));
for (var i = 0; i < count; i++)
{
    var msg = consumer.Receive(TimeSpan.FromSeconds(10))
        ?? throw new InvalidOperationException("timed out draining the queue");
    if (msg.Headers.GetValue("redelivered") == "true")
        throw new InvalidOperationException("unexpected redelivery");
    msg.Acknowledge(); // ACK by the 1.2 ack token
}
Console.WriteLine($"drained and acked {count} jobs");

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 = "/queue/jobs/email" # Queues pattern → channel jobs.email
COUNT = 5

# 1. Produce — SEND `count` jobs to the queue.
pub = Stomp::Client.new(hosts: hosts, connect_headers: { "accept-version" => "1.2", "host" => "/" })
(1..COUNT).each { |i| pub.publish(DESTINATION, "job-#{i}", "content-type" => "text/plain") }
pub.close

# 2. Consume — SUBSCRIBE ack:client-individual; ACK each delivery by its 1.2 token.
inbox = Thread::Queue.new
sub = Stomp::Client.new(hosts: hosts, connect_headers: { "accept-version" => "1.2", "host" => "/" })
sub.subscribe(DESTINATION, id: "jobs", ack: "client-individual") { |msg| inbox << msg }

COUNT.times do
  msg = Timeout.timeout(10) { inbox.pop }
  raise "unexpected redelivery" if msg.headers["redelivered"] == "true"

  sub.acknowledge(msg) # ACK by the 1.2 ack token
end
sub.close
puts "drained and acked #{COUNT} jobs"

use std::time::Duration;

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

const DESTINATION: &str = "/queue/jobs/email"; // Queues pattern → channel jobs.email
const COUNT: usize = 5;

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. Produce — SEND `count` jobs to the queue.
    let mut pub_conn = Connector::builder()
        .server(format!("{host}:{port}"))
        .virtualhost(&host)
        .connect()
        .await?;
    for i in 1..=COUNT {
        pub_conn
            .send(ToServer::Send {
                destination: DESTINATION.into(),
                transaction: None,
                headers: Some(vec![("content-type".into(), "text/plain".into())]),
                body: Some(format!("job-{i}").into_bytes()),
            }.into())
            .await?;
    }
    pub_conn.send(ToServer::Disconnect { receipt: None }.into()).await?;

    // 2. Consume — SUBSCRIBE ack:client-individual; ACK each by its 1.2 token.
    let mut sub = Connector::builder()
        .server(format!("{host}:{port}"))
        .virtualhost(&host)
        .connect()
        .await?;
    sub.send(ToServer::Subscribe {
        destination: DESTINATION.into(),
        id: "jobs".into(),
        ack: Some(AckMode::ClientIndividual),
    }.into())
    .await?;

    let mut got = 0;
    while got < COUNT {
        let frame = tokio::time::timeout(Duration::from_secs(10), sub.next())
            .await?
            .ok_or("stream closed")??;
        if let FromServer::Message { headers, .. } = frame.content {
            let ack_token = headers.iter().find(|(k, _)| k == "ack").map(|(_, v)| v.clone());
            if let Some(token) = ack_token {
                sub.send(ToServer::Ack { id: token, transaction: None }.into()).await?;
            }
            got += 1;
        }
    }
    println!("drained and acked {COUNT} jobs");
    Ok(())
}