# CloudEvents (/connectors/cloudevents) The **CloudEvents (CE) connector** is a built-in protocol gateway in kubemq-server that speaks the [CNCF CloudEvents](https://cloudevents.io/) specification over HTTP. It lets any HTTP client publish and subscribe to KubeMQ using a standard, interoperable event envelope — no KubeMQ SDK required. ## What is the CloudEvents connector [#what-is-the-cloudevents-connector] [CloudEvents](https://cloudevents.io/) is a CNCF specification that describes event metadata in a common format: a standard envelope of `specversion`, `type`, `source`, `id`, `subject`, `time`, `data`, and extensions that travels consistently across platforms and languages. The CE connector exposes this format natively over HTTP for **all five KubeMQ messaging patterns** — Events, Events Store, Queues, Commands, and Queries. A CloudEvent posted to `/ce/send/event` becomes a native KubeMQ message; a message delivered over Server-Sent Events (SSE) is reconstructed back into a CloudEvent. The connector is a *gateway*, not a client library — your application only needs an HTTP client (and, optionally, a CloudEvents SDK to build the envelope for you). Key capabilities: * **All messaging patterns** — pub/sub events, persistent events store, durable queues, and request/reply commands and queries, all over CloudEvents HTTP. * **Structured and binary content modes** — send the whole event as a JSON body (`application/cloudevents+json`) or carry attributes in `ce-*` HTTP headers; the connector detects either automatically. * **CESQL attribute routing** — route events to channels using CloudEvents SQL expressions evaluated server-side against event attributes. * **SSE subscriptions with replay** — subscribe to a long-lived event stream and resume an events-store subscription from a chosen position using `Last-Event-ID`. The CE connector runs on the **shared HTTP server (port 9090)** alongside the REST, MCP, and A2A connectors and is **enabled by default** — start kubemq-server and `/ce/*` is live. See [Shared HTTP server](/connectors/concepts/shared-http-server) for the port, middleware chain, and the disable model. ## How it works [#how-it-works] A CloudEvents client sends events to the CE connector, which maps them into native KubeMQ messages and hands them to the broker; subscribers receive the same events back as CloudEvents over SSE. *The CloudEvents connector translates CloudEvents HTTP requests into native KubeMQ messages and streams them back to subscribers over SSE.* ## Endpoint surface [#endpoint-surface] | Endpoint | Method | Pattern | | ---------------------------- | --------- | ------------------------------------------------------ | | `/ce/send/event` | POST | Events (fire-and-forget pub/sub) | | `/ce/send/event-store` | POST | Events Store (persistent, replayable) | | `/ce/send/command` | POST | Command request | | `/ce/send/query` | POST | Query request | | `/ce/send/response` | POST | Response to a command/query (`?request_id=`) | | `/ce/queue/send` | POST | Queue send | | `/ce/queue/receive` | POST | Queue receive (control op) | | `/ce/queue/ack_all` | POST | Queue ack-all (control op) | | `/ce/subscribe/events` | GET (SSE) | Subscribe to events | | `/ce/subscribe/events-store` | GET (SSE) | Subscribe to events store (replay via `Last-Event-ID`) | | `/ce/subscribe/commands` | GET (SSE) | Subscribe to commands | | `/ce/subscribe/queries` | GET (SSE) | Subscribe to queries | Every send endpoint accepts a CloudEvent in either structured or binary content mode. Channels resolve from the CloudEvent `subject` attribute, falling back to a `?channel=` query parameter. See [CE ↔ KubeMQ mapping](/connectors/cloudevents/reference/ce-to-kubemq-mapping) for the full attribute table and resolution rules. ## Send a CloudEvent [#send-a-cloudevent] Publish a fire-and-forget event with `POST /ce/send/event`. The example below sends a structured-mode CloudEvent whose `subject` (`notifications`) becomes the KubeMQ channel. A successful send returns HTTP 202 with `is_error: false`. ```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": "notifications", "datacontenttype": "application/json", "data": {"order_id": "12345", "amount": 99.99} }' ``` ```csharp using CloudNative.CloudEvents; using CloudNative.CloudEvents.SystemTextJson; using System.Net.Http.Headers; var base_ = "http://localhost:9090"; var channel = "notifications"; // Build and publish a 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:kubemq-ce-csharp-example"), 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); Console.WriteLine($"Published: status={resp.StatusCode}"); ``` ```go import ( "encoding/json" "net/http" "strings" cloudevents "github.com/cloudevents/sdk-go/v2" ) base := "http://localhost:9090" channel := "notifications" // 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 { panic(err) } defer resp.Body.Close() ``` ```java 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.net.URI; import java.net.http.HttpClient; import java.net.http.HttpRequest; import java.net.http.HttpResponse; import java.time.OffsetDateTime; import java.util.Map; import java.util.UUID; String base = "http://localhost:9090"; String channel = "notifications"; ObjectMapper mapper = new ObjectMapper(); // 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("kubemq-ce-java-example")) .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 response = httpClient.send(request, HttpResponse.BodyHandlers.ofString()); System.out.printf("Published: status=%d%n", response.statusCode()); ``` ```typescript import { CloudEvent, HTTP } from 'cloudevents'; const base = 'http://localhost:9090'; const channel = 'notifications'; // Build and publish CloudEvent (structured mode). const event = new CloudEvent({ type: 'com.kubemq.examples.events.sent', source: 'kubemq-ce-js-example', 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, 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}`); ``` ```python import requests from cloudevents.v1.conversion import to_structured from cloudevents.v1.http import CloudEvent base = "http://localhost:9090" channel = "notifications" # Build and send CloudEvent (structured mode). event = CloudEvent( attributes={ "type": "com.kubemq.examples.events.sent", "source": "kubemq-ce-python-example", "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')}") ``` ```ruby require "net/http" require "uri" require "json" require "securerandom" require "cloud_events" base = "http://localhost:9090" channel = "notifications" # 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:kubemq-ce-ruby-example"), subject: channel, spec_version: "1.0", data_content_type: CloudEvents::ContentType.new("application/json"), data: JSON.generate({ message: "Hello from Ruby CloudEvents example!" }) ) 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 ``` ```rust use cloudevents::{EventBuilder, EventBuilderV10}; use reqwest::Client; use serde_json::{json, Value}; use uuid::Uuid; let base = "http://localhost:9090"; let channel = "notifications"; let client = Client::new(); // Build CloudEvent (structured mode using cloudevents-sdk). let event = EventBuilderV10::new() .id(Uuid::new_v4().to_string()) .ty("com.kubemq.examples.events.sent") .source("urn:kubemq-ce-rust-example") .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: is_error={}", result["is_error"]); ``` ## Supported languages [#supported-languages] The CE connector is plain HTTP, so any language with an HTTP client works. The examples in this section use each language's official CloudEvents SDK to build the envelope: | Language | CloudEvents SDK | Version | Runtime | | ----------------------- | ---------------------------------- | ------- | ------------ | | Go | `github.com/cloudevents/sdk-go/v2` | 2.16.0 | Go 1.21+ | | Python | `cloudevents` | 2.0.0 | Python 3.10+ | | JavaScript / TypeScript | `cloudevents` | 10.0.0 | Node.js 18+ | | Java | `io.cloudevents:cloudevents-core` | 4.0.1 | Java 21+ | | C# | `CloudNative.CloudEvents` | 2.8.0 | .NET 8+ | | Ruby | `cloud_events` | 0.9 | Ruby 3.1+ | | Rust | `cloudevents-sdk` | 0.9 | Rust 1.75+ | The CloudEvents SDK is a convenience for building and parsing the envelope — it is not required. The structured-mode body and binary-mode `ce-*` headers are plain HTTP, so curl and any raw HTTP client work just as well. See [Content modes](/connectors/cloudevents/how-to/content-modes) for both wire formats. ## Next steps [#next-steps]