Commands & Queries

// Example: commands/round-trip
//
// Demonstrates an RPC command round-trip:
//   - Responder goroutine subscribes via SSE GET /ce/subscribe/commands
//   - Sender goroutine sends a command via POST /ce/send/command
//   - Responder extracts _kubemq_request_id and _kubemq_reply_channel
//   - Responder sends back a response via POST /ce/send/response?request_id=...
//   - Sender receives the execution acknowledgement in the POST response body
//
// Run: go run ./commands/round-trip/main.go
package main

import (
	"bufio"
	"encoding/json"
	"fmt"
	"log"
	"net/http"
	"os"
	"strings"
	"sync"
	"time"

	cloudevents "github.com/cloudevents/sdk-go/v2"
)

func serverURL() string {
	if u := os.Getenv("KUBEMQ_CE_URL"); u != "" {
		return u
	}
	return "http://localhost:9090"
}

type CEResponse struct {
	IsError bool            `json:"is_error"`
	Message string          `json:"message"`
	Data    json.RawMessage `json:"data"`
}

// startResponder subscribes to commands and replies to each one.
func startResponder(base, channel string, ready chan<- struct{}, wg *sync.WaitGroup) {
	defer wg.Done()
	sseURL := fmt.Sprintf("%s/ce/subscribe/commands?client_id=go-cmd-responder&channel=%s",
		base, channel)
	req, _ := http.NewRequest("GET", sseURL, nil)
	req.Header.Set("Accept", "text/event-stream")
	req.Header.Set("Cache-Control", "no-cache")

	client := &http.Client{Timeout: 0}
	resp, err := client.Do(req)
	if err != nil {
		log.Fatal("responder SSE connect:", err)
	}
	defer resp.Body.Close()

	close(ready) // signal that SSE is connected

	scanner := bufio.NewScanner(resp.Body)
	var evType, data string
	for scanner.Scan() {
		line := scanner.Text()
		if line == "" {
			if evType == "cloudevent" && data != "" {
				// Parse command to extract KubeMQ correlation fields.
				var raw map[string]interface{}
				_ = json.Unmarshal([]byte(data), &raw)

				requestID, _ := raw["_kubemq_request_id"].(string)
				replyChannel, _ := raw["_kubemq_reply_channel"].(string)
				fmt.Printf("[responder] command received: type=%v request_id=%s\n",
					raw["type"], requestID)

				// Send response back.
				sendResponse(base, requestID, replyChannel)
				return
			}
			if evType == "error" {
				log.Printf("[responder] SSE error: %s", data)
				return
			}
			evType, data = "", ""
			continue
		}
		if strings.HasPrefix(line, ":") {
			continue
		}
		if strings.HasPrefix(line, "event:") {
			evType = strings.TrimSpace(strings.TrimPrefix(line, "event:"))
		} else if strings.HasPrefix(line, "data:") {
			data = strings.TrimSpace(strings.TrimPrefix(line, "data:"))
		}
	}
}

func sendResponse(base, requestID, replyChannel string) {
	event := cloudevents.NewEvent()
	event.SetType("com.kubemq.examples.commands.response")
	event.SetSource("go-cmd-responder")
	event.SetSubject(replyChannel) // subject = reply channel
	_ = event.SetData(cloudevents.ApplicationJSON, map[string]interface{}{
		"executed": true,
		"status":   "command processed successfully",
	})

	body, _ := json.Marshal(event)
	url := fmt.Sprintf("%s/ce/send/response?request_id=%s", base, requestID)
	req, _ := http.NewRequest("POST", url, strings.NewReader(string(body)))
	req.Header.Set("Content-Type", "application/cloudevents+json")

	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		log.Fatal("send response:", err)
	}
	defer resp.Body.Close()

	var result CEResponse
	_ = json.NewDecoder(resp.Body).Decode(&result)
	fmt.Printf("[responder] response sent: is_error=%v\n", result.IsError)
}

func sendCommand(base, channel string) {
	event := cloudevents.NewEvent()
	event.SetType("com.kubemq.examples.commands.reboot")
	event.SetSource("kubemq-ce-go-sender")
	event.SetSubject(channel)
	_ = event.SetData(cloudevents.ApplicationJSON, map[string]string{
		"device_id": "sensor-42",
		"action":    "reboot",
	})

	body, _ := json.Marshal(event)
	req, _ := http.NewRequest("POST", base+"/ce/send/command", strings.NewReader(string(body)))
	req.Header.Set("Content-Type", "application/cloudevents+json")

	fmt.Println("[sender] sending command...")
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		log.Fatal("send command:", err)
	}
	defer resp.Body.Close()

	var result CEResponse
	_ = json.NewDecoder(resp.Body).Decode(&result)
	fmt.Printf("[sender] command ack received: status=%d is_error=%v data=%s\n",
		resp.StatusCode, result.IsError, string(result.Data))
}

func main() {
	base := serverURL()
	channel := "go-ce-commands.round-trip"

	ready := make(chan struct{})
	var wg sync.WaitGroup
	wg.Add(1)
	go startResponder(base, channel, ready, &wg)

	// Wait for SSE to be established.
	select {
	case <-ready:
	case <-time.After(5 * time.Second):
		log.Fatal("Timed out waiting for responder to connect")
	}
	time.Sleep(100 * time.Millisecond)

	// Send command — blocks until response arrives or timeout.
	sendCommand(base, channel)
	wg.Wait()
}

"""Example: commands/round_trip — RPC command with execution ack."""

from __future__ import annotations

import json
import os
import threading
import time

import requests
from cloudevents.v1.conversion import to_structured
from cloudevents.v1.http import CloudEvent


def server_url() -> str:
    return os.environ.get("KUBEMQ_CE_URL", "http://localhost:9090")


def responder(base: str, channel: str, ready: threading.Event) -> None:
    sse_url = (f"{base}/ce/subscribe/commands"
               f"?client_id=python-cmd-responder&channel={channel}")
    with requests.get(sse_url, stream=True, timeout=None,
                      headers={"Accept": "text/event-stream"}) as resp:
        ready.set()
        ev_type = data = ""
        for line in resp.iter_lines(decode_unicode=True):
            if line == "":
                if ev_type == "cloudevent" and data:
                    raw = json.loads(data)
                    request_id = raw.get("_kubemq_request_id", "")
                    reply_channel = raw.get("_kubemq_reply_channel", "")
                    print(f"[responder] command received: type={raw.get('type')} "
                          f"request_id={request_id}")
                    # Send response.
                    response_event = CloudEvent(
                        attributes={
                            "type": "com.kubemq.examples.commands.response",
                            "source": "python-cmd-responder",
                            "subject": reply_channel,
                            "datacontenttype": "application/json",
                        },
                        data={"executed": True, "status": "command processed"},
                    )
                    headers, body = to_structured(response_event)
                    r = requests.post(
                        f"{base}/ce/send/response",
                        params={"request_id": request_id},
                        data=body, headers=dict(headers), timeout=10,
                    )
                    print(f"[responder] response sent: is_error={r.json().get('is_error')}")
                    return
                ev_type = data = ""
                continue
            if line.startswith(":"):
                continue
            if line.startswith("event:"):
                ev_type = line[6:].strip()
            elif line.startswith("data:"):
                data = line[5:].strip()


