Channel Resolution

# subject sets the KubeMQ channel (structured mode)
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"}
  }'

# binary mode — subject travels in the ce-subject header
curl -X POST http://localhost:9090/ce/send/event \
  -H "Content-Type: application/json" \
  -H "ce-specversion: 1.0" \
  -H "ce-type: com.example.order.created" \
  -H "ce-source: order-service" \
  -H "ce-subject: orders" \
  -d '{"order_id": "12345"}'

// Example: events/BasicPubSub
// Run: dotnet run
using CloudNative.CloudEvents;
using CloudNative.CloudEvents.SystemTextJson;
using System.Net.Http.Headers;

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

// 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:{clientId}"),
    Subject = channel,            // subject = KubeMQ channel
    DataContentType = "application/json",
    Data = new { message = "Hello from C# CloudEvents example!" },
};

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 to channel '{channel}': status={resp.StatusCode}");

// Example: events/basic-pubsub
// Run: go run ./events/basic-pubsub/main.go
package main

import (
	"encoding/json"
	"fmt"
	"log"
	"net/http"
	"os"
	"strings"

	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"

	// 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()
	fmt.Printf("Published to channel %q: status=%d\n", channel, resp.StatusCode)
}

// 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.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;

public class Main {
    static final ObjectMapper MAPPER = new ObjectMapper();

    public static void main(String[] args) throws Exception {
        String base = System.getenv().getOrDefault("KUBEMQ_CE_URL", "http://localhost:9090");
        String channel = "java-ce-events.basic-pubsub";
        String clientId = "kubemq-ce-java-example";

        // 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)               // subject = KubeMQ 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);
        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.newHttpClient()
                .send(request, HttpResponse.BodyHandlers.ofString());
        System.out.printf("Published to channel '%s': status=%d%n", channel, response.statusCode());
    }
}

// Example: events/basic-pubsub
// Run: npx tsx events/basic-pubsub/index.ts
import { CloudEvent, HTTP } from 'cloudevents';

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

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

  // Build and publish CloudEvent (structured mode).
  const event = new CloudEvent({
    type: 'com.kubemq.examples.events.sent',
    source: clientId,
    subject: channel,                 // subject = KubeMQ 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,
  });
  console.log(`Published to channel '${channel}': status=${resp.status}`);
}

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 os

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 main() -> None:
    base = server_url()
    channel = "python-ce-events.basic-pubsub"
    client_id = "kubemq-ce-python-example"

    # Build and send CloudEvent (structured mode).
    event = CloudEvent(
        attributes={
            "type": "com.kubemq.examples.events.sent",
            "source": client_id,
            "subject": channel,           # subject = KubeMQ 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)
    print(f"Published to channel '{channel}': status={resp.status_code}")


if __name__ == "__main__":
    main()

# Example: events/basic_pubsub
# Run: ruby events/basic_pubsub/main.rb
require "net/http"
require "uri"
require "json"
require "securerandom"
require "cloud_events"

base    = ENV.fetch("KUBEMQ_CE_URL", "http://localhost:9090")
channel = "ruby-ce-events.basic-pubsub"
client_id = "kubemq-ce-ruby-example"

# 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,             # subject = KubeMQ 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)
  puts "Published to channel '#{channel}': status=#{res.code}"
end

//! Example: events/basic-pubsub
//! Run: cargo run -p basic-pubsub
use cloudevents::{EventBuilder, EventBuilderV10};
use reqwest::Client;
use serde_json::json;
use std::env;
use uuid::Uuid;

fn server_url() -> String {
    env::var("KUBEMQ_CE_URL").unwrap_or_else(|_| "http://localhost:9090".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";

    // 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) // subject = KubeMQ channel
        .data(
            "application/json",
            json!({"message": "Hello from Rust CloudEvents example!"}),
        )
        .build()?;

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

    println!("Published to channel '{}': status={}", channel, resp.status());
    Ok(())
}

# channel via query parameter — no subject on the event
curl -X POST "http://localhost:9090/ce/send/event?channel=orders" \
  -H "Content-Type: application/cloudevents+json" \
  -d '{
    "specversion": "1.0",
    "type": "com.example.order.created",
    "source": "order-service",
    "data": {"order_id": "12345"}
  }'