# Commands & Queries (/connectors/cloudevents/how-to/commands-queries)



Commands and queries give you **synchronous request-response (RPC)** over the CloudEvents HTTP interface. The sender publishes a CloudEvent and **blocks** until a responder processes the request and replies. Use it when the caller needs confirmation that an action ran (a command) or needs data back before continuing (a query).

## Overview [#overview]

A responder subscribes to a command or query channel over SSE, receives each request, and sends a CloudEvent response back through `POST /ce/send/response`, correlated by a request ID. The sender's original `POST` stays open and returns the result once the responder replies.

The two operations differ only in what comes back:

|                           | Commands                               | Queries                         |
| ------------------------- | -------------------------------------- | ------------------------------- |
| **Intent**                | Execute an action, confirm it ran      | Request data, get a result back |
| **Response payload**      | Acknowledgment (e.g. `executed: true`) | Data (e.g. an inventory count)  |
| **Send endpoint**         | `POST /ce/send/command`                | `POST /ce/send/query`           |
| **Subscribe endpoint**    | `GET /ce/subscribe/commands`           | `GET /ce/subscribe/queries`     |
| **Sender success status** | `202`                                  | `200`                           |

Both are timed out by the connector's `TimeoutSeconds` (default 60s) — if no responder replies in time, the send returns HTTP 504.

## How it works [#how-it-works]

The responder subscribes first; the sender then issues a blocking `POST`, the connector delivers the request over SSE, and the responder posts a correlated response that the connector routes back to the waiting sender.

<Mermaid
  chart="`
sequenceDiagram
participant Sender
participant CE as CloudEvents connector
participant KMQ as KubeMQ broker
participant Responder
Responder->>CE: GET /ce/subscribe/commands (SSE)
CE->>KMQ: subscribe
Sender->>CE: POST /ce/send/command (blocks)
CE->>KMQ: route request
KMQ-->>CE: deliver
CE-->>Responder: event: cloudevent
Note over Responder: read _kubemq_request_id<br/>and _kubemq_reply_channel
Responder->>CE: POST /ce/send/response?request_id=...
CE->>KMQ: correlate response
KMQ-->>CE: deliver to sender
CE-->>Sender: 202 / 200 (ack or data)
`"
/>

*A command round-trip: the responder subscribes, the sender blocks, and the response is correlated back by request ID.*

## Endpoints [#endpoints]

| Method | Endpoint                                       | Description                                     | Success status |
| ------ | ---------------------------------------------- | ----------------------------------------------- | -------------- |
| `POST` | `/ce/send/command`                             | Send a command; blocks until a response arrives | `202`          |
| `POST` | `/ce/send/query`                               | Send a query; blocks until a response arrives   | `200`          |
| `GET`  | `/ce/subscribe/commands?client_id=X&channel=Y` | Subscribe to commands via SSE                   | SSE stream     |
| `GET`  | `/ce/subscribe/queries?client_id=X&channel=Y`  | Subscribe to queries via SSE                    | SSE stream     |
| `POST` | `/ce/send/response?request_id=X`               | Send a response to a received command or query  | `202`          |

## Response correlation [#response-correlation]

When a command or query is delivered over SSE, the connector adds two correlation fields the responder must echo back. For **CloudEvent** messages (`event: cloudevent`), they are merged into the CE JSON with an underscore prefix:

* **`_kubemq_request_id`** — pass this as the `request_id` query parameter on `POST /ce/send/response`.
* **`_kubemq_reply_channel`** — set this as the `subject` of the response CloudEvent so it routes back to the original sender.

For non-CE messages (`event: message`), the same values appear as top-level `request_id` and `reply_channel` keys without the `_kubemq_` prefix. See [CE-to-KubeMQ mapping](/connectors/cloudevents/reference/ce-to-kubemq-mapping) for the full attribute table.

## Command round-trip [#command-round-trip]

A complete command cycle: the responder subscribes over SSE, receives the command, and posts an acknowledgment. The sender blocks until the ack arrives.

**curl** — in one terminal, subscribe to the command channel:

```bash
curl -N "http://localhost:9090/ce/subscribe/commands?client_id=responder&channel=device-commands"
```

In a second terminal, send a command. The connector blocks the response until the responder replies:

```bash
curl -X POST http://localhost:9090/ce/send/command \
  -H "Content-Type: application/cloudevents+json" \
  -d '{
    "specversion": "1.0",
    "type": "com.example.device.reboot",
    "source": "control-plane",
    "subject": "device-commands",
    "data": {"device_id": "sensor-42", "action": "reboot"}
  }'
```

The subscriber receives the command with `_kubemq_request_id`. Use it to send the response back:

```bash
curl -X POST "http://localhost:9090/ce/send/response?request_id=THE_REQUEST_ID" \
  -H "Content-Type: application/cloudevents+json" \
  -d '{
    "specversion": "1.0",
    "type": "com.example.device.reboot.result",
    "source": "device-agent",
    "subject": "device-commands",
    "data": {"executed": true, "status": "rebooting"}
  }'