def main() -> None:
    base = server_url()
    channel = "python-ce-commands.round-trip"

    ready = threading.Event()
    t = threading.Thread(target=responder, args=(base, channel, ready), daemon=True)
    t.start()

    ready.wait(timeout=5)
    time.sleep(0.1)

    event = CloudEvent(
        attributes={
            "type": "com.kubemq.examples.commands.reboot",
            "source": "kubemq-ce-python-sender",
            "subject": channel,
            "datacontenttype": "application/json",
        },
        data={"device_id": "sensor-42", "action": "reboot"},
    )
    headers, body = to_structured(event)
    print("[sender] sending command...")
    resp = requests.post(f"{base}/ce/send/command", data=body,
                          headers=dict(headers), timeout=30)
    result = resp.json()
    print(f"[sender] command ack: status={resp.status_code} is_error={result.get('is_error')}")
    print(f"[sender] response data: {result.get('data')}")
    t.join(timeout=5)


if __name__ == "__main__":
    main()

/**
 * Example: commands/round-trip — RPC command with execution ack.
 * Run: npx tsx commands/round-trip/index.ts
 */
import { CloudEvent, HTTP } from 'cloudevents';
import EventSource from 'eventsource';

function serverUrl(): string {
  return process.env.KUBEMQ_CE_URL ?? 'http://localhost:9090';
}

function startResponder(base: string, channel: string): Promise<void> {
  return new Promise((resolve) => {
    const url = `${base}/ce/subscribe/commands?client_id=js-cmd-responder&channel=${encodeURIComponent(channel)}`;
    const es = new EventSource(url);
    es.addEventListener('cloudevent', async (evt: MessageEvent) => {
      es.close();
      const raw = JSON.parse(evt.data) as Record<string, unknown>;
      const requestId = raw._kubemq_request_id as string;
      const replyChannel = raw._kubemq_reply_channel as string;
      console.log(`[responder] command received: type=${raw.type} request_id=${requestId}`);

      const response = new CloudEvent({
        type: 'com.kubemq.examples.commands.response',
        source: 'js-cmd-responder',
        subject: replyChannel,
        datacontenttype: 'application/json',
        data: { executed: true, status: 'command processed' },
      });
      const msg = HTTP.structured(response);
      const r = await fetch(`${base}/ce/send/response?request_id=${requestId}`, {
        method: 'POST',
        headers: msg.headers as Record<string, string>,
        body: msg.body as string,
      });
      const result = await r.json() as { is_error: boolean };
      console.log(`[responder] response sent: is_error=${result.is_error}`);
      resolve();
    });
    es.addEventListener('error', (evt: MessageEvent) => {
      if (evt.data) {
        const err = JSON.parse(evt.data) as { message: string };
        console.error('[responder] SSE error:', err.message);
        es.close();
      }
    });
  });
}

async function main(): Promise<void> {
  const base = serverUrl();
  const channel = 'js-ce-commands.round-trip';

  const responderDone = startResponder(base, channel);
  await new Promise((r) => setTimeout(r, 500));

  const event = new CloudEvent({
    type: 'com.kubemq.examples.commands.reboot',
    source: 'kubemq-ce-js-sender',
    subject: channel,
    datacontenttype: 'application/json',
    data: { device_id: 'sensor-42', action: 'reboot' },
  });
  const msg = HTTP.structured(event);
  console.log('[sender] sending command...');
  const resp = await fetch(`${base}/ce/send/command`, {
    method: 'POST',
    headers: msg.headers as Record<string, string>,
    body: msg.body as string,
  });
  const result = await resp.json() as { is_error: boolean; data: unknown };
  console.log(`[sender] command ack: status=${resp.status} is_error=${result.is_error}`);

  await responderDone;
}

main().catch(console.error);

package io.kubemq.examples.commands.roundtrip;

import com.fasterxml.jackson.databind.ObjectMapper;
import io.cloudevents.CloudEvent;
import io.cloudevents.core.builder.CloudEventBuilder;
import io.cloudevents.core.format.EventFormat;
import io.cloudevents.core.provider.EventFormatProvider;
import io.cloudevents.jackson.JsonFormat;

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.net.HttpURLConnection;
import java.net.URI;
import java.net.URL;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.nio.charset.StandardCharsets;
import java.time.OffsetDateTime;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * Example: commands/round-trip
 *
 * A responder subscribes to commands via SSE, receives the command, and sends
 * a CE response. A sender publishes a command via POST /ce/send/command
 * (which blocks until ack is received).
 *
 * Run: mvn compile exec:java
 */
public class Main {
    static String serverUrl() {
        String u = System.getenv("KUBEMQ_CE_URL");
        return (u != null && !u.isEmpty()) ? u : "http://localhost:9090";
    }
    static final ObjectMapper MAPPER = new ObjectMapper();

