# Events Reference (/learn/events/reference)



## Message Structure [#message-structure]

### Event Message (Send) [#event-message-send]

<TypeTable
  type="{
  channel: { type: 'string', description: 'Target channel name. Must follow channel naming rules.', required: true },
  clientId: { type: 'string', description: 'Sender client identifier. Must match ^[a-zA-Z0-9_-]+$.' },
  metadata: { type: 'string', description: 'Text metadata (at least one of body or metadata required).' },
  body: { type: 'bytes', description: 'Message payload (at least one of body or metadata required).' },
  tags: { type: 'map<string, string>', description: 'Key-value pairs for filtering and routing.' },
  store: { type: 'boolean', description: 'Must be false for Events (true = Events Store).', default: 'false' },
}"
/>

### Event Receive (Subscribe) [#event-receive-subscribe]

<TypeTable
  type="{
  eventId: { type: 'string', description: 'Server-assigned unique event identifier.' },
  channel: { type: 'string', description: 'Channel the event was published to.' },
  metadata: { type: 'string', description: 'Text metadata from publisher.' },
  body: { type: 'bytes', description: 'Message payload from publisher.' },
  tags: { type: 'map<string, string>', description: 'Tags from publisher.' },
}"
/>

<Callout type="info">
  Unlike Events Store, plain Events do not include `Timestamp` or `Sequence` fields because there is no persistence layer.
</Callout>

### Send Result [#send-result]

<TypeTable
  type="{
  eventId: { type: 'string', description: 'The event identifier (auto-generated if not provided).' },
  sent: { type: 'boolean', description: 'true if the event was published successfully.' },
  error: { type: 'string', description: 'Error message if the publish failed. Empty on success.' },
}"
/>

## Subscription Options [#subscription-options]

<TypeTable
  type="{
  channel: { type: 'string', description: 'Channel or wildcard pattern to subscribe to.', required: true },
  group: { type: 'string', description: 'Consumer group name for load-balanced delivery. Empty = receive all events.' },
  clientId: { type: 'string', description: 'Unique subscriber identifier. Must match ^[a-zA-Z0-9_-]+$.' },
}"
/>

### Consumer Groups [#consumer-groups]

When multiple subscribers specify the same `group` value on the same channel:

* Each event is delivered to exactly **one** member of the group (round-robin)
* When `group` is empty, every subscriber receives every event (standard fan-out)
* Groups are independent per channel
* There is no limit on the number of group members

See [Consumer Groups](/learn/events/tutorials/consumer-groups) and [Scale Subscribers](/learn/events/how-to/scale-subscribers) for examples.

## Channel Naming Rules [#channel-naming-rules]

| Rule                   | Constraint                                | Error Code |
| ---------------------- | ----------------------------------------- | ---------- |
| Required               | Cannot be empty                           | 102        |
| No trailing dot        | Cannot end with `.`                       | 119        |
| No whitespace          | Cannot contain spaces                     | 108        |
| No wildcards (publish) | Cannot contain `*` or `>` when publishing | 107        |
| Max length             | 256 characters                            | —          |

**Valid channel regex (publish):** `^[^\s*>]+[^.]$`

Wildcards (`*` and `>`) are allowed in **subscription** channel patterns only. See [Wildcard Subscriptions](/learn/events/tutorials/wildcard-subscriptions).

### Channel Naming Conventions [#channel-naming-conventions]

Use dot-separated hierarchical names for best results with wildcard subscriptions:

```text
{domain}.{entity}.{action}

orders.created
orders.us-east.shipped
payments.completed
inventory.reserved
```

## Routing (Multicast) [#routing-multicast]

Events support multicast publishing through special channel syntax:

| Character | Purpose                            | Example                            |
| --------- | ---------------------------------- | ---------------------------------- |
| `;`       | Separate channels of the same type | `orders;notifications`             |
| `:`       | Specify target pattern type        | `events:orders;events_store:audit` |

### Pattern Type Prefixes [#pattern-type-prefixes]

| Prefix          | Target Pattern               |
| --------------- | ---------------------------- |
| `events:`       | Events (fire-and-forget)     |
| `events_store:` | Events Store (persistent)    |
| `queues:`       | Queues (guaranteed delivery) |

Routed messages are automatically tagged with `X-KUBEMQ-ROUTED=true`. See [Multicast Events](/learn/events/tutorials/multicast).

## Transport Protocols [#transport-protocols]

### gRPC [#grpc]

* **Publish:** `SendEvent(pb.Event)` — unary RPC
* **Publish Stream:** `SendEventsStream()` — bidirectional streaming for high-throughput publishing
* **Subscribe:** `SubscribeToEvents(pb.Subscribe)` — server-streaming RPC
* Default port: `50000`
* Max message size: \~1 GB (configurable)

