Content Modes

# Structured mode — application/cloudevents+json body
curl -X POST http://localhost:9090/ce/send/event \
  -H "Content-Type: application/cloudevents+json" \
  -d '{
    "specversion": "1.0",
    "type": "com.kubemq.examples.events.content-mode",
    "source": "kubemq-ce-curl-example",
    "subject": "ce-events.content-modes",
    "datacontenttype": "application/json",
    "data": {"message": "structured mode payload"}
  }'

# Binary mode — ce-* headers, data-only body
curl -X POST http://localhost:9090/ce/send/event \
  -H "Content-Type: application/json" \
  -H "ce-specversion: 1.0" \
  -H "ce-type: com.kubemq.examples.events.content-mode" \
  -H "ce-source: kubemq-ce-curl-example" \
  -H "ce-subject: ce-events.content-modes" \
  -d '{"message": "binary mode payload"}'

// Example: events/ContentModes — structured and binary mode.
// Run: dotnet run
using CloudNative.CloudEvents;
using CloudNative.CloudEvents.SystemTextJson;
using System.Net.Http.Headers;
using System.Text.Json;

static string ServerUrl() => Environment.GetEnvironmentVariable("KUBEMQ_CE_URL") ?? "http://localhost:9090";

var base_ = ServerUrl();
var formatter = new JsonEventFormatter();
using var httpClient = new HttpClient();

var cloudEvent = new CloudEvent
{
    Id = Guid.NewGuid().ToString(),
    Type = "com.kubemq.examples.events.content-mode",
    Source = new Uri("urn:kubemq-ce-csharp-example"),
    Subject = "csharp-ce-events.content-modes",
    DataContentType = "application/json",
    Data = new { message = "hello content modes" },
};

// Structured mode
Console.WriteLine("Sending in structured mode:");
var structuredBytes = formatter.EncodeStructuredModeMessage(cloudEvent, out var structuredCt);
using var sc = new ByteArrayContent(structuredBytes.ToArray());
sc.Headers.ContentType = MediaTypeHeaderValue.Parse(structuredCt.ToString());
var r1 = await httpClient.PostAsync($"{base_}/ce/send/event", sc);
var j1 = JsonSerializer.Deserialize<JsonElement>(await r1.Content.ReadAsStringAsync());
Console.WriteLine($"[structured] status={r1.StatusCode} is_error={j1.GetProperty("is_error")}");

// Binary mode
Console.WriteLine("\nSending in binary mode:");
using var binaryContent = new StringContent(
    JsonSerializer.Serialize(new { message = "hello binary mode" }),
    System.Text.Encoding.UTF8, "application/json");
using var binaryRequest = new HttpRequestMessage(HttpMethod.Post, $"{base_}/ce/send/event") { Content = binaryContent };
binaryRequest.Headers.Add("ce-specversion", "1.0");
binaryRequest.Headers.Add("ce-type", cloudEvent.Type);
binaryRequest.Headers.Add("ce-source", cloudEvent.Source!.ToString());
binaryRequest.Headers.Add("ce-id", cloudEvent.Id);
binaryRequest.Headers.Add("ce-subject", cloudEvent.Subject);
binaryRequest.Headers.Add("ce-time", DateTimeOffset.UtcNow.ToString("O"));
var r2 = await httpClient.SendAsync(binaryRequest);
var j2 = JsonSerializer.Deserialize<JsonElement>(await r2.Content.ReadAsStringAsync());
Console.WriteLine($"[binary]     status={r2.StatusCode} is_error={j2.GetProperty("is_error")}");

Console.WriteLine("\nBoth content modes accepted.");

// Example: events/content-modes
// Sends the same event payload in both CloudEvents content modes:
//   - Structured: Content-Type: application/cloudevents+json, all attrs in JSON body
//   - Binary:     ce-* HTTP headers, raw data body
// Run: go run ./events/content-modes/main.go
package main

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

type CEResponse struct {
	IsError bool   `json:"is_error"`
	Message string `json:"message"`
}

func sendStructured(base string, event cloudevents.Event) error {
	body, err := json.Marshal(event)
	if err != nil {
		return err
	}
	req, err := http.NewRequest("POST", base+"/ce/send/event", bytes.NewReader(body))
	if err != nil {
		return err
	}
	req.Header.Set("Content-Type", "application/cloudevents+json")

	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()

	var result CEResponse
	_ = json.NewDecoder(resp.Body).Decode(&result)
	fmt.Printf("[structured] status=%d is_error=%v message=%s\n",
		resp.StatusCode, result.IsError, result.Message)
	return nil
}