    public static void main(String[] args) throws Exception {
        String base = serverUrl();
        String channel = "java-ce-commands.round-trip";

        EventFormatProvider.getInstance().registerFormat(new JsonFormat());
        EventFormat format = EventFormatProvider.getInstance().resolveFormat(JsonFormat.CONTENT_TYPE);
        HttpClient httpClient = HttpClient.newHttpClient();

        BlockingQueue<Boolean> responderReady = new ArrayBlockingQueue<>(1);

        // Start command responder.
        String sseUrl = base + "/ce/subscribe/commands?client_id=java-cmd-responder&channel=" + channel;
        Thread.ofVirtual().start(() -> {
            try {
                HttpURLConnection conn = (HttpURLConnection) new URL(sseUrl).openConnection();
                conn.setRequestProperty("Accept", "text/event-stream");
                conn.setReadTimeout(30_000);
                try (BufferedReader reader = new BufferedReader(
                        new InputStreamReader(conn.getInputStream(), StandardCharsets.UTF_8))) {
                    String line; String evType = null, data = null;
                    boolean ready = false;
                    while ((line = reader.readLine()) != null) {
                        if (!ready) { responderReady.offer(true); ready = true; }
                        if (line.isEmpty()) {
                            if ("cloudevent".equals(evType) && data != null) {
                                Map<?, ?> raw = MAPPER.readValue(data, Map.class);
                                String requestId = (String) raw.get("_kubemq_request_id");
                                String replyChannel = (String) raw.get("_kubemq_reply_channel");
                                System.out.println("[responder] command received: type=" + raw.get("type")
                                        + " request_id=" + requestId);

                                CloudEvent respEvent = CloudEventBuilder.v1()
                                        .withId(UUID.randomUUID().toString())
                                        .withType("com.kubemq.examples.commands.response")
                                        .withSource(URI.create("kubemq-ce-java-responder"))
                                        .withSubject(replyChannel)
                                        .withDataContentType("application/json")
                                        .withTime(OffsetDateTime.now())
                                        .withData("application/json",
                                                MAPPER.writeValueAsBytes(Map.of("executed", true, "status", "command processed")))
                                        .build();
                                String responseUrl = base + "/ce/send/response?request_id=" + requestId;
                                HttpResponse<String> r = httpClient.send(
                                        HttpRequest.newBuilder().uri(URI.create(responseUrl))
                                                .POST(HttpRequest.BodyPublishers.ofByteArray(format.serialize(respEvent)))
                                                .header("Content-Type", "application/cloudevents+json").build(),
                                        HttpResponse.BodyHandlers.ofString());
                                Map<?, ?> rResult = MAPPER.readValue(r.body(), Map.class);
                                System.out.println("[responder] response sent: is_error=" + rResult.get("is_error"));
                                return;
                            }
                            evType = null; data = null;
                        } else if (line.startsWith("event:")) evType = line.substring(6).trim();
                        else if (line.startsWith("data:")) data = line.substring(5).trim();
                    }
                }
            } catch (Exception e) { System.err.println("SSE error: " + e.getMessage()); }
        });

        responderReady.poll(5, TimeUnit.SECONDS);
        Thread.sleep(100);

        // Send command.
        CloudEvent cmd = CloudEventBuilder.v1()
                .withId(UUID.randomUUID().toString())
                .withType("com.kubemq.examples.commands.reboot")
                .withSource(URI.create("kubemq-ce-java-sender"))
                .withSubject(channel)
                .withDataContentType("application/json")
                .withTime(OffsetDateTime.now())
                .withData("application/json",
                        MAPPER.writeValueAsBytes(Map.of("device_id", "sensor-42", "action", "reboot")))
                .build();

        System.out.println("[sender] sending command...");
        HttpResponse<String> resp = httpClient.send(
                HttpRequest.newBuilder().uri(URI.create(base + "/ce/send/command"))
                        .POST(HttpRequest.BodyPublishers.ofByteArray(format.serialize(cmd)))
                        .header("Content-Type", "application/cloudevents+json").build(),
                HttpResponse.BodyHandlers.ofString());
        Map<?, ?> result = MAPPER.readValue(resp.body(), Map.class);
        System.out.println("[sender] command ack: status=" + resp.statusCode() + " is_error=" + result.get("is_error"));
    }
}

// Example: commands/RoundTrip — RPC command with execution ack.
// Run: dotnet run
using CloudNative.CloudEvents;
using CloudNative.CloudEvents.SystemTextJson;
using System.Net.Http.Headers;
using System.Text;
using System.Text.Json;

static string ServerUrl() => Environment.GetEnvironmentVariable("KUBEMQ_CE_URL") ?? "http://localhost:9090";

var base_ = ServerUrl();
var channel = "csharp-ce-commands.round-trip";
var formatter = new JsonEventFormatter();

async Task RunResponder()
{
    using var http = new HttpClient { Timeout = Timeout.InfiniteTimeSpan };
    var url = $"{base_}/ce/subscribe/commands?client_id=csharp-cmd-responder&channel={Uri.EscapeDataString(channel)}";
    using var req = new HttpRequestMessage(HttpMethod.Get, url);
    req.Headers.Accept.Add(new MediaTypeWithQualityHeaderValue("text/event-stream"));
    using var resp = await http.SendAsync(req, HttpCompletionOption.ResponseHeadersRead);
    using var stream = await resp.Content.ReadAsStreamAsync();
    using var reader = new StreamReader(stream, Encoding.UTF8);
    string? evType = null, data = null, line;
    while ((line = await reader.ReadLineAsync()) != null)
    {
        if (line == "")
        {
            if (evType == "cloudevent" && data != null)
            {
                var raw = JsonSerializer.Deserialize<JsonElement>(data);
                var requestId = raw.GetProperty("_kubemq_request_id").GetString()!;
                var replyChannel = raw.GetProperty("_kubemq_reply_channel").GetString()!;
                Console.WriteLine($"[responder] command received: type={raw.GetProperty("type")} request_id={requestId}");

                var responseEvent = new CloudEvent {
                    Id = Guid.NewGuid().ToString(),
                    Type = "com.kubemq.examples.commands.response",
                    Source = new Uri("urn:csharp-cmd-responder"),
                    Subject = replyChannel,
                    DataContentType = "application/json",
                    Data = new { executed = true, status = "command processed" },
                };
                var bytes = formatter.EncodeStructuredModeMessage(responseEvent, out var ct);
                using var rc = new ByteArrayContent(bytes.ToArray());
                rc.Headers.ContentType = MediaTypeHeaderValue.Parse(ct.ToString());
                using var sendHttp = new HttpClient();
                var r = await sendHttp.PostAsync($"{base_}/ce/send/response?request_id={requestId}", rc);
                var rj = JsonSerializer.Deserialize<JsonElement>(await r.Content.ReadAsStringAsync());
                Console.WriteLine($"[responder] response sent: is_error={rj.GetProperty("is_error")}");
                return;
            }
            evType = null; data = null;
        }
        else if (line.StartsWith("event:")) evType = line[6..].Trim();
        else if (line.StartsWith("data:")) data = line[5..].Trim();
    }
}

var responderTask = RunResponder();
await Task.Delay(500);