```

The following examples run the responder and sender together, using the CloudEvents SDK to build and encode each event:

<Tabs groupId="language" items="['Go','Python','JavaScript','Java','C#','Ruby','Rust']">
  <Tab value="Go">
    ```go
    // 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()
    }
    ```
  </Tab>

  <Tab value="Python">
    ```python
    """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()
    ```
  </Tab>

  <Tab value="JavaScript">
    ```typescript
    /**
     * 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);
    ```
  </Tab>

  <Tab value="Java">
    ```java
    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"));
        }
    }
    ```
  </Tab>

  <Tab value="C#">
    ```csharp
    // 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;
    ```
  </Tab>

  <Tab value="Ruby">
    ```ruby
    # 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
    ```
  </Tab>

  <Tab value="Rust">
    ```rust
    //! 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(())
    }
    ```
  </Tab>
</Tabs>

## Query round-trip [#query-round-trip]

A complete query cycle: the responder subscribes over SSE, receives the query, performs a lookup, and sends back a **data response**. The sender blocks and receives the data payload directly in the HTTP response.

**curl** — in one terminal, subscribe to the query channel:

```bash
curl -N "http://localhost:9090/ce/subscribe/queries?client_id=responder&channel=inventory-queries"
```

In a second terminal, send a query. The connector returns `200` with the responder's reply in `data`:

```bash
curl -X POST http://localhost:9090/ce/send/query \
  -H "Content-Type: application/cloudevents+json" \
  -d '{
    "specversion": "1.0",
    "type": "com.example.inventory.check",
    "source": "web-frontend",
    "subject": "inventory-queries",
    "data": {"sku": "WIDGET-100"}
  }'
```

The subscriber receives the query with `_kubemq_request_id` and replies with the result:

```bash
curl -X POST "http://localhost:9090/ce/send/response?request_id=THE_REQUEST_ID" \
  -H "Content-Type: application/cloudevents+json" \
  -d '{
    "specversion": "1.0",
    "type": "com.example.inventory.result",
    "source": "inventory-service",
    "subject": "inventory-queries",
    "data": {"sku": "WIDGET-100", "quantity": 42}
  }'
```

The following examples run a responder that looks up an inventory and a sender that prints the returned data:

<Tabs groupId="language" items="['Go','Python','JavaScript','Java','C#','Ruby','Rust']">
  <Tab value="Go">
    ```go
    // 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()
    }
    ```
  </Tab>

  <Tab value="Python">
    ```python
    """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()
    ```
  </Tab>

  <Tab value="JavaScript">
    ```typescript
    /**
     * 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);
    ```
  </Tab>

  <Tab value="Java">
    ```java
    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"));
        }
    }
    ```
  </Tab>

  <Tab value="C#">
    ```csharp
    // 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;
    ```
  </Tab>

  <Tab value="Ruby">
    ```ruby
    # 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
    ```
  </Tab>

  <Tab value="Rust">
    ```rust
    //! Example: queries/round-trip
    //!
    //! Responder subscribes to queries, receives a query, sends a CE response.
    //! Sender publishes via POST /ce/send/query (blocks until response arrives).
    //!
    //! Run: cargo run -p round-trip-queries

    use bytes::Bytes;
    use cloudevents::{EventBuilder, EventBuilderV10};
    use futures_util::StreamExt;
    use reqwest::Client;
    use serde_json::{json, Value};
    use std::{collections::HashMap, 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/queries?client_id=rust-query-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();
        let inventory: HashMap<&str, u32> = [("WIDGET-100", 42), ("GADGET-200", 7)].into();

        while let Some(chunk) = stream.next().await {
            let chunk: Bytes = chunk.unwrap_or_default();
            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();
                        let sku = raw["data"]["sku"].as_str().unwrap_or("");
                        let qty = *inventory.get(sku).unwrap_or(&0);
                        println!("[responder] query sku={} qty={} request_id={}", sku, qty, request_id);

                        let resp_event = EventBuilderV10::new()
                            .id(Uuid::new_v4().to_string())
                            .ty("com.kubemq.examples.queries.inventory-result")
                            .source("urn:rust-query-responder")
                            .subject(reply_channel.as_str())
                            .data("application/json", json!({"sku": sku, "quantity": qty}))
                            .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-queries.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.queries.inventory-check")
            .source("urn:kubemq-ce-rust-sender")
            .subject(channel.as_str())
            .data("application/json", json!({"sku": "WIDGET-100"}))
            .build()?;
        let body = serde_json::to_string(&event)?;
        println!("[sender] sending query for sku=WIDGET-100...");
        let resp = client.post(format!("{}/ce/send/query", base))
            .header("Content-Type", "application/cloudevents+json")
            .body(body).send().await?;
        let result: Value = resp.json().await?;
        println!("[sender] query response: status=200 is_error={}", result["is_error"]);
        println!("[sender] response data: {}", result["data"]);
        tokio::time::sleep(tokio::time::Duration::from_millis(200)).await;
        Ok(())
    }
    ```
  </Tab>
</Tabs>

<Callout type="warn">
  Commands and queries are synchronous. The responder must be online and reply within the connector's `TimeoutSeconds` (default 60s), or the send fails with HTTP 504.
</Callout>

<Callout type="info">
  The `?group=` load-balancing parameter is only available for [Events](/connectors/cloudevents/how-to/events) subscriptions. Command, query, and events-store subscriptions do not accept `group`.
</Callout>

## Related [#related]

<Cards>
  <Card title="Endpoints reference" href="/connectors/cloudevents/reference/endpoints" description="Full CloudEvents endpoint table and status codes." />

  <Card title="CE-to-KubeMQ mapping" href="/connectors/cloudevents/reference/ce-to-kubemq-mapping" description="Attribute mapping and command/query correlation fields." />

  <Card title="SSE behavior" href="/connectors/cloudevents/how-to/sse-behavior" description="Wire format, keepalive, and idle timeout for subscriptions." />

  <Card title="Authentication" href="/connectors/cloudevents/how-to/authentication" description="JWT auth and ClientID resolution for /ce/* endpoints." />
</Cards>