func sendBinary(base string, event cloudevents.Event) error {
	// In binary mode, CE attributes go into ce-* HTTP headers.
	// The body contains only the event data.
	dataJSON, err := json.Marshal(map[string]string{"message": "binary mode payload"})
	if err != nil {
		return err
	}
	req, err := http.NewRequest("POST", base+"/ce/send/event", bytes.NewReader(dataJSON))
	if err != nil {
		return err
	}
	req.Header.Set("Content-Type", "application/json")
	req.Header.Set("ce-specversion", "1.0")
	req.Header.Set("ce-type", event.Type())
	req.Header.Set("ce-source", event.Source())
	req.Header.Set("ce-subject", event.Subject())
	req.Header.Set("ce-id", event.ID())
	req.Header.Set("ce-time", event.Time().Format(time.RFC3339Nano))

	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()

	var result CEResponse
	_ = json.NewDecoder(resp.Body).Decode(&result)
	fmt.Printf("[binary]     status=%d is_error=%v message=%s\n",
		resp.StatusCode, result.IsError, result.Message)
	return nil
}

func main() {
	base := serverURL()

	// Build a CloudEvent to send in both modes.
	event := cloudevents.NewEvent()
	event.SetType("com.kubemq.examples.events.content-mode")
	event.SetSource("kubemq-ce-go-example")
	event.SetSubject("go-ce-events.content-modes")
	_ = event.SetData(cloudevents.ApplicationJSON, map[string]string{
		"message": "structured mode payload",
	})

	fmt.Println("Sending in structured mode (Content-Type: application/cloudevents+json):")
	if err := sendStructured(base, event); err != nil {
		log.Fatal("structured send:", err)
	}

	fmt.Println("\nSending in binary mode (ce-* HTTP headers):")
	if err := sendBinary(base, event); err != nil {
		log.Fatal("binary send:", err)
	}

	fmt.Println("\nBoth content modes accepted by KubeMQ CE connector.")
}

// Example: events/content-modes — structured and binary mode.
// Run: mvn compile exec:java
package io.kubemq.examples.events.contentmodes;

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 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();
        EventFormatProvider.getInstance().registerFormat(new JsonFormat());
        EventFormat format = EventFormatProvider.getInstance().resolveFormat(JsonFormat.CONTENT_TYPE);
        HttpClient httpClient = HttpClient.newHttpClient();

        CloudEvent event = CloudEventBuilder.v1()
                .withId(UUID.randomUUID().toString())
                .withType("com.kubemq.examples.events.content-mode")
                .withSource(URI.create("kubemq-ce-java-example"))
                .withSubject("java-ce-events.content-modes")
                .withDataContentType("application/json")
                .withTime(OffsetDateTime.now())
                .withData("application/json",
                        MAPPER.writeValueAsBytes(Map.of("message", "hello content modes")))
                .build();

        // --- Structured mode ---
        System.out.println("Sending in structured mode:");
        byte[] structuredBody = format.serialize(event);
        HttpResponse<String> r1 = httpClient.send(
                HttpRequest.newBuilder().uri(URI.create(base + "/ce/send/event"))
                        .POST(HttpRequest.BodyPublishers.ofByteArray(structuredBody))
                        .header("Content-Type", "application/cloudevents+json").build(),
                HttpResponse.BodyHandlers.ofString());
        Map<?, ?> res1 = MAPPER.readValue(r1.body(), Map.class);
        System.out.printf("[structured] status=%d is_error=%s%n", r1.statusCode(), res1.get("is_error"));

        // --- Binary mode ---
        System.out.println("\nSending in binary mode:");
        byte[] data = event.getData() != null ? event.getData().toBytes() : new byte[0];
        HttpResponse<String> r2 = httpClient.send(
                HttpRequest.newBuilder().uri(URI.create(base + "/ce/send/event"))
                        .POST(HttpRequest.BodyPublishers.ofByteArray(data))
                        .header("Content-Type", "application/json")
                        .header("ce-specversion", "1.0")
                        .header("ce-type", event.getType())
                        .header("ce-source", event.getSource().toString())
                        .header("ce-id", event.getId())
                        .header("ce-subject", event.getSubject())
                        .header("ce-time", event.getTime().toString())
                        .build(),
                HttpResponse.BodyHandlers.ofString());
        Map<?, ?> res2 = MAPPER.readValue(r2.body(), Map.class);
        System.out.printf("[binary]     status=%d is_error=%s%n", r2.statusCode(), res2.get("is_error"));

        System.out.println("\nBoth content modes accepted.");
    }
}

// Example: events/content-modes
// Sends the same event in structured and binary mode.
// Run: npx tsx events/content-modes/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 event = new CloudEvent({
    type: 'com.kubemq.examples.events.content-mode',
    source: 'kubemq-ce-js-example',
    subject: 'js-ce-events.content-modes',
    datacontenttype: 'application/json',
    data: { message: 'hello content modes' },
  });

  // Structured mode
  console.log('Sending in structured mode:');
  const structured = HTTP.structured(event);
  const r1 = await fetch(`${base}/ce/send/event`, {
    method: 'POST',
    headers: structured.headers as Record<string, string>,
    body: structured.body as string,
  });
  const res1 = await r1.json() as { is_error: boolean };
  console.log(`[structured] status=${r1.status} is_error=${res1.is_error}`);

  // Binary mode
  console.log('\nSending in binary mode:');
  const binary = HTTP.binary(event);
  const r2 = await fetch(`${base}/ce/send/event`, {
    method: 'POST',
    headers: binary.headers as Record<string, string>,
    body: binary.body as string,
  });
  const res2 = await r2.json() as { is_error: boolean };
  console.log(`[binary]     status=${r2.status} is_error=${res2.is_error}`);

  console.log('\nBoth content modes accepted.');
}