var cmdEvent = new CloudEvent {
    Id = Guid.NewGuid().ToString(),
    Type = "com.kubemq.examples.commands.reboot",
    Source = new Uri("urn:kubemq-ce-csharp-sender"),
    Subject = channel,
    DataContentType = "application/json",
    Data = new { device_id = "sensor-42", action = "reboot" },
};
var cmdBytes = formatter.EncodeStructuredModeMessage(cmdEvent, out var cmdCt);
using var cmdContent = new ByteArrayContent(cmdBytes.ToArray());
cmdContent.Headers.ContentType = MediaTypeHeaderValue.Parse(cmdCt.ToString());
using var senderHttp = new HttpClient();
Console.WriteLine("[sender] sending command...");
var cmdResp = await senderHttp.PostAsync($"{base_}/ce/send/command", cmdContent);
var cmdResult = JsonSerializer.Deserialize<JsonElement>(await cmdResp.Content.ReadAsStringAsync());
Console.WriteLine($"[sender] command ack: status={cmdResp.StatusCode} is_error={cmdResult.GetProperty("is_error")}");
await responderTask;

# Example: commands/round_trip — RPC command with execution ack.
require "net/http"; require "uri"; require "json"; require "timeout"; require "securerandom"; require "cloud_events"
def server_url = ENV.fetch("KUBEMQ_CE_URL", "http://localhost:9090")

base = server_url; channel = "ruby-ce-commands.round-trip"
sdk = CloudEvents::HttpBinding.default
ready = Queue.new

responder = Thread.new do
  uri = URI("#{base}/ce/subscribe/commands?client_id=ruby-cmd-responder&channel=#{URI.encode_www_form_component(channel)}")
  Net::HTTP.start(uri.host, uri.port) do |http|
    req = Net::HTTP::Get.new(uri); req["Accept"] = "text/event-stream"
    http.request(req) do |resp|
      ready.push(true) # signal that SSE connection is established
      ev_type = nil; data = nil
      resp.read_body do |chunk|
        chunk.each_line do |line|
          line.chomp!
          if line.empty?
            if ev_type == "cloudevent" && data
              raw = JSON.parse(data)
              request_id  = raw["_kubemq_request_id"]
              reply_channel = raw["_kubemq_reply_channel"]
              puts "[responder] command received: type=#{raw['type']} request_id=#{request_id}"
              resp_event = CloudEvents::Event::V1.new(
                id: SecureRandom.uuid, type: "com.kubemq.examples.commands.response",
                source: URI("urn:ruby-cmd-responder"), subject: reply_channel,
                spec_version: "1.0",
                data_content_type: CloudEvents::ContentType.new("application/json"),
                data: JSON.generate({ executed: true, status: "command processed" }))
              resp_h, resp_b = sdk.encode_event(resp_event, structured_format: "json")
              ruri = URI("#{base}/ce/send/response?request_id=#{URI.encode_www_form_component(request_id)}")
              Net::HTTP.start(ruri.host, ruri.port) do |rhttp|
                rreq = Net::HTTP::Post.new(ruri); resp_h.each{|k,v|rreq[k]=v}; rreq.body=resp_b
                rr = rhttp.request(rreq)
                puts "[responder] response sent: is_error=#{JSON.parse(rr.body)['is_error']}"
              end
              Thread.exit
            end
            ev_type = nil; data = nil
          elsif line.start_with?("event:") then ev_type = line.sub("event:","").strip
          elsif line.start_with?("data:") then data = line.sub("data:","").strip
          end
        end
      end
    end
  end
end

Timeout.timeout(5) { ready.pop }
sleep 0.1

cmd_event = CloudEvents::Event::V1.new(
  id: SecureRandom.uuid, type: "com.kubemq.examples.commands.reboot",
  source: URI("urn:kubemq-ce-ruby-sender"), subject: channel, spec_version: "1.0",
  data_content_type: CloudEvents::ContentType.new("application/json"),
  data: JSON.generate({ device_id: "sensor-42", action: "reboot" }))
cmd_h, cmd_b = sdk.encode_event(cmd_event, structured_format: "json")
uri = URI("#{base}/ce/send/command")
puts "[sender] sending command..."
Net::HTTP.start(uri.host, uri.port) do |http|
  req = Net::HTTP::Post.new(uri); cmd_h.each{|k,v|req[k]=v}; req.body=cmd_b
  res = http.request(req)
  r = JSON.parse(res.body)
  puts "[sender] command ack: status=#{res.code} is_error=#{r['is_error']}"
end
responder.join

//! Example: commands/round-trip — RPC command with execution ack.

use bytes::Bytes;
use cloudevents::{EventBuilder, EventBuilderV10};
use futures_util::StreamExt;
use reqwest::Client;
use serde_json::{json, Value};
use std::env;
use tokio::sync::oneshot;
use uuid::Uuid;

fn server_url() -> String {
    env::var("KUBEMQ_CE_URL").unwrap_or_else(|_| "http://localhost:9090".to_string())
}

async fn run_responder(base: String, channel: String, ready_tx: oneshot::Sender<()>) {
    let client = Client::new();
    let url = format!("{}/ce/subscribe/commands?client_id=rust-cmd-responder&channel={}", base, channel);
    let resp = client.get(&url)
        .header("Accept", "text/event-stream")
        .send().await.expect("SSE connect");
    // Signal ready as soon as the SSE connection is established
    let _ = ready_tx.send(());
    let mut stream = Box::pin(resp.bytes_stream());
    let mut buffer = String::new();
    let mut ev_type = String::new();
    let mut data_str = String::new();

    while let Some(chunk) = stream.next().await {
        let chunk: Bytes = chunk.unwrap();
        buffer.push_str(&String::from_utf8_lossy(&chunk));

        while let Some(pos) = buffer.find('\n') {
            let line = buffer[..pos].trim_end_matches('\r').to_string();
            buffer = buffer[pos + 1..].to_string();
            if line.is_empty() {
                if ev_type == "cloudevent" && !data_str.is_empty() {
                    let raw: Value = serde_json::from_str(&data_str).unwrap();
                    let request_id = raw["_kubemq_request_id"].as_str().unwrap_or("").to_string();
                    let reply_channel = raw["_kubemq_reply_channel"].as_str().unwrap_or("").to_string();
                    println!("[responder] command received: type={} request_id={}", raw["type"], request_id);

                    let resp_event = EventBuilderV10::new()
                        .id(Uuid::new_v4().to_string())
                        .ty("com.kubemq.examples.commands.response")
                        .source("urn:rust-cmd-responder")
                        .subject(reply_channel.as_str())
                        .data("application/json", json!({"executed": true, "status": "command processed"}))
                        .build().unwrap();
                    let body = serde_json::to_string(&resp_event).unwrap();
                    let r = client.post(format!("{}/ce/send/response?request_id={}", base, request_id))
                        .header("Content-Type", "application/cloudevents+json")
                        .body(body).send().await.unwrap();
                    let rj: Value = r.json().await.unwrap();
                    println!("[responder] response sent: is_error={}", rj["is_error"]);
                    return;
                }
                ev_type.clear(); data_str.clear();
            } else if line.starts_with(':') {
            } else if let Some(v) = line.strip_prefix("event:") { ev_type = v.trim().to_string(); }
            else if let Some(v) = line.strip_prefix("data:") { data_str = v.trim().to_string(); }
        }
    }
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let base = server_url();
    let channel = "rust-ce-commands.round-trip".to_string();
    let client = Client::new();

    let (ready_tx, ready_rx) = oneshot::channel::<()>();
    let base_clone = base.clone();
    let channel_clone = channel.clone();
    tokio::spawn(async move { run_responder(base_clone, channel_clone, ready_tx).await });

    tokio::time::timeout(tokio::time::Duration::from_secs(5), ready_rx).await??;
    tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;

    let event = EventBuilderV10::new()
        .id(Uuid::new_v4().to_string())
        .ty("com.kubemq.examples.commands.reboot")
        .source("urn:kubemq-ce-rust-sender")
        .subject(channel.as_str())
        .data("application/json", json!({"device_id": "sensor-42", "action": "reboot"}))
        .build()?;
    let body = serde_json::to_string(&event)?;
    println!("[sender] sending command...");
    let resp = client.post(format!("{}/ce/send/command", base))
        .header("Content-Type", "application/cloudevents+json")
        .body(body).send().await?;
    let result: Value = resp.json().await?;
    println!("[sender] command ack: status=202 is_error={}", result["is_error"]);
    tokio::time::sleep(tokio::time::Duration::from_millis(200)).await;
    Ok(())
}

