# Getting Started (/connectors/cloudevents/tutorials/getting-started)



The CloudEvents connector lets you publish and subscribe to KubeMQ messages over plain
HTTP using the CNCF [CloudEvents](https://cloudevents.io/) envelope — no KubeMQ SDK
required. You `POST` a CloudEvent to `/ce/send/event` and open a long-lived SSE stream on
`/ce/subscribe/events` to receive it. This walkthrough takes you from a running server to
a verified publish-and-receive round-trip.

## Prerequisites [#prerequisites]

* A running **kubemq-server** with the shared HTTP server reachable on **port 9090**.
* `curl` (or one of the language clients below) to publish and subscribe.
* For the language tabs, a CloudEvents SDK for your runtime (the examples use the official
  CNCF SDKs, sourced from `.kb/cloud-events/examples`).

Confirm the shared HTTP server is live:

```bash
curl http://localhost:9090/ready
```

## Enable / disable [#enable--disable]

The CloudEvents connector is **enabled by default**. Start kubemq-server and the `/ce/*`
routes are live immediately — there is **no `=true` flag to set**.

To **disable** CloudEvents, set its enable variable to `false`:

<RunKubeMQ variant="disable" ports="[50000]" env="{ CONNECTORSCE_ENABLE: 'false' }" />

<Callout type="warn">
  The disable variable name is irregular by design — the config key `Connectors.CE.Enable`
  snake-cases to `CONNECTORSCE_ENABLE`, with **no underscore** between `CONNECTORS` and `CE`.
  Older docs show `CONNECTORS_CE_ENABLE`, which does **not** match the live binding. See
  [Shared HTTP server](/connectors/concepts/shared-http-server) for the full enable model.
</Callout>

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

A publisher `POST`s a CloudEvent to the connector, which maps it to a KubeMQ event and
delivers it to every SSE subscriber on the channel. The subscriber receives it as an
`event: cloudevent` SSE frame, reconstructed as a CloudEvent JSON object.

<Mermaid
  chart="`
graph LR
PUB[&#x22;Publisher&#x22;]
CE[&#x22;CloudEvents connector<br/>:9090&#x22;]
CH{{&#x22;Events channel<br/>(CE subject)&#x22;}}
SUB[&#x22;SSE subscriber&#x22;]

PUB -- &#x22;POST /ce/send/event&#x22; --> CE
CE -- publish --> CH
CH -. &#x22;event: cloudevent (SSE)&#x22; .-> SUB

class PUB,SUB client
class CE connector
class CH events
`"
/>

*The connector maps the incoming CloudEvent to a KubeMQ event and streams it back to SSE subscribers.*

## Steps [#steps]

<Steps>
  <Step>
    ### Subscribe to the channel [#subscribe-to-the-channel]

    Open a long-lived SSE stream on `/ce/subscribe/events`. Pass a `client_id` and the
    `channel` to subscribe to. Run this in a separate terminal — it stays open and prints
    each event as it arrives.

    ```bash
    curl -N "http://localhost:9090/ce/subscribe/events?client_id=demo-subscriber&channel=notifications"
    ```

    The connector sends a `: keepalive` comment every 30 seconds to hold the connection open;
    standard EventSource clients ignore it.
  </Step>

  <Step>
    ### Publish a CloudEvent [#publish-a-cloudevent]

    Send a CloudEvent in **structured mode** (`Content-Type: application/cloudevents+json`)
    to `/ce/send/event`. The CloudEvent `subject` becomes the KubeMQ channel, so it must
    match the channel you subscribed to. A successful send returns **HTTP 202**.

    The language tabs below each run the complete round-trip from a single program: they open
    the SSE subscription, publish one CloudEvent, then print the received event.

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

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

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

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

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

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

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

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

  <Step>
    ### Verify the round-trip [#verify-the-round-trip]

    The subscriber terminal from step 1 prints the event as an `event: cloudevent` SSE frame.
    Because the message carried `ce_*` tags, the connector reconstructs it as a CloudEvent
    JSON object on the way out:

    ```text
    event: cloudevent
    data: {"specversion":"1.0","type":"com.example.greeting","source":"demo-publisher","id":"550e8400-e29b-41d4-a716-446655440000","subject":"notifications","time":"2026-06-08T10:30:00Z","data":{"message":"Hello, CloudEvents!"}}

    ```

    The publish call returns `HTTP 202` with `is_error: false`, and the subscriber receives the
    event with `id` and `time` auto-generated by the connector. That confirms a successful
    publish-and-receive round-trip through the CloudEvents connector.
  </Step>
</Steps>

## What's next [#whats-next]

<Cards>
  <Card title="Configuration" href="/connectors/cloudevents/concepts/configuration-model" description="Tune the request timeout, SSE buffer size, idle timeout, and connection limit." />

  <Card title="Events" href="/connectors/cloudevents/how-to/events" description="Pub/sub over CloudEvents — consumer groups and fan-out delivery." />

  <Card title="Content modes" href="/connectors/cloudevents/how-to/content-modes" description="Structured vs binary content modes, and using a CloudEvents SDK." />

  <Card title="SSE behavior" href="/connectors/cloudevents/how-to/sse-behavior" description="Wire format, keepalive, idle timeout, and Last-Event-ID resume." />
</Cards>