Queries

package main

import (
	"context"
	"fmt"
	"log"
	"net"
	"os"
	"strings"
	"time"

	"github.com/eclipse/paho.golang/paho"
	"github.com/google/uuid"
)

func brokerURL() string {
	if u := os.Getenv("KUBEMQ_MQTT_URL"); u != "" {
		return u
	}
	return "tcp://localhost:1883"
}

func tcpAddr(raw string) string {
	for _, pfx := range []string{"tcp://", "ws://", "tls://"} {
		if strings.HasPrefix(raw, pfx) {
			return raw[len(pfx):]
		}
	}
	return raw
}

func main() {
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()
	addr := tcpAddr(brokerURL())

	clientID := "go-query-" + uuid.NewString()[:8]
	correlationID := uuid.NewString()
	replyTopic := "$reply/" + clientID + "/inbox" // own namespace only
	queryTopic := "queries/demo/rpc"               // channel demo.rpc

	resp := make(chan *paho.Publish, 1)
	conn, err := net.Dial("tcp", addr)
	if err != nil {
		log.Fatalf("dial: %v", err)
	}
	client := paho.NewClient(paho.ClientConfig{
		Conn: conn,
		OnPublishReceived: []func(paho.PublishReceived) (bool, error){
			func(pr paho.PublishReceived) (bool, error) {
				select {
				case resp <- pr.Packet:
				default:
				}
				return true, nil
			},
		},
	})
	ack, err := client.Connect(ctx, &paho.Connect{ClientID: clientID, KeepAlive: 30, CleanStart: true})
	if err != nil || ack.ReasonCode != 0 {
		log.Fatalf("connect: %v (reason 0x%02X)", err, ack.ReasonCode)
	}

	// 1. Subscribe to your own reply topic BEFORE publishing.
	subAck, err := client.Subscribe(ctx, &paho.Subscribe{
		Subscriptions: []paho.SubscribeOptions{{Topic: replyTopic, QoS: 1}},
	})
	if err != nil {
		log.Fatalf("subscribe: %v", err)
	}
	if subAck.Reasons[0] == 0x83 {
		log.Fatal("SUBACK 0x83: reply topic must be $reply/<own-clientID>/...")
	}

	// 2. Publish the query with ResponseTopic + CorrelationData.
	pubAck, err := client.Publish(ctx, &paho.Publish{
		Topic:   queryTopic,
		QoS:     1,
		Payload: []byte(`{"action":"echo","data":"hello"}`),
		Properties: &paho.PublishProperties{
			ResponseTopic:   replyTopic,
			CorrelationData: []byte(correlationID),
		},
	})
	if err != nil {
		log.Fatalf("publish: %v", err)
	}
	// PUBACK is immediate — the response arrives later on the reply topic.
	if pubAck.ReasonCode != 0 {
		log.Fatalf("PUBACK reason=0x%02X (0x83=bad ResponseTopic, 0x97=RpcMaxPending)", pubAck.ReasonCode)
	}

	// 3. Wait for the response — queries carry a body (plus kubemq-metadata).
	select {
	case r := <-resp:
		var metadata string
		if r.Properties != nil {
			for _, up := range r.Properties.User {
				if up.Key == "kubemq-metadata" {
					metadata = up.Value
				}
			}
		}
		fmt.Printf("query response body: %s (metadata=%s)\n", r.Payload, metadata)
	case <-ctx.Done():
		log.Fatal("timed out — is a gRPC responder running on channel demo.rpc?")
	}
	_ = client.Disconnect(&paho.Disconnect{ReasonCode: 0})
}

import os
import threading
import time
import uuid

import paho.mqtt.client as mqtt
from paho.mqtt.enums import CallbackAPIVersion
from paho.mqtt.properties import Properties
from paho.mqtt.packettypes import PacketTypes

