Getting Started

curl -X POST http://localhost:9090/ce/send/event \
  -H "Content-Type: application/cloudevents+json" \
  -d '{
    "specversion": "1.0",
    "type": "com.example.greeting",
    "source": "demo-publisher",
    "subject": "notifications",
    "datacontenttype": "application/json",
    "data": {"message": "Hello, CloudEvents!"}
  }'

using CloudNative.CloudEvents;
using CloudNative.CloudEvents.SystemTextJson;
using System.Net.Http.Headers;
using System.Text;
using System.Text.Json;

var base_ = Environment.GetEnvironmentVariable("KUBEMQ_CE_URL") ?? "http://localhost:9090";
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.
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10));
var ce = await received.Task.WaitAsync(cts.Token);
Console.WriteLine($"Received: type={ce.GetProperty("type")} data={ce.GetProperty("data")}");

package main

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

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

func main() {
    base := "http://localhost:9090"
    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")

        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 data=%v\n", ce["type"], ce["data"])
    case <-ctx.Done():
        log.Fatal("Timed out waiting for event")
    }
}

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 final ObjectMapper MAPPER = new ObjectMapper();

    public static void main(String[] args) throws Exception {
        String base = "http://localhost:9090";
        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.setReadTimeout(15_000);
                try (BufferedReader reader = new BufferedReader(
                        new InputStreamReader(conn.getInputStream(), StandardCharsets.UTF_8))) {
                    String line, 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);
        if (data == null) {
            throw new RuntimeException("Timed out waiting for event");
        }
        Map<?, ?> ce = MAPPER.readValue(data, Map.class);
        System.out.println("Received: type=" + ce.get("type") + " data=" + ce.get("data"));
        subscriber.interrupt();
    }
}

import { CloudEvent, HTTP } from 'cloudevents';
import EventSource from 'eventsource';

const base = process.env.KUBEMQ_CE_URL ?? 'http://localhost:9090';
const channel = 'js-ce-events.basic-pubsub';
const clientId = 'kubemq-ce-js-example';

function waitForEvent(): Promise<Record<string, unknown>> {
  return new Promise((resolve, reject) => {
    const sseUrl = `${base}/ce/subscribe/events?client_id=${clientId}-sub&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> {
  // Start waiting for event (opens SSE stream).
  const eventPromise = waitForEvent();

  // 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 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: type=${received.type} data=${JSON.stringify(received.data)}`);
}

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

import json
import threading
import time

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

base = "http://localhost:9090"
channel = "python-ce-events.basic-pubsub"
client_id = "kubemq-ce-python-example"

received: list[str] = []


def subscribe() -> None:
    """Open SSE stream and collect one cloudevent."""
    sse_url = (
        f"{base}/ce/subscribe/events"
        f"?client_id={client_id}-sub&channel={channel}"
    )
    with requests.get(sse_url, stream=True, timeout=None,
                      headers={"Accept": "text/event-stream"}) 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()


# Start subscriber in background thread.
threading.Thread(target=subscribe, daemon=True).start()
time.sleep(0.5)  # allow SSE connection to establish

# 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)
if not received:
    raise TimeoutError("Timed out waiting for event")

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

require "net/http"
require "uri"
require "json"
require "timeout"
require "securerandom"
require "cloud_events"

base      = "http://localhost:9090"
channel   = "ruby-ce-events.basic-pubsub"
client_id = "kubemq-ce-ruby-example"

received = Queue.new

# SSE subscriber thread.
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"
    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

sleep 0.5 # allow subscription to establish

# 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!" })
)

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: type=#{ce['type']} data=#{ce['data']}"

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

/// 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 = "http://localhost:9090";
    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")
            .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: type={} data={}", ce["type"], ce["data"]);

    Ok(())
}