WebSocket Event Stream

EnerOS provides bidirectional real-time communication channels via WebSocket, used for scenarios such as Agent status push, grid event broadcast, and remote command dispatch. The underlying events are generated by eneros-eventbus, authenticated by eneros-gateway, and then pushed to subscribed clients. The WebSocket protocol is based on HTTP upgrade handshake, and the connection can be maintained as a long connection after establishment. Compared with REST’s repeated connection establishment, it can significantly reduce latency and resource consumption.

Endpoint Information

Connection Establishment

Handshake Flow

Client                                          Server
  │                                               │
  │── HTTP Upgrade + Sec-WebSocket-Protocol ────→ │
  │                                               │
  │←── 101 Switching Protocols ──────────────────│
  │                                               │
  │── auth frame (with token) ──────────────────→ │
  │                                               │
  │←── auth_ok frame ────────────────────────────│
  │                                               │
  │── subscribe frame (subscribe topics) ───────→ │
  │                                               │
  │←── subscribed frame (subscription confirmed) ─│
  │                                               │
  │←── event frame (event push) ─────────────────│
  │                                               │
  │── ping frame (every 30s) ───────────────────→ │
  │                                               │
  │←── pong frame ───────────────────────────────│

JavaScript Handshake Example

const ws = new WebSocket('ws://localhost:8080/ws', ['eneros.v1']);

ws.onopen = () => {
  console.log('Connection established, sending authentication');
  ws.send(JSON.stringify({
    type: 'auth',
    token: '<bearer-token>',
    client_id: 'dashboard-001',
    client_version: '1.0.0',
  }));
};

ws.onmessage = (event) => {
  const msg = JSON.parse(event.data);
  switch (msg.type) {
    case 'auth_ok':
      console.log('Authentication successful, starting subscription');
      ws.send(JSON.stringify({
        type: 'subscribe',
        topics: ['network.net_8f3a2b', 'agent.*', 'alarm.critical'],
        ack: true,
      }));
      break;
    case 'event':
      console.log('Received event:', msg.payload);
      break;
    case 'error':
      console.error('Error:', msg.payload);
      break;
  }
};

ws.onerror = (e) => console.error('Connection error', e);
ws.onclose = (e) => console.log('Connection closed:', e.code, e.reason);

Message Format

All messages are JSON text frames (binary frames not supported), with unified field conventions as follows:

Message Type Enumeration

auth Frame

{
  "type": "auth",
  "token": "<bearer-token>",
  "client_id": "dashboard-001",
  "client_version": "1.0.0",
  "capabilities": ["ack", "batch"]
}

auth_ok Frame

{
  "type": "auth_ok",
  "session_id": "sess_20260706_001",
  "tenant": "default",
  "expires_at": "2026-07-06T09:30:00Z",
  "server_time": "2026-07-06T08:30:00Z"
}

subscribe Frame

{
  "type": "subscribe",
  "topics": ["network.net_8f3a2b", "agent.*", "alarm.critical"],
  "ack": true,
  "last_event_id": "evt_20260706_001"
}

subscribed Frame

{
  "type": "subscribed",
  "topics": ["network.net_8f3a2b", "agent.*", "alarm.critical"],
  "pending_replay": 12
}

event Frame

{
  "type": "event",
  "id": "evt_a1b2c3",
  "topic": "network.net_8f3a2b",
  "timestamp": "2026-07-06T08:30:00.123Z",
  "payload": {
    "kind": "TopologyChanged",
    "network_id": "net_8f3a2b",
    "changes": [{"branch": "1-2", "status": "open"}],
    "actor": {"type": "agent", "id": "agent_dispatch_01"}
  }
}

command Frame

{
  "type": "command",
  "id": "cmd_001",
  "agent_id": "agent_dispatch_01",
  "command": "set_generation",
  "params": {"bus": 2, "mw": 50.0},
  "priority": "normal",
  "timeout_ms": 5000
}

command_result Frame

{
  "type": "command_result",
  "id": "cmd_001",
  "status": "success",
  "result": {"bus": 2, "mw_set": 50.0, "audit_id": "aud_001"},
  "duration_ms": 12,
  "timestamp": "2026-07-06T08:30:00Z"
}

error Frame

{
  "type": "error",
  "code": "TOPIC_DENIED",
  "message": "No permission to subscribe: audit.*",
  "details": {"topic": "audit.*"},
  "timestamp": "2026-07-06T08:30:00Z"
}

Event Topic List

Wildcard rules:

• * matches a single level, e.g., agent.* matches agent.dispatch_01 but not agent.dispatch_01.sub
• # matches multiple levels (MQTT compatibility mode only)

Event Type Enumeration

NetworkEvent

PowerFlowEvent

AgentEvent

AlarmEvent

DeviceEvent

ConstraintEvent

Heartbeat Mechanism

The server sends a WebSocket protocol-level ping frame every 30 seconds, and the client must respond with a pong frame within 60 seconds. If the client uses application-level heartbeats (e.g., browsers cannot control protocol frames), it can send a ping message:

{"type": "ping", "timestamp": "2026-07-06T08:30:00Z"}

Server response:

{"type": "pong", "timestamp": "2026-07-06T08:30:00.001Z"}

Reconnection Mechanism

Clients should implement exponential backoff reconnection after disconnection:

