# WebSocket Protocols (/operate/observability/api-reference/websockets) KubeMQ's management API exposes several WebSocket endpoints on port `8080` for real-time data: a continuous cluster-snapshot stream that keeps the dashboard in sync, a live message monitor for any channel, pub/sub and CQRS subscriptions, and a transactional queue-streaming protocol. All of them use the standard WebSocket upgrade handshake over HTTP `GET`. This page documents the client-facing protocol for each endpoint — the path, query parameters, and the JSON frames exchanged. For the full schema of every message frame, see [Data Models](/operate/observability/api-reference/data-models). For the response envelope and status-code rules, see the [Management API overview](/operate/observability/api-reference). ## Readiness gating [#readiness-gating] Most WebSocket endpoints under `/api/` are gated by the API readiness check. If the service is not ready, the upgrade does **not** happen — the server returns the standard error envelope as an ordinary HTTP response instead: ```json { "error": true, "error_string": "api service not ready", "data": null } ``` The only exception is `GET /api/monitor` (and its alias `GET /v1/stats/attach`), which is available as soon as the HTTP server starts. ## GET /api/connection [#get-apiconnection] A continuous stream of the latest cluster snapshot, pushed every 2 seconds. This is the primary endpoint for keeping the dashboard UI in sync — open it once on load and use the incoming messages to update the entire dashboard state, instead of polling `GET /api/cluster-snapshot`. ```text ws://:8080/api/connection ``` No query parameters are required. ### Protocol [#protocol] * The stream is **server → client only**; no client-to-server messages are expected. * Every 2 seconds the server pushes one complete `SnapshotDTO` as a JSON text frame. * The connection stays open until the client disconnects or the server shuts down. ### Server frames [#server-frames] Each frame is a complete `SnapshotDTO`, the same schema returned by `GET /api/cluster-snapshot` (see [Data Models → Snapshot DTOs](/operate/observability/api-reference/data-models)): ```json { "hostsInfo": [], "topologyNodes": {}, "trafficChannels": {}, "statsCards": {}, "topChannels": [], "lastActivity": 1709312400000, "lastActivityHuman": "2 seconds ago", "total": {}, "incoming": {}, "outgoing": {}, "channels": 10, "channelsHuman": "10", "clients": 8, "clientsHuman": "8", "activeChannels": 5, "queues": {}, "pubsub": {}, "commandsQueries": {}, "nodeType": "standalone", "node": "kubemq-node-0" } ``` ### Behavior notes [#behavior-notes] * If a snapshot is not yet available, no frame is sent for that tick. * The 2-second interval is fixed and not configurable. * The connection closes cleanly when the server is shutting down. ## GET /api/monitor [#get-apimonitor] A live monitor for a single channel — it forwards every message or request passing through the channel to the client. The alias `GET /v1/stats/attach` uses the same handler. Neither is gated by the readiness guard, so both are available as soon as the HTTP server starts. ```text ws://:8080/api/monitor?channel=&kind=&max_size= ``` ### Query parameters [#query-parameters] | Parameter | Required | Description | | ---------- | -------- | ----------------------------------------------------------------------------------------------------- | | `channel` | Yes | Channel name to monitor | | `kind` | Yes | Channel type: `events`, `events_store`, `commands`, `queries`, or `queue` | | `max_size` | No | Maximum body size in bytes. Parsed but **not currently enforced** — bodies are forwarded at full size | ### Start/stop control flow [#startstop-control-flow] The monitor does **not** stream automatically on connect. The client drives the data flow with text control messages: **Open the WebSocket** with the `channel` and `kind` query parameters. **Send `"start"`** as a text message to begin receiving data. The server replies with `" Broker connected."` to confirm, then begins forwarding monitored messages as JSON text frames. **Send `"stop"`** to pause the flow. The server replies with `"Broker disconnected."` and stops forwarding. Send `"start"` again to resume. | Client message | Server reply | Effect | | ---------------- | -------------------------------- | ------------------------------------ | | `"start"` | `" Broker connected."` | Begins forwarding monitored messages | | `"stop"` | `"Broker disconnected."` | Pauses forwarding | | Any other string | `"Unknown command"` | No state change | If you open the WebSocket but never send `"start"`, you will never receive any data and the connection will appear stuck. Always send `"start"` immediately after the upgrade. ### Server frames [#server-frames-1] Monitor frames are pretty-printed JSON objects whose shape depends on the `kind` being monitored. There are five distinct transport types, and there is no wrapping envelope or `type` discriminator — the client identifies the type by the fields present. | Monitored `kind` | Frame type | | ------------------------------- | ---------------------------------- | | `events`, `events_store` | `TransportPubSubMessageDto` | | `commands`, `queries` | `TransportRequestMessageDto` | | Response delivered to a request | `TransportResponseMessageDto` | | Request that ended in error | `TransportResponseErrorMessageDto` | | `queue` | `TransportQueueMessageDto` | The full field schema for each of these is documented in [Data Models → Monitor transports](/operate/observability/api-reference/data-models). An events frame looks like: ```json { "messageId": "evt-001", "metadata": "event metadata", "body": { "event": "user.created" }, "timestamp": "2024-03-01 12:00:05", "sequence": 42, "tags": "{\"source\":\"api\"}" } ``` ### Error handling [#error-handling] * A missing or invalid `kind`, or a missing `channel`, fails validation and no WebSocket upgrade occurs. * Transport errors during monitoring may arrive as plain-text WebSocket frames; in some failure paths the connection closes without a message. ## GET /api/subscribe/pubsub [#get-apisubscribepubsub] Subscribes to an Events or Events Store channel and streams received messages to the client. ```text ws://:8080/api/subscribe/pubsub?subscribe_type=&channel=&group=&events_store_type_data=&events_store_type_value= ``` ### Query parameters [#query-parameters-1] | Parameter | Required | Description | | ------------------------- | ----------------------------- | ----------------------------------------------- | | `subscribe_type` | Yes | `events` or `events_store` | | `channel` | Yes | Channel name to subscribe to | | `group` | No | Consumer group name, for load-balanced delivery | | `events_store_type_data` | Only for `events_store` | Subscription start position (see below) | | `events_store_type_value` | Depends on the start position | Value for the start position | ### Events Store start positions [#events-store-start-positions] For `events_store` subscriptions, `events_store_type_data` selects where replay begins: | Value | Name | `events_store_type_value` | Description | | ----- | ---------------- | --------------------------- | ------------------------------------- | | `1` | StartNewOnly | Not used | Only new messages from now on | | `2` | StartFromFirst | Not used | Replay from the first stored message | | `3` | StartFromLast | Not used | Start from the last stored message | | `4` | StartAtSequence | Sequence number (`int`) | Start from a specific sequence number | | `5` | StartAtTime | ISO 8601 / RFC3339 time | Start from a specific timestamp | | `6` | StartAtTimeDelta | Duration in seconds (`int`) | Start from N seconds ago | ### Protocol [#protocol-1] * The server streams received messages as JSON text frames. * The stream is effectively server → client; the client may send frames, but they are ignored — the read loop only keeps the connection alive. * When the client disconnects, the subscription is cleaned up automatically. ### Server frames [#server-frames-2] Each frame is a `SubscribePubSubMessage`: ```json { "messageId": "evt-001", "metadata": "event metadata", "body": { "event": "user.created", "userId": 123 }, "timestamp": 1709312400000, "tags": "{\"source\":\"api\"}", "sequence": 42 } ``` | Field | Type | Description | | ----------- | -------- | ----------------------------------------------------------------- | | `messageId` | `string` | Event/message ID | | `metadata` | `string` | Message metadata | | `body` | `any` | Message body (auto-detected: string, JSON object, or JSON array) | | `timestamp` | `int64` | Timestamp in Unix milliseconds | | `tags` | `string` | Tags as a JSON string (empty if none) | | `sequence` | `int64` | Sequence number — meaningful for `events_store`, `0` for `events` | ### Errors [#errors] If a subscription error occurs, it is sent as a plain-text WebSocket frame: ```text connection closed, reason: ``` If the initial subscription fails before the upgrade, the error is returned as a standard HTTP envelope response instead. ### Examples [#examples] ```text ws://localhost:8080/api/subscribe/pubsub?subscribe_type=events&channel=notifications&group= ws://localhost:8080/api/subscribe/pubsub?subscribe_type=events_store&channel=audit-log&events_store_type_data=4&events_store_type_value=100 ws://localhost:8080/api/subscribe/pubsub?subscribe_type=events_store&channel=audit-log&events_store_type_data=6&events_store_type_value=300 ``` ## GET /api/subscribe/cqrs [#get-apisubscribecqrs] Subscribes to a Commands or Queries channel and streams received requests to the client. ```text ws://:8080/api/subscribe/cqrs?subscribe_type=&channel=&group= ``` ### Query parameters [#query-parameters-2] | Parameter | Required | Description | | ---------------- | -------- | ---------------------------- | | `subscribe_type` | Yes | `commands` or `queries` | | `channel` | Yes | Channel name to subscribe to | | `group` | No | Consumer group name | ### Protocol [#protocol-2] * The server streams received requests as JSON text frames. * To answer a request, the client calls the HTTP action endpoint with the `send_cqrs_message_response` action — see [Action Endpoint](/operate/observability/api-reference/actions). * When the client disconnects, the subscription is cleaned up. ### Server frames [#server-frames-3] Each frame is a `SubscribeCQRSRequestMessage`: ```json { "requestId": "req-001", "metadata": "get-user", "body": { "userId": 123 }, "timestamp": 1709312400000, "replyChannel": "reply-abc123", "tags": "{\"version\":\"2\"}", "isCommand": false } ``` | Field | Type | Description | | -------------- | --------- | ---------------------------------------------------------------- | | `requestId` | `string` | Request ID — use it in the response | | `metadata` | `string` | Request metadata | | `body` | `any` | Request body (auto-detected format) | | `timestamp` | `int64` | Timestamp in Unix milliseconds | | `replyChannel` | `string` | Reply channel — **pass it back** in `send_cqrs_message_response` | | `tags` | `string` | Tags as a JSON string | | `isCommand` | `boolean` | `true` for a command, `false` for a query | ### Responding to a request [#responding-to-a-request] To respond to a received command or query, call the HTTP action endpoint with the `replyChannel` value taken verbatim from the received frame: ```json { "type": "send_cqrs_message_response", "data": { "requestId": "req-001", "replyChannel": "reply-abc123", "body": { "result": "user data" }, "executed": true } } ``` Subscription errors are reported the same way as for pub/sub — as plain-text WebSocket frames. ## GET /api/stream\_queue\_messages [#get-apistream_queue_messages] A bidirectional stream for consuming a queue one message at a time with manual acknowledge or reject. This implements a transactional queue-consumption pattern: only one message is in flight at a time, and the client must ack or reject it before requesting the next. ```text ws://:8080/api/stream_queue_messages ``` No query parameters are required — all configuration is sent over the WebSocket. ### Protocol [#protocol-3] **Open the WebSocket** and send a `stream_queue_messages` request to poll for the next message. **Receive the message.** The server responds with the next message (waiting up to one hour for one to arrive). **Ack or reject it.** Inspect the message, then send `ack_queue_messages` or `reject_queue_messages` with its sequence number. The server confirms. **Repeat.** Send another `stream_queue_messages` request to fetch the next message. ### Client frames — `StreamQueueMessagesRequest` [#client-frames--streamqueuemessagesrequest] Poll for the next message: ```json { "requestType": "stream_queue_messages", "channel": "orders.process", "visibilitySeconds": 30, "waitSeconds": 60 } ``` Acknowledge the current message: ```json { "requestType": "ack_queue_messages", "refSequence": 42 } ``` Reject the current message: ```json { "requestType": "reject_queue_messages", "refSequence": 42 } ``` | Field | Type | Required | Description | | ------------------- | -------- | --------------- | --------------------------------------------------------------------------------------------- | | `requestType` | `string` | Yes | `stream_queue_messages`, `ack_queue_messages`, or `reject_queue_messages` | | `channel` | `string` | Poll only | Queue channel to stream from | | `visibilitySeconds` | `int` | Poll only | Validated to be `> 0`, but not applied server-side — send a positive value to pass validation | | `waitSeconds` | `int` | Poll only | Validated to be `> 0`, but not applied server-side — send a positive value to pass validation | | `refSequence` | `int64` | Ack/reject only | Sequence number of the message being acked/rejected (must be `> 0`) | `visibilitySeconds` and `waitSeconds` must be `> 0` to pass request validation, but the server does not use them to control timing — each poll uses a fixed one-message, one-hour-wait downstream request. Send valid values, but do not expect them to change server-side behavior. ### Server frames — `StreamQueueMessagesResponse` [#server-frames--streamqueuemessagesresponse] A poll response carries the received message: ```json { "requestType": "stream_queue_messages", "message": { "messageId": "msg-001", "clientId": "order-service", "metadata": "order data", "body": { "orderId": 123 }, "timestamp": 1709312400000, "sequence": 42, "tags": "", "receivedCount": 1, "reRoutedFrom": "", "expirationAt": 0, "delayedTo": 0 }, "error": "", "isError": false } ``` An ack/reject response carries a `null` message: ```json { "requestType": "ack_queue_messages", "message": null, "error": "", "isError": false } ``` | Field | Type | Description | | ------------- | ---------------- | ----------------------------------------------------- | | `requestType` | `string` | Echoes the request type that triggered the response | | `message` | object or `null` | The received message; `null` for ack/reject responses | | `error` | `string` | Error message when `isError` is `true` | | `isError` | `boolean` | Whether this response indicates an error | ### Error scenarios [#error-scenarios] | Error | Cause | | -------------------------------------------------------------------- | --------------------------------------------------- | | `error polling queue messages, previous request is not finished yet` | Polled while a message was still pending ack/reject | | `no active stream` | Tried to ack/reject with no pending message | | `the broker is not ready to accept traffic` | The message broker is not ready | ## Related [#related]