# Events (/connectors/cloudevents/how-to/events)



Events are **fire-and-forget** pub/sub messages over the CloudEvents connector. A publisher sends a CloudEvent with `POST /ce/send/event` and returns immediately; every subscriber connected to the channel receives every event over an SSE stream.

## Overview [#overview]

The Events pattern maps the KubeMQ **Events** messaging primitive onto plain HTTP. Publishers do not wait for delivery confirmation, and events are **not persisted** — a subscriber must be connected at the moment of delivery to receive a message. Connect every subscriber to a channel and you get **fan-out** (each one receives every event); add a shared `group` and you get a **consumer group** where the server load-balances each event to exactly one member.

Use this pattern for real-time notifications, telemetry, and broadcast scenarios where missed messages are acceptable. When you need durability and replay, use [Events Store](/connectors/cloudevents/how-to/events-store) instead.

| Operation | Endpoint               | Method    |
| --------- | ---------------------- | --------- |
| Publish   | `/ce/send/event`       | POST      |
| Subscribe | `/ce/subscribe/events` | GET (SSE) |

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

A publisher posts one CloudEvent to the connector, which fans it out to every connected subscriber on the channel; subscribers in a shared group split the load round-robin.

<Mermaid
  chart="`
graph LR
PUB[&#x22;Publisher&#x22;]
CE[&#x22;CloudEvents connector<br/>:9090&#x22;]
CH{{&#x22;Events channel<br/>orders&#x22;}}
S1[&#x22;Subscriber A&#x22;]
S2[&#x22;Subscriber B&#x22;]
GRP[&#x22;Group 'workers'<br/>(one member receives)&#x22;]

PUB -- &#x22;POST /ce/send/event&#x22; --> CE
CE --> CH
CH -. &#x22;SSE: event: cloudevent&#x22; .-> S1
CH -. &#x22;SSE: event: cloudevent&#x22; .-> S2
CH -. &#x22;round-robin&#x22; .-> GRP

class CE connector
class CH events
class PUB,S1,S2 client
class GRP external
`"
/>

*A published CloudEvent fans out to all subscribers; a shared `group` load-balances it to one member.*

## Publishing [#publishing]

Send a CloudEvent in **structured mode** (`application/cloudevents+json`). The `subject` attribute selects the KubeMQ channel; the connector replies with HTTP `202 Accepted` and a `{"is_error": false, ...}` body. Each native example below opens an SSE subscriber, publishes one event, prints it, and exits.

<Tabs groupId="language" items="['curl','C#','Go','Java','JavaScript','Python','Ruby','Rust']">
  <Tab value="curl">
    ```bash
    curl -X POST http://localhost:9090/ce/send/event \
      -H "Content-Type: application/cloudevents+json" \
      -d '{
        "specversion": "1.0",
        "type": "com.example.order.created",
        "source": "order-service",
        "subject": "orders",
        "datacontenttype": "application/json",
        "data": {"order_id": "12345", "amount": 99.99}
      }'
    ```
  </Tab>

  <Tab value="C#">
    ```csharp
    // Example: events/BasicPubSub
    // 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-events.basic-pubsub";
    var clientId = "kubemq-ce-csharp-example";

    var received = new TaskCompletionSource<JsonElement>(
        TaskCreationOptions.RunContinuationsAsynchronously);

    // Start SSE subscriber.
    var subscriberTask = Task.Run(async () =>
    {
        using var httpClient = new HttpClient { Timeout = Timeout.InfiniteTimeSpan };
        var sseUrl = $"{base_}/ce/subscribe/events?client_id={clientId}-sub&channel={Uri.EscapeDataString(channel)}";
        using var request = new HttpRequestMessage(HttpMethod.Get, sseUrl);
        request.Headers.Accept.Add(new MediaTypeWithQualityHeaderValue("text/event-stream"));
        request.Headers.CacheControl = new CacheControlHeaderValue { NoCache = true };

        using var response = await httpClient.SendAsync(request, HttpCompletionOption.ResponseHeadersRead);
        using var stream = await response.Content.ReadAsStreamAsync();
        using var reader = new StreamReader(stream, Encoding.UTF8);

        string? eventType = null, data = null;
        string? line;
        while ((line = await reader.ReadLineAsync()) != null)
        {
            if (line == "")
            {
                if (eventType == "cloudevent" && data != null)
                {
                    received.TrySetResult(JsonSerializer.Deserialize<JsonElement>(data));
                    return;
                }
                eventType = null; data = null;
            }
            else if (line.StartsWith(":")) { /* keepalive */ }
            else if (line.StartsWith("event:")) eventType = line["event:".Length..].Trim();
            else if (line.StartsWith("data:")) data = line["data:".Length..].Trim();
        }
    });

    // Allow subscription to establish.
    await Task.Delay(500);

    // Build and publish CloudEvent (structured mode).
    var formatter = new JsonEventFormatter();
    var cloudEvent = new CloudEvent
    {
        Id = Guid.NewGuid().ToString(),
        Type = "com.kubemq.examples.events.sent",
        Source = new Uri($"urn:{clientId}"),
        Subject = channel,
        DataContentType = "application/json",
        Data = new { message = "Hello from C# CloudEvents example!" },
    };
    cloudEvent.SetAttributeFromString("time", DateTimeOffset.UtcNow.ToString("O"));

    var eventBytes = formatter.EncodeStructuredModeMessage(cloudEvent, out var contentType);
    using var httpClient = new HttpClient();
    using var content = new ByteArrayContent(eventBytes.ToArray());
    content.Headers.ContentType = MediaTypeHeaderValue.Parse(contentType.ToString());

    var resp = await httpClient.PostAsync($"{base_}/ce/send/event", content);
    var resultJson = await resp.Content.ReadAsStringAsync();
    using var resultDoc = JsonDocument.Parse(resultJson);
    Console.WriteLine($"Published: status={resp.StatusCode} is_error={resultDoc.RootElement.GetProperty("is_error")}");

    // Wait for the event.
    var ce = await received.Task;
    Console.WriteLine($"Received: {ce.GetProperty("type")} / {ce.GetProperty("data")}");
    ```
  </Tab>

  <Tab value="Go">
    ```go
    // Example: events/basic-pubsub
    // Run: go run ./events/basic-pubsub/main.go
    package main

    import (
    	"bufio"
    	"context"
    	"encoding/json"
    	"fmt"
    	"log"
    	"net/http"
    	"os"
    	"strings"
    	"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"
    }

    func main() {
    	base := serverURL()
    	channel := "go-ce-events.basic-pubsub"
    	clientID := "kubemq-ce-go-example"

    	received := make(chan string, 1)

    	// Start SSE subscriber in background goroutine.
    	go func() {
    		sseURL := fmt.Sprintf("%s/ce/subscribe/events?client_id=%s&channel=%s",
    			base, clientID+"-sub", channel)
    		req, _ := http.NewRequest("GET", sseURL, nil)
    		req.Header.Set("Accept", "text/event-stream")
    		req.Header.Set("Cache-Control", "no-cache")

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

    		scanner := bufio.NewScanner(resp.Body)
    		var eventType, data string
    		for scanner.Scan() {
    			line := scanner.Text()
    			if line == "" {
    				if eventType == "cloudevent" && data != "" {
    					received <- data
    					return
    				}
    				eventType, data = "", ""
    				continue
    			}
    			if strings.HasPrefix(line, ":") {
    				continue // keepalive
    			}
    			if strings.HasPrefix(line, "event:") {
    				eventType = strings.TrimSpace(strings.TrimPrefix(line, "event:"))
    			} else if strings.HasPrefix(line, "data:") {
    				data = strings.TrimSpace(strings.TrimPrefix(line, "data:"))
    			}
    		}
    	}()

    	// Allow SSE subscription to establish.
    	time.Sleep(500 * time.Millisecond)

    	// Build and send CloudEvent (structured mode).
    	event := cloudevents.NewEvent()
    	event.SetType("com.kubemq.examples.events.sent")
    	event.SetSource("kubemq-ce-go-example")
    	event.SetSubject(channel) // subject = KubeMQ channel
    	_ = event.SetData(cloudevents.ApplicationJSON, map[string]string{
    		"message": "Hello from Go CloudEvents example!",
    	})

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

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

    	var result map[string]interface{}
    	_ = json.NewDecoder(resp.Body).Decode(&result)
    	fmt.Printf("Published: status=%d is_error=%v\n", resp.StatusCode, result["is_error"])

    	// Wait for subscriber to receive the event.
    	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
    	defer cancel()

    	select {
    	case data := <-received:
    		var ce map[string]interface{}
    		_ = json.Unmarshal([]byte(data), &ce)
    		fmt.Printf("Received: type=%v subject=%v data=%v\n", ce["type"], ce["subject"], ce["data"])
    	case <-ctx.Done():
    		log.Fatal("Timed out waiting for event")
    	}
    }
    ```
  </Tab>

  <Tab value="Java">
    ```java
    // Example: events/basic-pubsub
    // Run: mvn compile exec:java
    package io.kubemq.examples.events.basicpubsub;

    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;

    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-events.basic-pubsub";
            String clientId = "kubemq-ce-java-example";

            BlockingQueue<String> received = new ArrayBlockingQueue<>(1);

            // Start SSE subscriber in background thread.
            String sseUrl = base + "/ce/subscribe/events?client_id=" + clientId
                    + "-sub&channel=" + channel;
            Thread subscriber = Thread.ofVirtual().start(() -> {
                try {
                    HttpURLConnection conn = (HttpURLConnection) new URL(sseUrl).openConnection();
                    conn.setRequestMethod("GET");
                    conn.setRequestProperty("Accept", "text/event-stream");
                    conn.setRequestProperty("Cache-Control", "no-cache");
                    conn.setDoInput(true);
                    conn.setReadTimeout(15_000);

                    try (BufferedReader reader = new BufferedReader(
                            new InputStreamReader(conn.getInputStream(), StandardCharsets.UTF_8))) {
                        String line;
                        String eventType = null, data = null;
                        while ((line = reader.readLine()) != null) {
                            if (line.isEmpty()) {
                                if ("cloudevent".equals(eventType) && data != null) {
                                    received.offer(data);
                                    return;
                                }
                                eventType = null;
                                data = null;
                            } else if (line.startsWith(":")) {
                                // keepalive
                            } else if (line.startsWith("event:")) {
                                eventType = line.substring("event:".length()).trim();
                            } else if (line.startsWith("data:")) {
                                data = line.substring("data:".length()).trim();
                            }
                        }
                    }
                } catch (Exception e) {
                    System.err.println("SSE error: " + e.getMessage());
                }
            });

            // Allow subscription to establish.
            Thread.sleep(500);

            // Build CloudEvent (structured mode).
            EventFormatProvider.getInstance().registerFormat(new JsonFormat());
            EventFormat format = EventFormatProvider.getInstance().resolveFormat(JsonFormat.CONTENT_TYPE);

            CloudEvent event = CloudEventBuilder.v1()
                    .withId(UUID.randomUUID().toString())
                    .withType("com.kubemq.examples.events.sent")
                    .withSource(URI.create(clientId))
                    .withSubject(channel)
                    .withDataContentType("application/json")
                    .withTime(OffsetDateTime.now())
                    .withData("application/json",
                            MAPPER.writeValueAsBytes(Map.of("message", "Hello from Java CloudEvents example!")))
                    .build();

            byte[] body = format.serialize(event);

            HttpClient httpClient = HttpClient.newHttpClient();
            HttpRequest request = HttpRequest.newBuilder()
                    .uri(URI.create(base + "/ce/send/event"))
                    .POST(HttpRequest.BodyPublishers.ofByteArray(body))
                    .header("Content-Type", "application/cloudevents+json")
                    .build();

            HttpResponse<String> response = httpClient.send(request,
                    HttpResponse.BodyHandlers.ofString());
            Map<?, ?> result = MAPPER.readValue(response.body(), Map.class);
            System.out.printf("Published: status=%d is_error=%s%n",
                    response.statusCode(), result.get("is_error"));

            // Wait for event.
            String data = received.poll(10, TimeUnit.SECONDS);
            Map<?, ?> ce = MAPPER.readValue(data, Map.class);
            System.out.println("Received: " + ce.get("type") + " / " + ce.get("data"));
            subscriber.interrupt();
        }
    }
    ```
  </Tab>

  <Tab value="JavaScript">
    ```typescript
    // Example: events/basic-pubsub
    // Run: npx tsx events/basic-pubsub/index.ts
    import { CloudEvent, HTTP } from 'cloudevents';
    import EventSource from 'eventsource';

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

    async function waitForEvent(
      base: string,
      channel: string,
      clientId: string,
    ): Promise<Record<string, unknown>> {
      return new Promise((resolve, reject) => {
        const sseUrl = `${base}/ce/subscribe/events?client_id=${clientId}&channel=${encodeURIComponent(channel)}`;
        const es = new EventSource(sseUrl);

        const timer = setTimeout(() => {
          es.close();
          reject(new Error('Timed out waiting for event'));
        }, 10_000);

        es.addEventListener('cloudevent', (evt: MessageEvent) => {
          clearTimeout(timer);
          es.close();
          resolve(JSON.parse(evt.data) as Record<string, unknown>);
        });
      });
    }

    async function main(): Promise<void> {
      const base = serverUrl();
      const channel = 'js-ce-events.basic-pubsub';
      const clientId = 'kubemq-ce-js-example';

      // Start waiting for event (opens SSE stream).
      const eventPromise = waitForEvent(base, channel, `${clientId}-sub`);

      // Allow SSE connection to establish.
      await new Promise((r) => setTimeout(r, 500));

      // Build and publish CloudEvent (structured mode).
      const event = new CloudEvent({
        type: 'com.kubemq.examples.events.sent',
        source: clientId,
        subject: channel,
        datacontenttype: 'application/json',
        data: { message: 'Hello from JavaScript/TypeScript CloudEvents example!' },
      });

      const message = HTTP.structured(event);
      const resp = await fetch(`${base}/ce/send/event`, {
        method: 'POST',
        headers: message.headers as Record<string, string>,
        body: message.body as string,
      });
      const result = await resp.json() as { is_error: boolean };
      console.log(`Published: status=${resp.status} is_error=${result.is_error}`);

      // Wait for subscriber.
      const received = await eventPromise;
      console.log(`Received: ${received.type} / ${JSON.stringify(received.data)}`);
    }

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

  <Tab value="Python">
    ```python
    # Example: events/basic_pubsub
    # Run: python events/basic_pubsub/main.py
    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 subscribe(base: str, channel: str, client_id: str, received: list[str]) -> None:
        """Open SSE stream and collect one cloudevent."""
        sse_url = (
            f"{base}/ce/subscribe/events"
            f"?client_id={client_id}&channel={channel}"
        )
        with requests.get(sse_url, stream=True, timeout=None,
                          headers={"Accept": "text/event-stream",
                                   "Cache-Control": "no-cache"}) as resp:
            event_type = ""
            data = ""
            for line in resp.iter_lines(decode_unicode=True):
                if line == "":
                    if event_type == "cloudevent" and data:
                        received.append(data)
                        return
                    event_type = ""
                    data = ""
                    continue
                if line.startswith(":"):
                    continue  # keepalive
                if line.startswith("event:"):
                    event_type = line[len("event:"):].strip()
                elif line.startswith("data:"):
                    data = line[len("data:"):].strip()


    def main() -> None:
        base = server_url()
        channel = "python-ce-events.basic-pubsub"
        client_id = "kubemq-ce-python-example"

        received: list[str] = []

        # Start subscriber in background thread.
        t = threading.Thread(
            target=subscribe,
            args=(base, channel, client_id + "-sub", received),
            daemon=True,
        )
        t.start()

        # Allow SSE connection to establish.
        time.sleep(0.5)

        # Build and send CloudEvent (structured mode).
        event = CloudEvent(
            attributes={
                "type": "com.kubemq.examples.events.sent",
                "source": client_id,
                "subject": channel,
                "datacontenttype": "application/json",
            },
            data={"message": "Hello from Python CloudEvents example!"},
        )

        headers, body = to_structured(event)
        resp = requests.post(
            f"{base}/ce/send/event",
            data=body,
            headers=dict(headers),
            timeout=10,
        )
        result = resp.json()
        print(f"Published: status={resp.status_code} is_error={result.get('is_error')}")

        # Wait for subscriber.
        deadline = time.time() + 10
        while not received and time.time() < deadline:
            time.sleep(0.1)

        ce = json.loads(received[0])
        print(f"Received: {ce.get('type')} / {ce.get('data')}")


    if __name__ == "__main__":
        main()
    ```
  </Tab>

  <Tab value="Ruby">
    ```ruby
    # Example: events/basic_pubsub
    # Run: ruby events/basic_pubsub/main.rb
    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")
    end

    base    = server_url
    channel = "ruby-ce-events.basic-pubsub"
    client_id = "kubemq-ce-ruby-example"

    received = Queue.new

    # SSE subscriber thread.
    subscriber = Thread.new do
      uri = URI("#{base}/ce/subscribe/events?client_id=#{client_id}-sub&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"
        req["Cache-Control"] = "no-cache"
        http.request(req) do |resp|
          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
                  received.push(data)
                  Thread.exit
                end
                ev_type = nil
                data    = nil
              elsif line.start_with?(":") # keepalive
              elsif line.start_with?("event:")
                ev_type = line.sub("event:", "").strip
              elsif line.start_with?("data:")
                data = line.sub("data:", "").strip
              end
            end
          end
        end
      end
    end

    # Allow subscription to establish.
    sleep 0.5

    # Build and publish CloudEvent (structured mode).
    sdk  = CloudEvents::HttpBinding.default
    event = CloudEvents::Event::V1.new(
      id:              SecureRandom.uuid,
      type:            "com.kubemq.examples.events.sent",
      source:          URI("urn:#{client_id}"),
      subject:         channel,
      spec_version:    "1.0",
      data_content_type: CloudEvents::ContentType.new("application/json"),
      data:            JSON.generate({ message: "Hello from Ruby CloudEvents example!" })
    )

    # Encode as structured mode.
    headers, body = sdk.encode_event(event, structured_format: "json")
    uri = URI("#{base}/ce/send/event")
    Net::HTTP.start(uri.host, uri.port) do |http|
      req = Net::HTTP::Post.new(uri)
      req["Content-Type"] = headers["Content-Type"]
      req.body = body
      res = http.request(req)
      result = JSON.parse(res.body)
      puts "Published: status=#{res.code} is_error=#{result['is_error']}"
    end

    # Wait for subscriber.
    data = nil
    Timeout.timeout(10) { data = received.pop }
    ce = JSON.parse(data)
    puts "Received: #{ce['type']} / #{ce['data']}"
    ```
  </Tab>

  <Tab value="Rust">
    ```rust
    //! Example: events/basic-pubsub
    //! Run: cargo run -p basic-pubsub
    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())
    }

    /// Parse SSE lines and return the data of the first cloudevent.
    async fn wait_for_cloudevent(
        mut stream: impl futures_util::Stream<Item = reqwest::Result<Bytes>> + Unpin,
        tx: oneshot::Sender<String>,
    ) {
        let mut event_type = String::new();
        let mut data = String::new();
        let mut buffer = String::new();

        while let Some(chunk) = stream.next().await {
            let chunk = match chunk {
                Ok(c) => c,
                Err(e) => { eprintln!("SSE read error: {}", e); break; }
            };
            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 event_type == "cloudevent" && !data.is_empty() {
                        let _ = tx.send(data.clone());
                        return;
                    }
                    event_type.clear();
                    data.clear();
                } else if line.starts_with(':') {
                    // keepalive comment — ignore
                } else if let Some(v) = line.strip_prefix("event:") {
                    event_type = v.trim().to_string();
                } else if let Some(v) = line.strip_prefix("data:") {
                    data = v.trim().to_string();
                }
            }
        }
    }

    #[tokio::main]
    async fn main() -> Result<(), Box<dyn std::error::Error>> {
        let base = server_url();
        let channel = "rust-ce-events.basic-pubsub";
        let client_id = "kubemq-ce-rust-example";

        let client = Client::new();

        // Start SSE subscriber.
        let (tx, rx) = oneshot::channel::<String>();
        let sub_url = format!(
            "{}/ce/subscribe/events?client_id={}-sub&channel={}",
            base, client_id, channel
        );
        let sub_client = client.clone();
        tokio::spawn(async move {
            let stream = sub_client
                .get(&sub_url)
                .header("Accept", "text/event-stream")
                .header("Cache-Control", "no-cache")
                .send()
                .await
                .expect("SSE connect failed")
                .bytes_stream();
            wait_for_cloudevent(stream, tx).await;
        });

        // Allow SSE to establish.
        tokio::time::sleep(tokio::time::Duration::from_millis(500)).await;

        // Build CloudEvent (structured mode using cloudevents-sdk).
        let event = EventBuilderV10::new()
            .id(Uuid::new_v4().to_string())
            .ty("com.kubemq.examples.events.sent")
            .source(format!("urn:{}", client_id))
            .subject(channel)
            .data(
                "application/json",
                json!({"message": "Hello from Rust CloudEvents example!"}),
            )
            .build()?;

        // Serialize to structured mode JSON.
        let body = serde_json::to_string(&event)?;
        let resp = client
            .post(format!("{}/ce/send/event", base))
            .header("Content-Type", "application/cloudevents+json")
            .body(body)
            .send()
            .await?;

        let result: Value = resp.json().await?;
        println!("Published: status=202 is_error={}", result["is_error"]);

        // Wait for received event.
        let data = tokio::time::timeout(tokio::time::Duration::from_secs(10), rx)
            .await
            .expect("Timed out waiting for event")
            .expect("Channel closed");

        let ce: Value = serde_json::from_str(&data)?;
        println!("Received: {} / {}", ce["type"], ce["data"]);

        Ok(())
    }
    ```
  </Tab>
</Tabs>

The publish call returns `HTTP 202 Accepted`:

```json
{ "is_error": false, "message": "OK", "data": {} }
```

## Subscribing [#subscribing]

Open a long-lived SSE stream with `GET /ce/subscribe/events`. The connector streams `event: cloudevent` frames as messages arrive; each `data:` line holds the reconstructed CloudEvent JSON.

<Tabs groupId="language" items="['curl']">
  <Tab value="curl">
    ```bash
    curl -N "http://localhost:9090/ce/subscribe/events?client_id=my-client&channel=orders"
    ```
  </Tab>
</Tabs>

A delivered frame looks like:

```text
event: cloudevent
data: {"specversion":"1.0","type":"com.example.order.created","source":"order-service","subject":"orders","data":{"order_id":"12345","amount":99.99}}

```

Subscription accepts these query parameters:

| Parameter   | Type   | Required | Description                                                        |
| ----------- | ------ | -------- | ------------------------------------------------------------------ |
| `client_id` | string | Yes      | Client identifier (overridden by auth claims when auth is enabled) |
| `channel`   | string | Yes      | Channel to subscribe to                                            |
| `group`     | string | No       | Load-balancing group name                                          |

## Consumer groups [#consumer-groups]

Without a `group`, every subscriber on a channel receives every event (fan-out). Add the **same** `group` to multiple subscribers and they form a load-balancing pool — each event is delivered to exactly **one** member, round-robin.

<Tabs groupId="language" items="['curl','Go','Python']">
  <Tab value="curl">
    ```bash
    # Terminal 1 — worker 1 in group "workers"
    curl -N "http://localhost:9090/ce/subscribe/events?client_id=w1&channel=orders&group=workers"

    # Terminal 2 — worker 2 in the same group
    curl -N "http://localhost:9090/ce/subscribe/events?client_id=w2&channel=orders&group=workers"
    ```
  </Tab>

  <Tab value="Go">
    ```go
    // Example: events/consumer-group — each subscriber gets one of two events.
    func subscribe(base, clientID, channel, group string, results chan<- string, wg *sync.WaitGroup) {
    	defer wg.Done()
    	sseURL := fmt.Sprintf("%s/ce/subscribe/events?client_id=%s&channel=%s&group=%s",
    		base, clientID, channel, group)
    	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} // no timeout for SSE
    	resp, err := client.Do(req)
    	if err != nil {
    		log.Printf("[%s] SSE connect error: %v", clientID, err)
    		return
    	}
    	defer resp.Body.Close()

    	scanner := bufio.NewScanner(resp.Body)
    	var eventType, data string
    	for scanner.Scan() {
    		line := scanner.Text()
    		if line == "" {
    			if eventType == "cloudevent" && data != "" {
    				results <- fmt.Sprintf("[%s] received: %s", clientID, data)
    				return
    			}
    			eventType, data = "", ""
    			continue
    		}
    		if strings.HasPrefix(line, ":") {
    			continue
    		}
    		if strings.HasPrefix(line, "event:") {
    			eventType = strings.TrimSpace(strings.TrimPrefix(line, "event:"))
    		} else if strings.HasPrefix(line, "data:") {
    			data = strings.TrimSpace(strings.TrimPrefix(line, "data:"))
    		}
    	}
    }

    func main() {
    	base := serverURL()
    	channel := "go-ce-events.consumer-group"
    	group := "workers"

    	results := make(chan string, 2)
    	var wg sync.WaitGroup

    	// Two subscribers in the same group.
    	for i := 1; i <= 2; i++ {
    		wg.Add(1)
    		go subscribe(base, fmt.Sprintf("worker-%d", i), channel, group, results, &wg)
    	}

    	time.Sleep(600 * time.Millisecond)

    	// Publish two events — each subscriber should get exactly one.
    	sendEvent(base, channel, 1)
    	sendEvent(base, channel, 2)

    	deadline := time.After(10 * time.Second)
    	for received := 0; received < 2; {
    		select {
    		case msg := <-results:
    			fmt.Println(msg)
    			received++
    		case <-deadline:
    			log.Fatal("Timed out waiting for events")
    		}
    	}
    }
    ```
  </Tab>

  <Tab value="Python">
    ```python
    # Example: events/consumer_group — load-balanced subscriber groups.
    import threading
    import time

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


    def subscribe_group(base, channel, group, client_id, results, stop):
        sse_url = (f"{base}/ce/subscribe/events"
                   f"?client_id={client_id}&channel={channel}&group={group}")
        with requests.get(sse_url, stream=True, timeout=None,
                          headers={"Accept": "text/event-stream"}) as resp:
            ev_type = ""
            data = ""
            for line in resp.iter_lines(decode_unicode=True):
                if stop.is_set():
                    return
                if line == "":
                    if ev_type == "cloudevent" and data:
                        results.append(f"[{client_id}] received: {data[:80]}")
                        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():
        base = "http://localhost:9090"
        channel = "python-ce-events.consumer-group"
        group = "workers"

        results, stop = [], threading.Event()
        for i in range(1, 3):
            threading.Thread(
                target=subscribe_group,
                args=(base, channel, group, f"worker-{i}", results, stop),
                daemon=True,
            ).start()

        time.sleep(0.6)

        for seq in range(1, 3):
            event = CloudEvent(
                attributes={
                    "type": "com.kubemq.examples.events.grouped",
                    "source": "kubemq-ce-python-example",
                    "subject": channel,
                    "datacontenttype": "application/json",
                },
                data={"seq": seq},
            )
            headers, body = to_structured(event)
            resp = requests.post(f"{base}/ce/send/event", data=body,
                                 headers=dict(headers), timeout=10)
            print(f"Published event seq={seq} (status={resp.status_code})")

        deadline = time.time() + 10
        while len(results) < 2 and time.time() < deadline:
            time.sleep(0.1)

        stop.set()
        for r in results:
            print(r)


    if __name__ == "__main__":
        main()
    ```
  </Tab>
</Tabs>

<Callout type="info">
  Consumer groups are server-side and ephemeral — they exist only while subscribers are connected. Because events are not persisted, a group that has no connected members when an event is published will not receive it. For durable, replayable delivery, use [Events Store](/connectors/cloudevents/how-to/events-store).
</Callout>

## Related [#related]

<Cards>
  <Card title="Events Store" href="/connectors/cloudevents/how-to/events-store" description="Persistent events with replay via Last-Event-ID and six start positions." />

  <Card title="Content modes" href="/connectors/cloudevents/how-to/content-modes" description="Structured vs binary CloudEvents and how the connector detects each." />

  <Card title="SSE behavior" href="/connectors/cloudevents/how-to/sse-behavior" description="Wire format, keepalives, idle timeouts, and reconnection." />

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