CLIENT_ID = "py-query-client"
QUERY_TOPIC = "queries/demo/rpc"               # channel demo.rpc
REPLY_TOPIC = f"$reply/{CLIENT_ID}/inbox"      # own namespace only
CORRELATION = uuid.uuid4().bytes


def parse_url(url: str) -> tuple[str, int]:
    scheme, rest = url.split("://", 1)
    host, _, port = rest.rstrip("/").partition(":")
    return host, int(port) if port else {"tcp": 1883, "tls": 8883, "ws": 8083}[scheme]


def main() -> None:
    host, port = parse_url(os.environ.get("KUBEMQ_MQTT_URL", "tcp://localhost:1883"))
    response = threading.Event()
    result: dict = {}

    def on_message(client, userdata, msg):
        props = {}
        if msg.properties and hasattr(msg.properties, "UserProperty"):
            for k, v in (msg.properties.UserProperty or []):
                props[k] = v
        result["body"] = msg.payload
        result["metadata"] = props.get("kubemq-metadata")
        response.set()

    client = mqtt.Client(CallbackAPIVersion.VERSION2, client_id=CLIENT_ID, protocol=mqtt.MQTTv5)
    # 1. Subscribe to your own reply topic BEFORE publishing.
    client.on_connect = lambda c, *_: c.subscribe(REPLY_TOPIC, qos=1)
    client.on_message = on_message
    client.connect(host, port, keepalive=30)
    client.loop_start()
    time.sleep(0.5)

    # 2. Publish the query with ResponseTopic + CorrelationData.
    pub_props = Properties(PacketTypes.PUBLISH)
    pub_props.ResponseTopic = REPLY_TOPIC
    pub_props.CorrelationData = CORRELATION
    # PUBACK is immediate — the response arrives later on the reply topic.
    client.publish(QUERY_TOPIC, payload=b'{"query": "hello"}', qos=1, properties=pub_props).wait_for_publish(10)

    # 3. Wait for the response — queries carry a body (plus kubemq-metadata).
    if not response.wait(timeout=30):
        raise TimeoutError("no response — is a gRPC responder running on channel demo.rpc?")
    print(f"query response body: {result['body'].decode()!r} (metadata={result['metadata']})")

    client.loop_stop()
    client.disconnect()


if __name__ == "__main__":
    main()

import java.nio.charset.StandardCharsets;
import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;

import org.eclipse.paho.mqttv5.client.MqttAsyncClient;
import org.eclipse.paho.mqttv5.client.MqttCallback;
import org.eclipse.paho.mqttv5.client.MqttConnectionOptions;
import org.eclipse.paho.mqttv5.client.MqttDisconnectResponse;
import org.eclipse.paho.mqttv5.client.IMqttToken;
import org.eclipse.paho.mqttv5.common.MqttException;
import org.eclipse.paho.mqttv5.common.MqttMessage;
import org.eclipse.paho.mqttv5.common.packet.MqttProperties;
import org.eclipse.paho.mqttv5.common.packet.UserProperty;