### REST [#rest]

* **Publish:** `POST /send` with JSON body
* **Subscribe:** WebSocket upgrade for streaming delivery
* Default port: `9090`
* Max body size: 100 MB (configurable)

### WebSocket [#websocket]

* Publish and subscribe over persistent WebSocket connections
* JSON-encoded messages
* Read limit: 1 MB

## Configuration [#configuration]

### Server Configuration [#server-configuration]

| Setting                     | Default | Description           |
| --------------------------- | ------- | --------------------- |
| `Connectors.Grpc.BodyLimit` | \~1 GB  | Max gRPC message size |
| `Connectors.Rest.BodyLimit` | 100 MB  | Max REST body size    |

### Client Configuration [#client-configuration]

<TypeTable
  type="{
  address: { type: 'string', description: 'KubeMQ server address.', default: 'localhost:50000' },
  clientId: { type: 'string', description: 'Unique client identifier.', default: 'auto-generated' },
  reconnect: { type: 'boolean', description: 'Auto-reconnect on connection loss.', default: 'true' },
  tls: { type: 'TLSConfig', description: 'TLS/mTLS configuration.', default: 'disabled' },
}"
/>

## Slow Consumer Handling [#slow-consumer-handling]

When a subscriber's receive buffer is full, KubeMQ waits up to the **write deadline** (default 2 seconds) for the buffer to clear. If the deadline expires:

* The event is **dropped** for that subscriber
* A warning is logged server-side with the channel, event ID, and metadata
* Other subscribers are not affected

See [Handle Slow Consumers](/learn/events/how-to/handle-slow-consumers) for mitigation strategies.

## Delivery Semantics [#delivery-semantics]

| Aspect             | Behavior                                          |
| ------------------ | ------------------------------------------------- |
| Delivery guarantee | **At-most-once**                                  |
| Persistence        | None — events flow through memory only            |
| Ordering           | Events are delivered in publish order per channel |
| Duplicates         | No duplicates (single delivery attempt)           |
| Acknowledgment     | None — fire-and-forget                            |
| Retry              | None — failed deliveries are not retried          |

## Events vs Events Store [#events-vs-events-store]

| Feature                | Events                             | Events Store                      |
| ---------------------- | ---------------------------------- | --------------------------------- |
| Persistence            | No                                 | Yes (disk-backed)                 |
| Replay                 | No                                 | Yes (from offset, time, sequence) |
| Delivery guarantee     | At-most-once                       | At-least-once                     |
| Wildcard subscriptions | Yes (`*`, `>`)                     | No                                |
| Consumer groups        | Yes (round-robin)                  | Yes (durable)                     |
| Latency                | Lowest                             | Slightly higher (disk write)      |
| Use cases              | Real-time notifications, streaming | Audit trails, event sourcing      |

## Error Codes [#error-codes]

| Code | Error                     | Description                                |
| ---- | ------------------------- | ------------------------------------------ |
| 101  | Invalid ClientID          | ClientID is empty                          |
| 102  | Invalid Channel           | Channel is empty                           |
| 107  | Invalid Wildcards         | Channel contains `*` or `>` (publish only) |
| 108  | Invalid Whitespace        | Channel contains spaces                    |
| 110  | Invalid Message           | Both `Body` and `Metadata` are empty       |
| 119  | Invalid Channel Separator | Channel ends with `.`                      |

### Runtime Errors [#runtime-errors]

| Error                      | Cause                                | Resolution                               |
| -------------------------- | ------------------------------------ | ---------------------------------------- |
| `ErrShutdownMode`          | Server is shutting down              | Reconnect after server restart           |
| `ErrConnectionNoAvailable` | broker connection is down            | SDK auto-reconnects; check server health |
| Authorization denied       | Casbin policy rejected the operation | Verify client permissions                |

## SDK Quick Reference [#sdk-quick-reference]

