Events

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}
  }'

// 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")}");

// 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")
	}
}

// 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();
    }
}

// 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);
});

# 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()

# 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']}"

//! 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(())
}

curl -N "http://localhost:9090/ce/subscribe/events?client_id=my-client&channel=orders"

# 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"

// 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")
		}
	}
}

# 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()