main().catch(console.error);

# Example: events/content_modes — structured vs binary CloudEvents mode.
# Run: python events/content_modes/main.py
from __future__ import annotations

import os

import requests
from cloudevents.v1.conversion import to_binary, 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()

    event = CloudEvent(
        attributes={
            "type": "com.kubemq.examples.events.content-mode",
            "source": "kubemq-ce-python-example",
            "subject": "python-ce-events.content-modes",
            "datacontenttype": "application/json",
        },
        data={"message": "hello content modes"},
    )

    # --- Structured mode ---
    print("Sending in structured mode (Content-Type: application/cloudevents+json):")
    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"[structured] status={resp.status_code} is_error={result.get('is_error')}")

    # --- Binary mode ---
    print("\nSending in binary mode (ce-* HTTP headers):")
    headers, body = to_binary(event)
    resp = requests.post(f"{base}/ce/send/event", data=body,
                         headers=dict(headers), timeout=10)
    result = resp.json()
    print(f"[binary]     status={resp.status_code} is_error={result.get('is_error')}")

    print("\nBoth content modes accepted by KubeMQ CE connector.")


if __name__ == "__main__":
    main()

# Example: events/content_modes — structured and binary mode.
# Run: ruby events/content_modes/main.rb
require "net/http"
require "uri"
require "json"
require "time"
require "securerandom"
require "cloud_events"

def server_url = ENV.fetch("KUBEMQ_CE_URL", "http://localhost:9090")

base = server_url
sdk  = CloudEvents::HttpBinding.default

event = CloudEvents::Event::V1.new(
  id:           SecureRandom.uuid,
  type:         "com.kubemq.examples.events.content-mode",
  source:       URI("urn:kubemq-ce-ruby-example"),
  subject:      "ruby-ce-events.content-modes",
  spec_version: "1.0",
  data_content_type: CloudEvents::ContentType.new("application/json"),
  data: JSON.generate({ message: "hello content modes" })
)

def post_event(base, headers_hash, body)
  uri = URI("#{base}/ce/send/event")
  Net::HTTP.start(uri.host, uri.port) do |http|
    req = Net::HTTP::Post.new(uri)
    headers_hash.each { |k, v| req[k] = v }
    req.body = body
    res = http.request(req)
    JSON.parse(res.body).merge("status" => res.code)
  end
end

# Structured mode
puts "Sending in structured mode:"
enc_headers, enc_body = sdk.encode_event(event, structured_format: "json")
r1 = post_event(base, enc_headers, enc_body)
puts "[structured] status=#{r1['status']} is_error=#{r1['is_error']}"

# Binary mode — CE attrs in headers, data as body
puts "\nSending in binary mode:"
binary_headers = {
  "Content-Type"   => "application/json",
  "ce-specversion" => "1.0",
  "ce-type"        => event.type,
  "ce-source"      => event.source.to_s,
  "ce-id"          => event.id,
  "ce-subject"     => event.subject,
  "ce-time"        => Time.now.utc.iso8601(9),
}
r2 = post_event(base, binary_headers, event.data)
puts "[binary]     status=#{r2['status']} is_error=#{r2['is_error']}"

puts "\nBoth content modes accepted."

//! Example: events/content-modes — structured and binary mode.
//! Run: cargo run -p content-modes
use cloudevents::{AttributesReader, EventBuilder, EventBuilderV10};
use reqwest::Client;
use serde_json::{json, Value};
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 client = Client::new();

    let event = EventBuilderV10::new()
        .id(Uuid::new_v4().to_string())
        .ty("com.kubemq.examples.events.content-mode")
        .source("urn:kubemq-ce-rust-example")
        .subject("rust-ce-events.content-modes")
        .data("application/json", json!({"message": "hello content modes"}))
        .build()?;

    // Structured mode
    println!("Sending in structured mode:");
    let body = serde_json::to_string(&event)?;
    let r1 = client.post(format!("{}/ce/send/event", base))
        .header("Content-Type", "application/cloudevents+json")
        .body(body)
        .send().await?;
    let j1: Value = r1.json().await?;
    println!("[structured] status=202 is_error={}", j1["is_error"]);

    // Binary mode — CE attrs in ce-* headers, JSON data as body
    println!("\nSending in binary mode:");
    let data_body = serde_json::to_string(&json!({"message": "binary mode payload"}))?;
    let r2 = client.post(format!("{}/ce/send/event", base))
        .header("Content-Type", "application/json")
        .header("ce-specversion", "1.0")
        .header("ce-type", event.ty())
        .header("ce-source", event.source().as_str())
        .header("ce-id", event.id())
        .header("ce-subject", event.subject().unwrap_or(""))
        .body(data_body)
        .send().await?;
    let j2: Value = r2.json().await?;
    println!("[binary]     status=202 is_error={}", j2["is_error"]);

    println!("\nBoth content modes accepted.");
    Ok(())
}