public final class Main {
    public static void main(String[] args) throws Exception {
        String broker = System.getenv().getOrDefault("KUBEMQ_MQTT_URL", "tcp://localhost:1883");
        String clientId = "java-query-" + UUID.randomUUID().toString().substring(0, 6);
        String queryTopic = "queries/demo/rpc";             // channel demo.rpc
        String replyTopic = "$reply/" + clientId + "/inbox"; // own namespace only
        byte[] correlation = "corr-123".getBytes(StandardCharsets.UTF_8);

        BlockingQueue<MqttMessage> responses = new ArrayBlockingQueue<>(1);

        MqttAsyncClient client = new MqttAsyncClient(broker, clientId);
        client.setCallback(new MqttCallback() {
            public void messageArrived(String t, MqttMessage m) { responses.offer(m); }
            public void disconnected(MqttDisconnectResponse r) {}
            public void mqttErrorOccurred(MqttException e) {}
            public void deliveryComplete(IMqttToken t) {}
            public void connectComplete(boolean reconnect, String uri) {}
            public void authPacketArrived(int code, MqttProperties props) {}
        });

        MqttConnectionOptions opts = new MqttConnectionOptions();
        opts.setCleanStart(true);
        opts.setKeepAliveInterval(30);
        client.connect(opts).waitForCompletion(5_000);

        // 1. Subscribe to your own reply topic BEFORE publishing.
        client.subscribe(replyTopic, 1).waitForCompletion(5_000);
        Thread.sleep(200);

        // 2. Publish the query with ResponseTopic + CorrelationData.
        MqttProperties pubProps = new MqttProperties();
        pubProps.setResponseTopic(replyTopic);
        pubProps.setCorrelationData(correlation);
        MqttMessage msg = new MqttMessage("ping".getBytes(StandardCharsets.UTF_8));
        msg.setQos(1);
        msg.setProperties(pubProps);
        // PUBACK is immediate — the response arrives later on the reply topic.
        client.publish(queryTopic, msg).waitForCompletion(5_000);

        // 3. Wait for the response — queries carry a body (plus kubemq-metadata).
        MqttMessage response = responses.poll(15, TimeUnit.SECONDS);
        if (response == null) {
            throw new IllegalStateException("no response — is a gRPC responder running on channel demo.rpc?");
        }
        String body = new String(response.getPayload(), StandardCharsets.UTF_8);
        String metadata = "";
        MqttProperties props = response.getProperties();
        if (props != null && props.getUserProperties() != null) {
            for (UserProperty up : props.getUserProperties()) {
                if ("kubemq-metadata".equals(up.getKey())) metadata = up.getValue();
            }
        }
        System.out.printf("query response body: %s (metadata=%s)%n", body, metadata);

        client.disconnect().waitForCompletion(3_000);
        client.close();
    }
}

import mqtt, { type MqttClient } from "mqtt";
import crypto from "node:crypto";

const CLIENT_ID = `js-query-${crypto.randomBytes(4).toString("hex")}`;
const REPLY_TOPIC = `$reply/${CLIENT_ID}/inbox`; // own namespace only
const QUERY_TOPIC = "queries/demo/rpc";           // channel demo.rpc
const CORRELATION_DATA = Buffer.from(crypto.randomUUID());
const RPC_TIMEOUT_MS = 30_000;

async function main(): Promise<void> {
  const url = process.env["KUBEMQ_MQTT_URL"] ?? "tcp://localhost:1883";
  const client: MqttClient = mqtt.connect(url, {
    clientId: CLIENT_ID,
    protocolVersion: 5,
    clean: true,
    keepalive: 30,
  });
  await new Promise<void>((res, rej) => {
    client.once("connect", () => res());
    client.once("error", rej);
  });

  // 1. Subscribe to your own reply topic BEFORE publishing.
  await new Promise<void>((resolve, reject) => {
    client.subscribe(REPLY_TOPIC, { qos: 1 }, (err, granted) => {
      if (err) return reject(err);
      if (((granted?.[0]?.qos as number) ?? -1) > 2) return reject(new Error("reply subscribe rejected"));
      resolve();
    });
  });

  // 2. Arm the response handler, then publish with ResponseTopic + CorrelationData.
  const responded = new Promise<void>((resolve, reject) => {
    const timer = setTimeout(
      () => reject(new Error("RPC timeout — is a gRPC responder running on channel demo.rpc?")),
      RPC_TIMEOUT_MS,
    );
    client.on("message", (topic, payload, packet) => {
      if (topic !== REPLY_TOPIC) return;
      clearTimeout(timer);
      const userProps = (packet.properties as { userProperties?: Record<string, string> } | undefined)
        ?.userProperties ?? {};
      // Query responses carry a body (unlike commands).
      console.log(`query response body: ${payload.toString()} (metadata=${userProps["kubemq-metadata"] ?? ""})`);
      resolve();
    });
  });

  await new Promise<void>((resolve, reject) => {
    client.publish(
      QUERY_TOPIC,
      JSON.stringify({ request: "hello" }),
      { qos: 1, properties: { responseTopic: REPLY_TOPIC, correlationData: CORRELATION_DATA } },
      (err) => (err ? reject(err) : resolve()), // PUBACK is immediate, not the response
    );
  });

  await responded;
  await client.endAsync();
}

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