// Example: queries/round-trip
//
// Demonstrates an RPC query round-trip:
//   - Responder subscribes via SSE GET /ce/subscribe/queries
//   - Sender sends a query via POST /ce/send/query (blocks for response)
//   - Responder extracts _kubemq_request_id and sends data response
//   - Sender receives the data payload in the query response
//
// Run: go run ./queries/round-trip/main.go
package main

import (
	"bufio"
	"encoding/json"
	"fmt"
	"log"
	"net/http"
	"os"
	"strings"
	"sync"
	"time"

	cloudevents "github.com/cloudevents/sdk-go/v2"
)

func serverURL() string {
	if u := os.Getenv("KUBEMQ_CE_URL"); u != "" {
		return u
	}
	return "http://localhost:9090"
}

type CEResponse struct {
	IsError bool            `json:"is_error"`
	Message string          `json:"message"`
	Data    json.RawMessage `json:"data"`
}

func startQueryResponder(base, channel string, ready chan<- struct{}, wg *sync.WaitGroup) {
	defer wg.Done()
	sseURL := fmt.Sprintf("%s/ce/subscribe/queries?client_id=go-query-responder&channel=%s",
		base, channel)
	req, _ := http.NewRequest("GET", sseURL, nil)
	req.Header.Set("Accept", "text/event-stream")
	req.Header.Set("Cache-Control", "no-cache")

	client := &http.Client{Timeout: 0}
	resp, err := client.Do(req)
	if err != nil {
		log.Fatal("query responder SSE:", err)
	}
	defer resp.Body.Close()

	close(ready)

	scanner := bufio.NewScanner(resp.Body)
	var evType, data string
	for scanner.Scan() {
		line := scanner.Text()
		if line == "" {
			if evType == "cloudevent" && data != "" {
				var raw map[string]interface{}
				_ = json.Unmarshal([]byte(data), &raw)

				requestID, _ := raw["_kubemq_request_id"].(string)
				replyChannel, _ := raw["_kubemq_reply_channel"].(string)

				// Extract the query payload.
				queryData, _ := raw["data"].(map[string]interface{})
				sku, _ := queryData["sku"].(string)
				fmt.Printf("[responder] query received: sku=%s request_id=%s\n", sku, requestID)

				// Send data response.
				sendQueryResponse(base, requestID, replyChannel, sku)
				return
			}
			evType, data = "", ""
			continue
		}
		if strings.HasPrefix(line, ":") {
			continue
		}
		if strings.HasPrefix(line, "event:") {
			evType = strings.TrimSpace(strings.TrimPrefix(line, "event:"))
		} else if strings.HasPrefix(line, "data:") {
			data = strings.TrimSpace(strings.TrimPrefix(line, "data:"))
		}
	}
}

func sendQueryResponse(base, requestID, replyChannel, sku string) {
	// Simulate database lookup.
	inventory := map[string]int{"WIDGET-100": 42, "GADGET-200": 7}
	qty := inventory[sku]

	event := cloudevents.NewEvent()
	event.SetType("com.kubemq.examples.queries.inventory-result")
	event.SetSource("go-query-responder")
	event.SetSubject(replyChannel)
	_ = event.SetData(cloudevents.ApplicationJSON, map[string]interface{}{
		"sku":      sku,
		"quantity": qty,
	})

	body, _ := json.Marshal(event)
	url := fmt.Sprintf("%s/ce/send/response?request_id=%s", base, requestID)
	req, _ := http.NewRequest("POST", url, strings.NewReader(string(body)))
	req.Header.Set("Content-Type", "application/cloudevents+json")

	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		log.Fatal("send query response:", err)
	}
	defer resp.Body.Close()
	fmt.Println("[responder] response sent.")
}

func sendQuery(base, channel string) {
	event := cloudevents.NewEvent()
	event.SetType("com.kubemq.examples.queries.inventory-check")
	event.SetSource("kubemq-ce-go-sender")
	event.SetSubject(channel)
	_ = event.SetData(cloudevents.ApplicationJSON, map[string]string{"sku": "WIDGET-100"})

	body, _ := json.Marshal(event)
	req, _ := http.NewRequest("POST", base+"/ce/send/query", strings.NewReader(string(body)))
	req.Header.Set("Content-Type", "application/cloudevents+json")

	fmt.Println("[sender] sending query for sku=WIDGET-100...")
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		log.Fatal("send query:", err)
	}
	defer resp.Body.Close()

	var result CEResponse
	_ = json.NewDecoder(resp.Body).Decode(&result)

	// The response data contains the CE response from the responder.
	fmt.Printf("[sender] query response: status=%d is_error=%v\n",
		resp.StatusCode, result.IsError)
	fmt.Printf("[sender] response data: %s\n", string(result.Data))
}