<Tabs groupId="language" items="['Go', 'Python', 'Node.js', 'Java', 'C#', 'Kotlin', 'C++', 'Rust', 'Ruby', 'Elixir']">
  <Tab value="Go">
    ```go
    // Publish
    client.SendEvent(ctx, kubemq.NewEvent().
        SetChannel("ch").SetBody([]byte("data")))

    // Subscribe
    client.SubscribeToEvents(ctx, "ch", "group",
        kubemq.WithOnEvent(handler),
        kubemq.WithOnError(errHandler))

    // Stream Publish
    streamCh := make(chan *kubemq.Event, 100)
    resultCh := make(chan *kubemq.EventSendResult, 100)
    go client.StreamEvents(ctx, streamCh, resultCh)
    ```
  </Tab>

  <Tab value="Python">
    ```python
    # Publish
    client.send_event(EventMessage(channel="ch", body=b"data"))

    # Subscribe
    client.subscribe_to_events(
        EventsSubscription(channel="ch", group="group",
            on_receive_event_callback=handler,
            on_error_callback=err_handler),
        cancel=CancellationToken())

    # Stream Publish
    stream = client.open_events_stream()
    stream.send(EventMessage(channel="ch", body=b"data"))
    ```
  </Tab>

  <Tab value="Node.js">
    ```javascript
    // Publish
    await client.sendEvent({ channel: "ch", body: Buffer.from("data") });

    // Subscribe
    client.subscribeToEvents({
      channel: "ch", group: "group",
      onEvent: handler, onError: errHandler
    });

    // Stream Publish
    const stream = client.createEventStream();
    stream.send({ channel: "ch", body: Buffer.from("data") });
    ```
  </Tab>

  <Tab value="Java">
    ```java
    // Publish
    client.sendEventsMessage(EventMessage.builder()
        .channel("ch").body("data".getBytes()).build());

    // Subscribe
    client.subscribeToEvents(EventsSubscription.builder()
        .channel("ch").group("group")
        .onReceiveEventCallback(handler)
        .onErrorCallback(errHandler).build());

    // Stream Publish
    EventsStream stream = client.openEventsStream();
    stream.send(EventMessage.builder()
        .channel("ch").body("data".getBytes()).build());
    ```
  </Tab>

  <Tab value="C#">
    ```csharp
    // Publish
    await client.SendEventAsync(new EventMessage {
        Channel = "ch", Body = Encoding.UTF8.GetBytes("data") });

    // Subscribe
    await foreach (var msg in client.SubscribeToEventsAsync(
        new EventsSubscription { Channel = "ch", Group = "group" })) { }

    // Stream Publish
    var stream = client.OpenEventsStream();
    await stream.SendAsync(new EventMessage {
        Channel = "ch", Body = Encoding.UTF8.GetBytes("data") });
    ```
  </Tab>

  <Tab value="Kotlin">
    ```kotlin
    // Publish
    client.sendEvent(EventMessage(
        channel = "ch", body = "data".toByteArray()))

    // Subscribe
    client.subscribeToEvents(
        channel = "ch", group = "group",
        onEvent = handler, onError = errHandler)

    // Stream Publish
    val stream = client.openEventsStream()
    stream.send(EventMessage(channel = "ch", body = "data".toByteArray()))
    ```
  </Tab>

  <Tab value="C++">
    ```cpp
    // Publish
    kubemq::EventMessage event;
    event.channel = "ch";
    event.body = "data";
    client.sendEvent(event);

    // Subscribe
    client.subscribeToEvents("ch", "group", handler, errHandler);

    // Stream Publish
    auto stream = client.openEventsStream();
    stream.send(event);
    ```
  </Tab>

  <Tab value="Rust">
    ```rust
    // Publish
    let event = EventBuilder::new()
        .channel("ch").body(b"data".to_vec()).build();
    client.send_event(event).await?;

    // Subscribe
    let sub = client.subscribe_to_events("ch", "group",
        |event| Box::pin(async move { handle(event).await }),
        None).await?;

    // Stream Publish
    let mut stream = client.send_event_stream().await?;
    stream.send(EventBuilder::new()
        .channel("ch").body(b"data".to_vec()).build()).await?;
    ```
  </Tab>

  <Tab value="Ruby">
    ```ruby
    # Publish
    client.send_event(KubeMQ::PubSub::EventMessage.new(
      channel: "ch", body: "data"))

    # Subscribe
    sub = KubeMQ::PubSub::EventsSubscription.new(channel: "ch", group: "group")
    client.subscribe_to_events(sub, cancellation_token: cancel,
      on_error: ->(e) { handle_error(e) }) { |event| handle(event) }

    # Stream Publish
    sender = client.create_events_sender
    sender.publish(KubeMQ::PubSub::EventMessage.new(
      channel: "ch", body: "data"))
    ```
  </Tab>

  <Tab value="Elixir">
    ```elixir
    # Publish
    event = KubeMQ.Event.new(channel: "ch", body: "data")
    KubeMQ.Client.send_event(client, event)

    # Subscribe
    {:ok, sub} = KubeMQ.Client.subscribe_to_events(client, "ch",
      group: "group", on_event: fn event -> handle(event) end)

    # Stream Publish
    {:ok, handle} = KubeMQ.Client.send_event_stream(client)
    KubeMQ.EventStreamHandle.send(handle,
      KubeMQ.Event.new(channel: "ch", body: "data"))
    ```
  </Tab>
</Tabs>

## Related [#related]

* [Getting Started with Events](/learn/events/getting-started)
* [Events Store Reference](/learn/events-store/reference) for the persistent variant
* [Queues Reference](/learn/queues/reference) for guaranteed delivery