using System.Text;
using MQTTnet;
using MQTTnet.Client;
using MQTTnet.Protocol;

static (string host, int port) Endpoint()
{
    var url = Environment.GetEnvironmentVariable("KUBEMQ_MQTT_URL") ?? "tcp://localhost:1883";
    foreach (var pfx in new[] { "tcp://", "tls://", "ws://" })
        if (url.StartsWith(pfx, StringComparison.OrdinalIgnoreCase)) url = url[pfx.Length..];
    var parts = url.TrimEnd('/').Split(':');
    return (parts[0], parts.Length > 1 ? int.Parse(parts[1]) : 1883);
}

var (host, port) = Endpoint();
var clientId = $"csharp-query-{Guid.NewGuid():N}"[..23];
var requestId = Guid.NewGuid().ToString("N");
var replyTopic = $"$reply/{clientId}/inbox"; // own namespace only
const string queryTopic = "queries/demo/rpc"; // channel demo.rpc

var factory = new MqttFactory();
var responded = new TaskCompletionSource<(string body, string metadata)>(
    TaskCreationOptions.RunContinuationsAsynchronously);

using var client = factory.CreateMqttClient();
client.ApplicationMessageReceivedAsync += e =>
{
    var body = Encoding.UTF8.GetString(e.ApplicationMessage.PayloadSegment);
    var metadata = string.Empty;
    foreach (var p in e.ApplicationMessage.UserProperties ?? new())
        if (p.Name == "kubemq-metadata") metadata = p.Value;
    responded.TrySetResult((body, metadata));
    return Task.CompletedTask;
};

var options = new MqttClientOptionsBuilder()
    .WithTcpServer(host, port)
    .WithClientId(clientId)
    .WithProtocolVersion(MQTTnet.Formatter.MqttProtocolVersion.V500)
    .WithCleanSession(true)
    .Build();
await client.ConnectAsync(options);

// 1. Subscribe to your own reply topic BEFORE publishing.
var subResult = await client.SubscribeAsync(new MqttClientSubscribeOptionsBuilder()
    .WithTopicFilter(f => f.WithTopic(replyTopic).WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce))
    .Build());
if (subResult.Items.First().ResultCode > MqttClientSubscribeResultCode.GrantedQoS2)
    throw new Exception("reply subscribe rejected — use $reply/<own-clientID>/ namespace");
await Task.Delay(200);

// 2. Publish the query with ResponseTopic + CorrelationData.
var pubResult = await client.PublishAsync(new MqttApplicationMessageBuilder()
    .WithTopic(queryTopic)
    .WithPayload(Encoding.UTF8.GetBytes($"{{\"query\":\"hello\",\"requestId\":\"{requestId}\"}}"))
    .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce)
    .WithResponseTopic(replyTopic)
    .WithCorrelationData(Encoding.UTF8.GetBytes(requestId))
    .Build());
// PUBACK is immediate — the response arrives later on the reply topic.
if (pubResult.ReasonCode != MqttClientPublishReasonCode.Success)
    throw new Exception($"PUBACK reason: {pubResult.ReasonCode}");

// 3. Wait for the response — queries carry a body (plus kubemq-metadata).
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(15));
var (body, metadata) = await responded.Task.WaitAsync(cts.Token);
Console.WriteLine($"query response body: {body} (metadata={metadata})");

await client.DisconnectAsync();