func main() {
	base := serverURL()
	channel := "go-ce-queries.round-trip"

	ready := make(chan struct{})
	var wg sync.WaitGroup
	wg.Add(1)
	go startQueryResponder(base, channel, ready, &wg)

	select {
	case <-ready:
	case <-time.After(5 * time.Second):
		log.Fatal("Timed out waiting for query responder")
	}
	time.Sleep(100 * time.Millisecond)

	sendQuery(base, channel)
	wg.Wait()
}

"""Example: queries/round_trip — RPC query with data response."""

from __future__ import annotations

import json
import os
import threading
import time

import requests
from cloudevents.v1.conversion import to_structured
from cloudevents.v1.http import CloudEvent


def server_url() -> str:
    return os.environ.get("KUBEMQ_CE_URL", "http://localhost:9090")


def responder(base: str, channel: str, ready: threading.Event) -> None:
    sse_url = (f"{base}/ce/subscribe/queries"
               f"?client_id=python-query-responder&channel={channel}")
    inventory = {"WIDGET-100": 42, "GADGET-200": 7}

    with requests.get(sse_url, stream=True, timeout=None,
                      headers={"Accept": "text/event-stream"}) as resp:
        ready.set()
        ev_type = data = ""
        for line in resp.iter_lines(decode_unicode=True):
            if line == "":
                if ev_type == "cloudevent" and data:
                    raw = json.loads(data)
                    request_id = raw.get("_kubemq_request_id", "")
                    reply_channel = raw.get("_kubemq_reply_channel", "")
                    query_data = raw.get("data", {})
                    sku = query_data.get("sku", "")
                    qty = inventory.get(sku, 0)
                    print(f"[responder] query sku={sku} qty={qty} request_id={request_id}")

                    response_event = CloudEvent(
                        attributes={
                            "type": "com.kubemq.examples.queries.inventory-result",
                            "source": "python-query-responder",
                            "subject": reply_channel,
                            "datacontenttype": "application/json",
                        },
                        data={"sku": sku, "quantity": qty},
                    )
                    headers, body = to_structured(response_event)
                    requests.post(
                        f"{base}/ce/send/response",
                        params={"request_id": request_id},
                        data=body, headers=dict(headers), timeout=10,
                    )
                    print("[responder] response sent.")
                    return
                ev_type = data = ""
                continue
            if line.startswith(":"):
                continue
            if line.startswith("event:"):
                ev_type = line[6:].strip()
            elif line.startswith("data:"):
                data = line[5:].strip()


def main() -> None:
    base = server_url()
    channel = "python-ce-queries.round-trip"

    ready = threading.Event()
    t = threading.Thread(target=responder, args=(base, channel, ready), daemon=True)
    t.start()

    ready.wait(timeout=5)
    time.sleep(0.1)

    event = CloudEvent(
        attributes={
            "type": "com.kubemq.examples.queries.inventory-check",
            "source": "kubemq-ce-python-sender",
            "subject": channel,
            "datacontenttype": "application/json",
        },
        data={"sku": "WIDGET-100"},
    )
    headers, body = to_structured(event)
    print("[sender] sending query for sku=WIDGET-100...")
    resp = requests.post(f"{base}/ce/send/query", data=body,
                          headers=dict(headers), timeout=30)
    result = resp.json()
    print(f"[sender] query response: status={resp.status_code} is_error={result.get('is_error')}")
    print(f"[sender] response data: {result.get('data')}")
    t.join(timeout=5)


if __name__ == "__main__":
    main()

/**
 * Example: queries/round-trip — RPC query with data response.
 * Run: npx tsx queries/round-trip/index.ts
 */
import { CloudEvent, HTTP } from 'cloudevents';
import EventSource from 'eventsource';

function serverUrl(): string {
  return process.env.KUBEMQ_CE_URL ?? 'http://localhost:9090';
}

const INVENTORY: Record<string, number> = { 'WIDGET-100': 42, 'GADGET-200': 7 };

function startQueryResponder(base: string, channel: string): Promise<void> {
  return new Promise((resolve) => {
    const url = `${base}/ce/subscribe/queries?client_id=js-query-responder&channel=${encodeURIComponent(channel)}`;
    const es = new EventSource(url);
    es.addEventListener('cloudevent', async (evt: MessageEvent) => {
      es.close();
      const raw = JSON.parse(evt.data) as Record<string, unknown>;
      const requestId = raw._kubemq_request_id as string;
      const replyChannel = raw._kubemq_reply_channel as string;
      const queryData = raw.data as Record<string, string>;
      const sku = queryData.sku;
      const qty = INVENTORY[sku] ?? 0;
      console.log(`[responder] query sku=${sku} qty=${qty} request_id=${requestId}`);

      const response = new CloudEvent({
        type: 'com.kubemq.examples.queries.inventory-result',
        source: 'js-query-responder',
        subject: replyChannel,
        datacontenttype: 'application/json',
        data: { sku, quantity: qty },
      });
      const msg = HTTP.structured(response);
      await fetch(`${base}/ce/send/response?request_id=${requestId}`, {
        method: 'POST',
        headers: msg.headers as Record<string, string>,
        body: msg.body as string,
      });
      console.log('[responder] response sent.');
      resolve();
    });
    es.addEventListener('error', (evt: MessageEvent) => {
      if (evt.data) {
        const err = JSON.parse(evt.data) as { message: string };
        console.error('[responder] SSE error:', err.message);
        es.close();
      }
    });
  });
}

async function main(): Promise<void> {
  const base = serverUrl();
  const channel = 'js-ce-queries.round-trip';

  const responderDone = startQueryResponder(base, channel);
  await new Promise((r) => setTimeout(r, 500));

  const event = new CloudEvent({
    type: 'com.kubemq.examples.queries.inventory-check',
    source: 'kubemq-ce-js-sender',
    subject: channel,
    datacontenttype: 'application/json',
    data: { sku: 'WIDGET-100' },
  });
  const msg = HTTP.structured(event);
  console.log('[sender] sending query for sku=WIDGET-100...');
  const resp = await fetch(`${base}/ce/send/query`, {
    method: 'POST',
    headers: msg.headers as Record<string, string>,
    body: msg.body as string,
  });
  const result = await resp.json() as { is_error: boolean; data: unknown };
  console.log(`[sender] query response: status=${resp.status} is_error=${result.is_error}`);
  console.log(`[sender] response data: ${JSON.stringify(result.data)}`);

  await responderDone;
}

main().catch(console.error);

package io.kubemq.examples.queries.roundtrip;

