# Channel Resolution (/connectors/cloudevents/how-to/channel-resolution)



Every CloudEvent the connector receives must land on exactly one KubeMQ **channel**, and is attributed to one **ClientID**. The connector derives both from the event itself — no separate addressing layer — so a well-formed CloudEvent is self-routing.

## Overview [#overview]

The CloudEvents connector accepts events over plain HTTP without any KubeMQ-specific addressing fields. Instead, it reads two things from each incoming event:

* the **target channel** — which KubeMQ channel the message is published to, and
* the **ClientID** — the identity recorded on the message.

Both are resolved on a fixed priority order, with sensible auto-generation for the optional `id` and `time` attributes. Getting this mapping right keeps your producers portable: the same CloudEvent works against KubeMQ or any other CloudEvents endpoint.

## Channel resolution [#channel-resolution]

The destination channel is resolved in priority order:

1. **CE `subject` attribute** — if the event sets `subject`, its value is used directly as the channel name.
2. **`?channel=` query parameter** — used only when `subject` is absent (or empty).
3. **HTTP 400** — if neither is present, the request is rejected with `{"is_error": true, "message": "channel is required"}`.

This applies to every send endpoint (`/ce/send/event`, `/ce/send/event-store`, `/ce/send/command`, `/ce/send/query`, `/ce/queue/send`). SSE subscribe endpoints always take the channel from the required `?channel=` query parameter instead.

### How it works [#how-it-works]

A `subject` on the event wins; otherwise the connector falls back to the `?channel=` query parameter, and rejects the request when neither is set.

<Mermaid
  chart="`
graph LR
CLIENT[&#x22;CloudEvents producer&#x22;]
CE[&#x22;CloudEvents connector<br/>:9090&#x22;]
SUB{{&#x22;subject =<br/>orders&#x22;}}
QP{{&#x22;?channel=<br/>orders&#x22;}}
REJ[&#x22;HTTP 400<br/>channel is required&#x22;]
CH{{&#x22;KubeMQ channel<br/>orders&#x22;}}

CLIENT -- &#x22;POST /ce/send/*&#x22; --> CE
CE -- &#x22;1. subject set&#x22; --> SUB
CE -. &#x22;2. else ?channel=&#x22; .-> QP
CE -. &#x22;3. else reject&#x22; .-> REJ
SUB --> CH
QP --> CH

class CE connector
class SUB,QP,CH events
class CLIENT client
class REJ external
`"
/>

*The `subject` attribute resolves the channel first; `?channel=` is the fallback, and a missing channel is a 400.*

## Using `subject` (recommended) [#using-subject-recommended]

The `subject` attribute is part of the standard CloudEvents envelope and carries semantic meaning — what the event is about. Using it for channel resolution makes the event **self-describing** and portable: the same event routes correctly without any KubeMQ-specific query string.

The native examples below all set `subject` to the channel name (`event.SetSubject(channel)`, `.withSubject(channel)`, `subject=channel`, …), publish one CloudEvent, and confirm the subscriber received it on that channel.

<Tabs groupId="language" items="['curl','C#','Go','Java','JavaScript','Python','Ruby','Rust']">
  <Tab value="curl">
    ```bash
    # 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"}'
    ```
  </Tab>

  <Tab value="C#">
    ```csharp
    // 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}");
    ```
  </Tab>

  <Tab value="Go">
    ```go
    // 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)
    }
    ```
  </Tab>

  <Tab value="Java">
    ```java
    // 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());
        }
    }
    ```
  </Tab>

  <Tab value="JavaScript">
    ```typescript
    // 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);
    });
    ```
  </Tab>

  <Tab value="Python">
    ```python
    # 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()
    ```
  </Tab>

  <Tab value="Ruby">
    ```ruby
    # 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
    ```
  </Tab>

  <Tab value="Rust">
    ```rust
    //! 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(())
    }
    ```
  </Tab>
</Tabs>

<Callout type="info">
  If `subject` is present but **empty**, the connector treats it as absent and falls through to the `?channel=` query parameter.
</Callout>

## Using `?channel=` [#using-channel]

The `?channel=` query parameter is the fallback for cases where `subject` is unavailable or reserved for a different meaning. It works in both structured and binary content modes.

<Tabs groupId="language" items="['curl']">
  <Tab value="curl">
    ```bash
    # 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"}
      }'
    ```
  </Tab>
</Tabs>

Use `?channel=` when:

* the CloudEvent schema reserves `subject` for a different semantic purpose;
* you are integrating with external CE producers that do not set `subject`; or
* you need to route the same event payload to different channels per deployment.

<Callout type="warn">
  Avoid setting `subject` to a value different from the intended channel. Downstream [CESQL routing](/connectors/cloudevents/how-to/cesql-routing) can read CE attributes including `subject`, so a mismatch produces confusing routing behavior. When neither `subject` nor `?channel=` is supplied, the connector returns HTTP 400 with `{"is_error": true, "message": "channel is required"}`.
</Callout>

## ClientID resolution [#clientid-resolution]

Every message also carries a KubeMQ **ClientID**. The connector resolves it as follows:

1. **Auth claims** — when authentication is enabled and the request carries valid credentials, the authenticated `ClientID` from the JWT claims **overrides** every other source.
2. **CE `source` attribute** — used as the default ClientID when auth is disabled, or when the authenticated claim is `anonymous`.

So with auth off, the event's `source` becomes the ClientID; with auth on, the verified caller identity always wins, regardless of what `source` says. See [Authentication](/connectors/cloudevents/how-to/authentication) for how claims are established.

| Condition                            | ClientID used                        |
| ------------------------------------ | ------------------------------------ |
| Auth enabled, valid claims           | Authenticated `ClientID` from claims |
| Auth disabled (or claim `anonymous`) | CE `source` attribute                |

## Auto-generated attributes [#auto-generated-attributes]

Two optional CloudEvent attributes are filled in by the connector before the event is processed, so you may omit them:

| Attribute | Auto-generated when | Value                              |
| --------- | ------------------- | ---------------------------------- |
| `id`      | empty or missing    | a new UUID v4                      |
| `time`    | zero or missing     | the current UTC time (RFC3339Nano) |

The required attributes — `specversion`, `type`, and `source` — are never auto-generated; omitting any of them is a validation error. For the full attribute-to-tag mapping (including how every CE attribute is stored as a `ce_*` tag), see the [CE to KubeMQ mapping](/connectors/cloudevents/reference/ce-to-kubemq-mapping) reference.

## Best practices [#best-practices]

* **Prefer `subject`** for channel resolution — it makes the CloudEvent self-describing and portable across environments.
* Use `?channel=` only for compatibility with external CE sources that cannot set `subject`.
* Keep channel names consistent across publishers and subscribers to avoid silent routing mismatches.
* When authentication is enabled, rely on the authenticated ClientID rather than `source` for identity-sensitive logic — claims always override `source`.

## Related [#related]

<Cards>
  <Card title="CE to KubeMQ mapping" href="/connectors/cloudevents/reference/ce-to-kubemq-mapping" description="The full attribute-to-tag table, extension handling, and data vs data_base64." />

  <Card title="Events" href="/connectors/cloudevents/how-to/events" description="Publish and subscribe to fire-and-forget CloudEvents on a channel." />

  <Card title="CESQL routing" href="/connectors/cloudevents/how-to/cesql-routing" description="Route events by CloudEvent attributes instead of channel name alone." />

  <Card title="Authentication" href="/connectors/cloudevents/how-to/authentication" description="JWT Bearer auth and how claims override the source-derived ClientID." />
</Cards>