use bytes::Bytes;
use rumqttc::v5::mqttbytes::v5::{Packet, PublishProperties};
use rumqttc::v5::mqttbytes::QoS;
use rumqttc::v5::{AsyncClient, Event, MqttOptions};
use std::env;
use tokio::sync::oneshot;
use tokio::time::{timeout, Duration};
use uuid::Uuid;

fn parse_host_port(url: &str) -> (String, u16) {
    let stripped = url
        .trim_start_matches("tcp://")
        .trim_start_matches("tls://")
        .trim_start_matches("ws://");
    let host_port = stripped.split('/').next().unwrap_or(stripped);
    let mut parts = host_port.splitn(2, ':');
    let host = parts.next().unwrap_or("localhost").to_string();
    let port: u16 = parts.next().and_then(|p| p.parse().ok()).unwrap_or(1883);
    (host, port)
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let url = env::var("KUBEMQ_MQTT_URL").unwrap_or_else(|_| "tcp://localhost:1883".to_string());
    let (host, port) = parse_host_port(&url);

    let client_id = format!("rust-query-{}", &Uuid::new_v4().to_string()[..8]);
    let reply_topic = format!("$reply/{}/inbox", client_id); // own namespace only
    let query_topic = "queries/demo/rpc";                     // channel demo.rpc
    let correlation = Uuid::new_v4().to_string();
    let corr_bytes = Bytes::from(correlation.clone().into_bytes());

    let mut opts = MqttOptions::new(&client_id, &host, port);
    opts.set_keep_alive(Duration::from_secs(30));
    let (client, mut eventloop) = AsyncClient::new(opts, 10);

    let (tx, rx) = oneshot::channel::<(String, Option<String>)>();
    let reply_clone = reply_topic.clone();
    let corr_check = corr_bytes.clone();
    tokio::spawn(async move {
        let mut tx = Some(tx);
        loop {
            match eventloop.poll().await {
                Ok(Event::Incoming(Packet::Publish(p))) => {
                    if String::from_utf8_lossy(&p.topic) != reply_clone {
                        continue;
                    }
                    let props = p.properties.as_ref();
                    // Match the response to the request via CorrelationData.
                    if props.and_then(|pr| pr.correlation_data.clone()).as_deref()
                        != Some(corr_check.as_ref())
                    {
                        continue;
                    }
                    // Query responses carry a body (plus kubemq-metadata).
                    let body = String::from_utf8_lossy(&p.payload).to_string();
                    let metadata = props
                        .and_then(|pr| pr.user_properties.iter().find(|(k, _)| k == "kubemq-metadata"))
                        .map(|(_, v)| v.clone());
                    if let Some(tx) = tx.take() {
                        let _ = tx.send((body, metadata));
                        return;
                    }
                }
                Ok(_) => {}
                Err(e) => { eprintln!("event loop: {e}"); return; }
            }
        }
    });

    // 1. Subscribe to your own reply topic BEFORE publishing.
    client.subscribe(&reply_topic, QoS::AtLeastOnce).await?;
    tokio::time::sleep(Duration::from_millis(300)).await;

    // 2. Publish the query with ResponseTopic + CorrelationData.
    let props = PublishProperties {
        response_topic: Some(reply_topic.clone()),
        correlation_data: Some(corr_bytes),
        ..Default::default()
    };
    // PUBACK is immediate — the response arrives later on the reply topic.
    client
        .publish_with_properties(
            query_topic,
            QoS::AtLeastOnce,
            false,
            br#"{"action":"echo","data":"hello"}"#.as_ref(),
            props,
        )
        .await?;

    // 3. Wait for the response — queries carry a body (plus kubemq-metadata).
    let (body, metadata) = timeout(Duration::from_secs(30), rx)
        .await
        .map_err(|_| "timed out — is a gRPC responder running on channel demo.rpc?")??;
    println!("query response body: {body} (metadata={})", metadata.as_deref().unwrap_or(""));
    Ok(())
}