import com.fasterxml.jackson.databind.ObjectMapper;
import io.cloudevents.CloudEvent;
import io.cloudevents.core.builder.CloudEventBuilder;
import io.cloudevents.core.format.EventFormat;
import io.cloudevents.core.provider.EventFormatProvider;
import io.cloudevents.jackson.JsonFormat;

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.net.HttpURLConnection;
import java.net.URI;
import java.net.URL;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.nio.charset.StandardCharsets;
import java.time.OffsetDateTime;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * Example: queries/round-trip
 *
 * A responder subscribes to queries, receives a query, looks up inventory,
 * and sends a CE response. A sender publishes a query via POST /ce/send/query
 * and prints the response returned directly in the HTTP response body.
 *
 * Run: mvn compile exec:java
 */
public class Main {
    static String serverUrl() {
        String u = System.getenv("KUBEMQ_CE_URL");
        return (u != null && !u.isEmpty()) ? u : "http://localhost:9090";
    }
    static final ObjectMapper MAPPER = new ObjectMapper();
    static final Map<String, Integer> INVENTORY = Map.of("WIDGET-100", 42, "GADGET-200", 7);

    public static void main(String[] args) throws Exception {
        String base = serverUrl();
        String channel = "java-ce-queries.round-trip";

        EventFormatProvider.getInstance().registerFormat(new JsonFormat());
        EventFormat format = EventFormatProvider.getInstance().resolveFormat(JsonFormat.CONTENT_TYPE);
        HttpClient httpClient = HttpClient.newHttpClient();

        BlockingQueue<Boolean> responderReady = new ArrayBlockingQueue<>(1);

        // Start query responder.
        String sseUrl = base + "/ce/subscribe/queries?client_id=java-query-responder&channel=" + channel;
        Thread.ofVirtual().start(() -> {
            try {
                HttpURLConnection conn = (HttpURLConnection) new URL(sseUrl).openConnection();
                conn.setRequestProperty("Accept", "text/event-stream");
                conn.setReadTimeout(30_000);
                try (BufferedReader reader = new BufferedReader(
                        new InputStreamReader(conn.getInputStream(), StandardCharsets.UTF_8))) {
                    String line; String evType = null, data = null;
                    boolean ready = false;
                    while ((line = reader.readLine()) != null) {
                        if (!ready) { responderReady.offer(true); ready = true; }
                        if (line.isEmpty()) {
                            if ("cloudevent".equals(evType) && data != null) {
                                Map<?, ?> raw = MAPPER.readValue(data, Map.class);
                                String requestId = (String) raw.get("_kubemq_request_id");
                                String replyChannel = (String) raw.get("_kubemq_reply_channel");
                                Map<?, ?> qdata = (Map<?, ?>) raw.get("data");
                                String sku = qdata != null ? (String) qdata.get("sku") : "";
                                int qty = INVENTORY.getOrDefault(sku, 0);
                                System.out.println("[responder] query sku=" + sku + " qty=" + qty);

                                CloudEvent respEvent = CloudEventBuilder.v1()
                                        .withId(UUID.randomUUID().toString())
                                        .withType("com.kubemq.examples.queries.inventory-result")
                                        .withSource(URI.create("kubemq-ce-java-responder"))
                                        .withSubject(replyChannel)
                                        .withDataContentType("application/json")
                                        .withTime(OffsetDateTime.now())
                                        .withData("application/json",
                                                MAPPER.writeValueAsBytes(Map.of("sku", sku, "quantity", qty)))
                                        .build();
                                String responseUrl = base + "/ce/send/response?request_id=" + requestId;
                                httpClient.send(
                                        HttpRequest.newBuilder().uri(URI.create(responseUrl))
                                                .POST(HttpRequest.BodyPublishers.ofByteArray(format.serialize(respEvent)))
                                                .header("Content-Type", "application/cloudevents+json").build(),
                                        HttpResponse.BodyHandlers.ofString());
                                System.out.println("[responder] response sent.");
                                return;
                            }
                            evType = null; data = null;
                        } else if (line.startsWith("event:")) evType = line.substring(6).trim();
                        else if (line.startsWith("data:")) data = line.substring(5).trim();
                    }
                }
            } catch (Exception e) { System.err.println("SSE error: " + e.getMessage()); }
        });

        responderReady.poll(5, TimeUnit.SECONDS);
        Thread.sleep(100);

        // Send query.
        CloudEvent query = CloudEventBuilder.v1()
                .withId(UUID.randomUUID().toString())
                .withType("com.kubemq.examples.queries.inventory-check")
                .withSource(URI.create("kubemq-ce-java-sender"))
                .withSubject(channel)
                .withDataContentType("application/json")
                .withTime(OffsetDateTime.now())
                .withData("application/json", MAPPER.writeValueAsBytes(Map.of("sku", "WIDGET-100")))
                .build();

        System.out.println("[sender] sending query for sku=WIDGET-100...");
        HttpResponse<String> resp = httpClient.send(
                HttpRequest.newBuilder().uri(URI.create(base + "/ce/send/query"))
                        .POST(HttpRequest.BodyPublishers.ofByteArray(format.serialize(query)))
                        .header("Content-Type", "application/cloudevents+json").build(),
                HttpResponse.BodyHandlers.ofString());

        Map<?, ?> result = MAPPER.readValue(resp.body(), Map.class);
        System.out.println("[sender] query response: status=" + resp.statusCode() + " is_error=" + result.get("is_error"));
        System.out.println("[sender] response data: " + result.get("data"));
    }
}

// Example: queries/RoundTrip — RPC query with data response.
// Run: dotnet run
using CloudNative.CloudEvents;
using CloudNative.CloudEvents.SystemTextJson;
using System.Net.Http.Headers;
using System.Text;
using System.Text.Json;

static string ServerUrl() => Environment.GetEnvironmentVariable("KUBEMQ_CE_URL") ?? "http://localhost:9090";

var base_ = ServerUrl();
var channel = "csharp-ce-queries.round-trip";
var formatter = new JsonEventFormatter();
var inventory = new Dictionary<string, int> { ["WIDGET-100"] = 42, ["GADGET-200"] = 7 };
var responderDone = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);