class EnerOSWebSocket {
  constructor(url, token) {
    this.url = url;
    this.token = token;
    this.reconnectAttempts = 0;
    this.maxReconnectAttempts = 10;
    this.lastEventId = null;
    this.subscriptions = new Set();
    this.connect();
  }

  connect() {
    this.ws = new WebSocket(this.url, ['eneros.v1']);
    this.ws.onopen = () => this.onOpen();
    this.ws.onmessage = (e) => this.onMessage(e);
    this.ws.onclose = (e) => this.onClose(e);
    this.ws.onerror = (e) => console.error('WS error', e);
  }

  onOpen() {
    this.reconnectAttempts = 0;
    this.ws.send(JSON.stringify({
      type: 'auth',
      token: this.token,
    }));
  }

  onMessage(event) {
    const msg = JSON.parse(event.data);
    if (msg.type === 'auth_ok') {
      if (this.subscriptions.size > 0) {
        this.ws.send(JSON.stringify({
          type: 'subscribe',
          topics: Array.from(this.subscriptions),
          last_event_id: this.lastEventId,
        }));
      }
    } else if (msg.type === 'event') {
      this.lastEventId = msg.id;
      this.handleEvent(msg);
    }
  }

  onClose(event) {
    if (this.reconnectAttempts >= this.maxReconnectAttempts) {
      console.error('Max reconnection attempts reached, giving up');
      return;
    }
    const delay = Math.min(1000 * Math.pow(2, this.reconnectAttempts), 30000);
    this.reconnectAttempts++;
    console.log(`Reconnecting in ${delay}ms (attempt ${this.reconnectAttempts})`);
    setTimeout(() => this.connect(), delay);
  }

  handleEvent(msg) {
    console.log('Event:', msg);
  }

  subscribe(topic) {
    this.subscriptions.add(topic);
    if (this.ws.readyState === WebSocket.OPEN) {
      this.ws.send(JSON.stringify({type: 'subscribe', topics: [topic]}));
    }
  }

  unsubscribe(topic) {
    this.subscriptions.delete(topic);
    if (this.ws.readyState === WebSocket.OPEN) {
      this.ws.send(JSON.stringify({type: 'unsubscribe', topics: [topic]}));
    }
  }
}

Rust Client Example

use tokio_tungstenite::{connect_async, tungstenite::Message};
use futures_util::{SinkExt, StreamExt};
use serde_json::json;

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let (ws_stream, _) = connect_async("ws://localhost:8080/ws").await?;
    let (mut write, mut read) = ws_stream.split();

    // Authentication
    let auth_msg = json!({
        "type": "auth",
        "token": "<bearer-token>",
        "client_id": "rust-client-01",
    });
    write.send(Message::Text(auth_msg.to_string())).await?;

    // Wait for auth_ok
    if let Some(Ok(Message::Text(text))) = read.next().await {
        let resp: serde_json::Value = serde_json::from_str(&text)?;
        if resp["type"] == "auth_ok" {
            println!("Authentication successful");
        }
    }

    // Subscribe
    let sub_msg = json!({
        "type": "subscribe",
        "topics": ["network.net_8f3a2b", "alarm.critical"],
    });
    write.send(Message::Text(sub_msg.to_string())).await?;

    // Receive event loop
    while let Some(Ok(msg)) = read.next().await {
        match msg {
            Message::Text(text) => {
                let evt: serde_json::Value = serde_json::from_str(&text)?;
                if evt["type"] == "event" {
                    println!(
                        "[{}] {}: {:?}",
                        evt["timestamp"], evt["topic"], evt["payload"]
                    );
                }
            }
            Message::Ping(data) => {
                write.send(Message::Pong(data)).await?;
            }
            _ => {}
        }
    }

    Ok(())
}

Python Client Example

import asyncio
import json
import websockets

async def main():
    uri = "ws://localhost:8080/ws"
    async with websockets.connect(uri, subprotocols=["eneros.v1"]) as ws:
        # Authentication
        await ws.send(json.dumps({
            "type": "auth",
            "token": "<bearer-token>",
            "client_id": "python-client-01",
        }))

        # Wait for auth_ok
        response = json.loads(await ws.recv())
        if response["type"] != "auth_ok":
            print("Authentication failed:", response)
            return

        print("Authentication successful")

        # Subscribe
        await ws.send(json.dumps({
            "type": "subscribe",
            "topics": ["network.net_8f3a2b", "alarm.critical"],
        }))

        # Event loop
        async for message in ws:
            evt = json.loads(message)
            if evt["type"] == "event":
                print(f"[{evt['timestamp']}] {evt['topic']}: {evt['payload']}")
            elif evt["type"] == "ping":
                await ws.send(json.dumps({"type": "pong"}))

asyncio.run(main())

ACK and Reliable Delivery

When ACK is enabled, the client must send an ack frame for each event:

{
  "type": "ack",
  "event_id": "evt_a1b2c3"
}

If no ACK is received within 5 seconds, the server will resend the event (up to 3 times). ACK mode is suitable for critical alarm and control command scenarios.

Error Codes

Close Codes

WebSocket close codes follow RFC 6455, with the following custom close codes extended:

Performance and Limits

Related Documentation

• API Overview — Comparison of four API types
• SSE Real-time Push — Lightweight unidirectional alternative
• GraphQL Subscription — GraphQL subscriptions
• Rate Limit and Quota — WebSocket rate limits
• eneros-eventbus crate — Event bus implementation