WebSocket 事件流

EnerOS 通过 WebSocket 提供双向实时通信通道，用于 Agent 状态推送、电网事件广播、远程指令下发等场景。底层由 eneros-eventbus 产生事件，经 eneros-gateway 鉴权后推送至订阅客户端。WebSocket 协议基于 HTTP 升级握手，连接建立后可保持长连接，相比 REST 反复建连，可显著降低延迟与资源消耗。

端点信息

连接建立

握手流程

客户端                                          服务端
  │                                               │
  │── HTTP Upgrade + Sec-WebSocket-Protocol ────→ │
  │                                               │
  │←── 101 Switching Protocols ──────────────────│
  │                                               │
  │── auth 帧（含 token） ──────────────────────→ │
  │                                               │
  │←── auth_ok 帧 ───────────────────────────────│
  │                                               │
  │── subscribe 帧（订阅主题） ─────────────────→ │
  │                                               │
  │←── subscribed 帧（订阅确认） ────────────────│
  │                                               │
  │←── event 帧（事件推送） ─────────────────────│
  │                                               │
  │── ping 帧（每 30s） ────────────────────────→ │
  │                                               │
  │←── pong 帧 ───────────────────────────────────│

JavaScript 握手示例

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

ws.onopen = () => {
  console.log('连接已建立，发送鉴权');
  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('鉴权成功，开始订阅');
      ws.send(JSON.stringify({
        type: 'subscribe',
        topics: ['network.net_8f3a2b', 'agent.*', 'alarm.critical'],
        ack: true,
      }));
      break;
    case 'event':
      console.log('收到事件:', msg.payload);
      break;
    case 'error':
      console.error('错误:', msg.payload);
      break;
  }
};

ws.onerror = (e) => console.error('连接错误', e);
ws.onclose = (e) => console.log('连接关闭:', e.code, e.reason);

消息格式

所有消息为 JSON 文本帧（不支持二进制帧），统一字段约定如下：

消息类型枚举

auth 帧

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

auth_ok 帧

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

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

subscribed 帧

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

event 帧

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

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

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

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

事件主题列表

通配符规则：

• * 匹配单层，如 agent.* 匹配 agent.dispatch_01 但不匹配 agent.dispatch_01.sub
• # 匹配多层（仅 MQTT 兼容模式）

事件类型枚举

NetworkEvent

PowerFlowEvent

AgentEvent

AlarmEvent

DeviceEvent

ConstraintEvent

心跳机制

服务端每 30 秒发送 WebSocket 协议级 ping 帧，客户端需在 60 秒内回复 pong 帧。若客户端使用应用层心跳（如浏览器无法控制协议帧），可发送 ping 消息：

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

服务端响应：

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

重连机制

客户端断开后应实现指数退避重连：

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('达到最大重连次数，放弃重连');
      return;
    }
    const delay = Math.min(1000 * Math.pow(2, this.reconnectAttempts), 30000);
    this.reconnectAttempts++;
    console.log(`${delay}ms 后重连（第 ${this.reconnectAttempts} 次）`);
    setTimeout(() => this.connect(), delay);
  }

  handleEvent(msg) {
    console.log('事件:', 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 客户端示例

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();

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

    // 等待 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!("鉴权成功");
        }
    }

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

    // 接收事件循环
    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 客户端示例

import asyncio
import json
import websockets

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

        # 等待 auth_ok
        response = json.loads(await ws.recv())
        if response["type"] != "auth_ok":
            print("鉴权失败:", response)
            return

        print("鉴权成功")

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

        # 事件循环
        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 与可靠传输

启用 ACK 后，客户端需对每个事件发送 ack 帧：

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

若 5 秒内未收到 ACK，服务端将重发该事件（最多 3 次）。ACK 模式适用于关键告警与控制指令场景。

错误码

关闭码

WebSocket 关闭码遵循 RFC 6455，并扩展以下自定义关闭码：

性能与限制

相关文档

• API 总览 — 四种 API 对比
• SSE 实时推送 — 单向轻量替代方案
• GraphQL Subscription — GraphQL 订阅
• 限流与配额 — WebSocket 速率限制
• eneros-eventbus crate — 事件总线实现