async Task RunResponder()
{
    using var http = new HttpClient { Timeout = Timeout.InfiniteTimeSpan };
    var url = $"{base_}/ce/subscribe/queries?client_id=csharp-query-responder&channel={Uri.EscapeDataString(channel)}";
    using var req = new HttpRequestMessage(HttpMethod.Get, url);
    req.Headers.Accept.Add(new MediaTypeWithQualityHeaderValue("text/event-stream"));
    using var resp = await http.SendAsync(req, HttpCompletionOption.ResponseHeadersRead);
    using var reader = new StreamReader(await resp.Content.ReadAsStreamAsync(), Encoding.UTF8);
    string? evType = null, data = null, line;
    while ((line = await reader.ReadLineAsync()) != null)
    {
        if (line == "") {
            if (evType == "cloudevent" && data != null) {
                var raw = JsonSerializer.Deserialize<JsonElement>(data);
                var requestId = raw.GetProperty("_kubemq_request_id").GetString()!;
                var replyChannel = raw.GetProperty("_kubemq_reply_channel").GetString()!;
                var sku = raw.GetProperty("data").GetProperty("sku").GetString()!;
                var qty = inventory.GetValueOrDefault(sku, 0);
                Console.WriteLine($"[responder] query sku={sku} qty={qty} request_id={requestId}");

                var respEvent = new CloudEvent {
                    Id = Guid.NewGuid().ToString(),
                    Type = "com.kubemq.examples.queries.inventory-result",
                    Source = new Uri("urn:csharp-query-responder"),
                    Subject = replyChannel,
                    DataContentType = "application/json",
                    Data = new { sku, quantity = qty },
                };
                var bytes = formatter.EncodeStructuredModeMessage(respEvent, out var ct);
                using var rc = new ByteArrayContent(bytes.ToArray());
                rc.Headers.ContentType = MediaTypeHeaderValue.Parse(ct.ToString());
                using var sendHttp = new HttpClient();
                await sendHttp.PostAsync($"{base_}/ce/send/response?request_id={requestId}", rc);
                Console.WriteLine("[responder] response sent.");
                responderDone.TrySetResult();
                return;
            }
            evType = null; data = null;
        }
        else if (line.StartsWith("event:")) evType = line[6..].Trim();
        else if (line.StartsWith("data:")) data = line[5..].Trim();
    }
}

var responderTask = RunResponder();
await Task.Delay(500);

// Send query.
var query = new CloudEvent {
    Id = Guid.NewGuid().ToString(),
    Type = "com.kubemq.examples.queries.inventory-check",
    Source = new Uri("urn:kubemq-ce-csharp-sender"),
    Subject = channel,
    DataContentType = "application/json",
    Data = new { sku = "WIDGET-100" },
};
var qBytes = formatter.EncodeStructuredModeMessage(query, out var qCt);
using var qContent = new ByteArrayContent(qBytes.ToArray());
qContent.Headers.ContentType = MediaTypeHeaderValue.Parse(qCt.ToString());
using var senderHttp = new HttpClient();
Console.WriteLine("[sender] sending query for sku=WIDGET-100...");
var queryResp = await senderHttp.PostAsync($"{base_}/ce/send/query", qContent);
var queryJ = JsonSerializer.Deserialize<JsonElement>(await queryResp.Content.ReadAsStringAsync());
Console.WriteLine($"[sender] query response: status={queryResp.StatusCode} is_error={queryJ.GetProperty("is_error")}");
Console.WriteLine($"[sender] response data: {queryJ.GetProperty("data")}");
await responderDone.Task;

# Example: queries/round_trip — RPC query with data response.
require "net/http"; require "uri"; require "json"; require "timeout"; require "securerandom"; require "cloud_events"
def server_url = ENV.fetch("KUBEMQ_CE_URL", "http://localhost:9090")

base = server_url; channel = "ruby-ce-queries.round-trip"
sdk = CloudEvents::HttpBinding.default
inventory = { "WIDGET-100" => 42, "GADGET-200" => 7 }
ready = Queue.new

responder = Thread.new do
  uri = URI("#{base}/ce/subscribe/queries?client_id=ruby-query-responder&channel=#{URI.encode_www_form_component(channel)}")
  Net::HTTP.start(uri.host, uri.port) do |http|
    req = Net::HTTP::Get.new(uri); req["Accept"] = "text/event-stream"
    http.request(req) do |resp|
      ready.push(true) # signal that SSE connection is established
      ev_type = nil; data = nil
      resp.read_body do |chunk|
        chunk.each_line do |line|
          line.chomp!
          if line.empty?
            if ev_type == "cloudevent" && data
              raw = JSON.parse(data)
              request_id = raw["_kubemq_request_id"]
              reply_channel = raw["_kubemq_reply_channel"]
              raw_data = raw["data"]
              raw_data = JSON.parse(raw_data) if raw_data.is_a?(String)
              sku = raw_data.is_a?(Hash) ? raw_data["sku"] : nil
              qty = inventory[sku] || 0
              puts "[responder] query sku=#{sku} qty=#{qty} request_id=#{request_id}"
              resp_ev = CloudEvents::Event::V1.new(
                id: SecureRandom.uuid,
                type: "com.kubemq.examples.queries.inventory-result",
                source: URI("urn:ruby-query-responder"), subject: reply_channel,
                spec_version: "1.0",
                data_content_type: CloudEvents::ContentType.new("application/json"),
                data: JSON.generate({ sku: sku, quantity: qty }))
              resp_h, resp_b = sdk.encode_event(resp_ev, structured_format: "json")
              ruri = URI("#{base}/ce/send/response?request_id=#{URI.encode_www_form_component(request_id)}")
              Net::HTTP.start(ruri.host, ruri.port) do |rhttp|
                rreq = Net::HTTP::Post.new(ruri); resp_h.each{|k,v|rreq[k]=v}; rreq.body=resp_b
                rhttp.request(rreq)
              end
              puts "[responder] response sent."
              Thread.exit
            end
            ev_type = nil; data = nil
          elsif line.start_with?("event:") then ev_type = line.sub("event:","").strip
          elsif line.start_with?("data:") then data = line.sub("data:","").strip
          end
        end
      end
    end
  end
end

Timeout.timeout(5) { ready.pop }; sleep 0.1

ev = CloudEvents::Event::V1.new(
  id: SecureRandom.uuid, type: "com.kubemq.examples.queries.inventory-check",
  source: URI("urn:kubemq-ce-ruby-sender"), subject: channel, spec_version: "1.0",
  data_content_type: CloudEvents::ContentType.new("application/json"),
  data: JSON.generate({ sku: "WIDGET-100" }))
ev_h, ev_b = sdk.encode_event(ev, structured_format: "json")
uri = URI("#{base}/ce/send/query")
puts "[sender] sending query for sku=WIDGET-100..."
Net::HTTP.start(uri.host, uri.port) do |http|
  req = Net::HTTP::Post.new(uri); ev_h.each{|k,v|req[k]=v}; req.body=ev_b
  res = http.request(req)
  r = JSON.parse(res.body)
  puts "[sender] query response: status=#{res.code} is_error=#{r['is_error']}"
  puts "[sender] response data: #{r['data']}"
end
responder.join