WebSocket Protocols

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. For the response envelope and status-code rules, see the Management API overview.

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:

{
  "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

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.

ws://<host>:8080/api/connection

No query parameters are required.

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

Each frame is a complete SnapshotDTO, the same schema returned by GET /api/cluster-snapshot (see Data Models → Snapshot DTOs):

{
  "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

• 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

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.

ws://<host>:8080/api/monitor?channel=<channel>&kind=<kind>&max_size=<max_size>

Query parameters

Start/stop control flow

The monitor does not stream automatically on connect. The client drives the data flow with text control messages:

Server frames

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.

The full field schema for each of these is documented in Data Models → Monitor transports. An events frame looks like:

{
  "messageId": "evt-001",
  "metadata": "event metadata",
  "body": { "event": "user.created" },
  "timestamp": "2024-03-01 12:00:05",
  "sequence": 42,
  "tags": "{\"source\":\"api\"}"
}

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

Subscribes to an Events or Events Store channel and streams received messages to the client.

ws://<host>:8080/api/subscribe/pubsub?subscribe_type=<type>&channel=<channel>&group=<group>&events_store_type_data=<sub_type>&events_store_type_value=<sub_value>

Query parameters

Events Store start positions

For events_store subscriptions, events_store_type_data selects where replay begins:

Protocol

• 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

Each frame is a SubscribePubSubMessage:

{
  "messageId": "evt-001",
  "metadata": "event metadata",
  "body": { "event": "user.created", "userId": 123 },
  "timestamp": 1709312400000,
  "tags": "{\"source\":\"api\"}",
  "sequence": 42
}

Errors

If a subscription error occurs, it is sent as a plain-text WebSocket frame:

connection closed, reason: <error details>

If the initial subscription fails before the upgrade, the error is returned as a standard HTTP envelope response instead.

Examples

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

Subscribes to a Commands or Queries channel and streams received requests to the client.

ws://<host>:8080/api/subscribe/cqrs?subscribe_type=<type>&channel=<channel>&group=<group>

Query parameters

Protocol

• 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.
• When the client disconnects, the subscription is cleaned up.

Server frames

Each frame is a SubscribeCQRSRequestMessage:

{
  "requestId": "req-001",
  "metadata": "get-user",
  "body": { "userId": 123 },
  "timestamp": 1709312400000,
  "replyChannel": "reply-abc123",
  "tags": "{\"version\":\"2\"}",
  "isCommand": false
}

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:

{
  "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

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.

ws://<host>:8080/api/stream_queue_messages

No query parameters are required — all configuration is sent over the WebSocket.

Protocol

Client frames — StreamQueueMessagesRequest

Poll for the next message:

{
  "requestType": "stream_queue_messages",
  "channel": "orders.process",
  "visibilitySeconds": 30,
  "waitSeconds": 60
}

Acknowledge the current message:

{
  "requestType": "ack_queue_messages",
  "refSequence": 42
}

Reject the current message:

{
  "requestType": "reject_queue_messages",
  "refSequence": 42
}

Server frames — StreamQueueMessagesResponse

A poll response carries the received message:

{
  "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:

{
  "requestType": "ack_queue_messages",
  "message": null,
  "error": "",
  "isError": false
}

Error scenarios

Related