WebSocket 实时流

CoreC 提供五个 WebSocket 端点，用于实时推送数据点、日志、流量统计、内存信息与告警事件。所有端点均通过 HTTP GET 升级为 WebSocket 连接，使用 github.com/coder/websocket 库实现。

所有 WebSocket 端点均需认证。

连接方式

鉴权

WebSocket 端点同样受认证中间件保护。由于浏览器 WebSocket API 不支持自定义请求头，推荐通过 Query 参数 传递 Token：

wscat（命令行）浏览器 JavaScriptAuthorization Header（非浏览器客户端）

bash

wscat -c "ws://localhost:9090/tags/stream?token=corec-secret-token"

const ws = new WebSocket('ws://localhost:9090/tags/stream?token=corec-secret-token')
ws.onmessage = (event) => {
  const point = JSON.parse(event.data)
  console.log(point)
}

bash

# 部分客户端库支持自定义 Header
wscat -c "ws://localhost:9090/tags/stream" \
  -H "Authorization: Bearer corec-secret-token"

通用行为

行为	说明
协议	WebSocket（RFC 6455）
消息格式	JSON 文本帧
读取方向	服务端关闭读取循环（`CloseRead`），不处理客户端发送的消息
关闭	客户端断开连接时，服务端以 `StatusNormalClosure` 关闭
写入超时	每条消息写入超时 5 秒

/tags/stream — 数据点实时流

实时推送引擎采集的每一个 DataPoint。数据来自核心 Subscribe 机制，在应用坏质量策略之后、规则匹配之前广播给订阅者——订阅者收到的是原始数据点，与规则匹配结果无关。

请求

http

GET /tags/stream?driver={name}

查询参数

参数	类型	必填	说明
`driver`	string	否	仅订阅指定驱动名称的数据点；省略时订阅全部驱动

消息格式

每条消息为一个 DataPoint 对象（JSON 文本帧）：

json

{
  "driver": "plc1",
  "device": "",
  "group": "g1",
  "tag": "temperature",
  "value": 42.5,
  "type": "float64",
  "quality": 0,
  "timestamp": "2024-09-08T10:30:00.123456789Z",
  "metadata": {
    "source": "register-40001"
  }
}

DataPoint 字段说明

字段	类型	说明
`driver`	string	采集驱动名称
`device`	string	设备标识
`group`	string	采集分组
`tag`	string	标签名称
`value`	any	标签值
`type`	string	数据类型枚举（`DataType` 实现自定义 `MarshalJSON`，按字符串序列化）：`bool, int8, int16, …, float32, float64, string, bytes`
`quality`	int	数据质量枚举（`Quality` 为 `int`，按整数序列化）：`0=good, 1=bad, 2=uncertain`
`timestamp`	string (RFC 3339)	采集时间戳
`metadata`	object	附加元数据（可选，存在时才出现）

示例

订阅全部驱动按驱动过滤浏览器示例

bash

wscat -c "ws://localhost:9090/tags/stream?token=corec-secret-token"

bash

wscat -c "ws://localhost:9090/tags/stream?driver=plc1&token=corec-secret-token"

const ws = new WebSocket(
  'ws://localhost:9090/tags/stream?driver=plc1&token=corec-secret-token'
)

ws.onopen = () => console.log('已连接数据流')
ws.onmessage = (e) => {
  const point = JSON.parse(e.data)
  console.log(`[${point.tag}] = ${point.value} (${point.quality})`)
}
ws.onerror = (e) => console.error('连接错误', e)
ws.onclose = () => console.log('连接已关闭')

/logs — 日志实时流

实时推送核心日志事件。数据来自日志总线 log.Subscribe()，源通道缓冲为 1024（DefaultLogBufferSize），每个订阅者通道缓冲为 128（observable.DefaultSubscriberBuffer）。

请求

http

GET /logs

消息格式

每条消息为一个日志 Event 对象：

json

{
  "level": 0,
  "type": "info",
  "payload": "RESTful API listening at 0.0.0.0:9090",
  "timestamp": "2024-09-08T10:30:00.123456789Z"
}

Event 字段说明

字段	类型	说明
`level`	int	日志级别数值（`slog.Level`，按整数序列化），见下表
`type`	string	日志类型字符串（`debug`/`info`/`warning`/`error`）
`payload`	string	日志消息内容
`timestamp`	string (RFC 3339)	日志产生时间

日志级别（level）

level 为 slog.Level（int），按整数序列化：

数值	常量	字符串
`-4`	`LevelDebug`	`debug`
`0`	`LevelInfo`	`info`
`4`	`LevelWarn`	`warning`
`8`	`LevelError`	`error`
`18`	`LevelSilent`	`silent`

级别过滤

日志级别由全局配置 global.log-level 控制。低于设定级别的事件不会产生，因此也不会通过 WebSocket 推送。可通过 PATCH /configs 动态调整日志级别。

示例

bash

wscat -c "ws://localhost:9090/logs?token=corec-secret-token"

/alerts/stream — 一等告警实时流

服务端告警注册表（audit OBS-4/HUB-6）的事件流：每次告警触发/确认（ack）/清除（静默到期）推送一个 AlertEvent 对象。字段与语义详见告警 API（GET /alerts 活动告警 + POST /alerts/ack 确认 + 历史）。

连接方式与 /logs 相同（Bearer 认证，?token= 查询串亦受支持）。消息格式：

json

{
  "type": "fired | acked | cleared",
  "alert": { "key": "high-temp|plc1|temp", "rule": "high-temp", "driver": "plc1", "tag": "temp", "count": 3, "acked": true, "last_fired": "2026-10-21T12:00:00Z" },
  "at": "2026-10-21T12:00:00Z"
}

/traffic — 流量统计实时流

定期推送引擎吞吐速率（读取/发布/丢弃，每秒），服务端完成差分——客户端拿到的就是可直接绘制的速率曲线，没有重启归零导致的负值帧，也没有"第一帧之前是死区"的问题。

请求

http

GET /traffic?interval={duration}&mode={totals|rate}

查询参数

参数	类型	必填	说明
`interval`	string (duration)	否	推送间隔，如 `1s`、`500ms`；省略或非法值时默认 `1s`
`mode`	string	否	`rate`（默认）：速率帧；`totals`：历史累计计数帧（旧客户端的差分用法兼容）

消息格式（默认，速率）

每次间隔推送一个速率对象（单位：次/秒，按 interval 差分）：

json

{
  "read_rate": 125.0,
  "publish_rate": 124.8,
  "dropped_rate": 0.0
}

连接后的第一帧没有差分基线：read_rate 携带引擎的 60s 窗口速率（与 corec_points_per_second 同口径），publish_rate/dropped_rate 为 0——永不出现负值或空洞。

消息格式（`mode=totals`，兼容）

json

{
  "read": 45000,
  "publish": 44997,
  "dropped": 0
}

字段说明

字段	类型	说明
`read_rate`	float	读取速率（次/秒）
`publish_rate`	float	发布速率（次/秒）
`dropped_rate`	float	丢弃速率（次/秒）
`read`	uint64	仅 `mode=totals`：累计读取（`EngineStats.TotalRead`）
`publish`	uint64	仅 `mode=totals`：累计发布
`dropped`	uint64	仅 `mode=totals`：累计丢弃

推送频率

默认每 1 秒推送一次，可通过 ?interval= 自定义间隔。

示例

wscat浏览器仪表盘

bash

wscat -c "ws://localhost:9090/traffic?token=corec-secret-token"

const ws = new WebSocket('ws://localhost:9090/traffic?token=corec-secret-token')

ws.onmessage = (e) => {
  const { read_rate, publish_rate, dropped_rate } = JSON.parse(e.data)
  document.getElementById('read').textContent = read_rate
  document.getElementById('publish').textContent = publish_rate
  document.getElementById('dropped').textContent = dropped_rate
}

/memory — 内存监控实时流

定期推送 Go 运行时内存统计，用于监控核心内存占用和 GC 活动。数据经 runtime/metrics 采样（不触发 Stop-The-World 停顿），推送本身不会造成延迟毛刺。

请求

http

GET /memory?interval={duration}

查询参数

参数	类型	必填	说明
`interval`	string (duration)	否	推送间隔，如 `1s`、`2s`；省略或非法值时默认 `1s`

消息格式

每次间隔推送一个内存统计对象：

json

{
  "alloc": 12345678,
  "total_alloc": 98765432,
  "sys": 52428800,
  "num_gc": 42,
  "goroutines": 15
}

字段说明

字段	类型	说明
`alloc`	uint64	当前堆内对象字节数（`/memory/classes/heap/objects:bytes`，等价 `MemStats.HeapAlloc`）
`total_alloc`	uint64	累计分配字节数（`/gc/heap/allocs:bytes`，等价 `MemStats.TotalAlloc`）
`sys`	uint64	从操作系统获取的内存总字节数（`/memory/classes/total:bytes`，等价 `MemStats.Sys`）
`num_gc`	uint64	GC 完成次数（`/gc/cycles/total:gc-cycles`，等价 `MemStats.NumGC`）
`goroutines`	int	当前 goroutine 数量（`runtime.NumGoroutine()`）

推送频率

默认每 1 秒推送一次（time.NewTicker(wsPushInterval)），可通过 ?interval= 查询参数自定义间隔。

示例

wscat浏览器仪表盘

bash

wscat -c "ws://localhost:9090/memory?token=corec-secret-token"

const ws = new WebSocket('ws://localhost:9090/memory?token=corec-secret-token')

ws.onmessage = (e) => {
  const { alloc, sys, num_gc, goroutines } = JSON.parse(e.data)
  console.log(`内存: ${(alloc / 1048576).toFixed(1)} MB`)
  console.log(`GC 次数: ${num_gc}, Goroutines: ${goroutines}`)
}

连接生命周期

所有 WebSocket 端点遵循相同的连接生命周期：

客户端                          服务端
  │                               │
  │── HTTP GET (Upgrade) ────────→│  websocket.Accept()
  │←─ 101 Switching Protocols ────│
  │                               │
  │                               │  CloseRead(ctx)  ← 关闭读取循环
  │                               │
  │←─ JSON 消息帧 ────────────────│  wsjson.Write()  ← 持续推送
  │←─ JSON 消息帧 ────────────────│
  │←─ JSON 消息帧 ────────────────│
  │                               │
  │── Close Frame ───────────────→│  ctx.Done()
  │←─ Close Frame ────────────────│  StatusNormalClosure
  │                               │

关闭码

关闭码	说明
`1000` (Normal Closure)	客户端正常断开
`1011` (Internal Error)	服务端内部错误（defer 默认关闭码）

重连建议

客户端应实现自动重连机制，在连接断开后使用指数退避策略重新建立连接。建议初始延迟 1 秒，最大延迟 30 秒。

WebSocket 实时流 ​

连接方式 ​

鉴权 ​

通用行为 ​

/tags/stream — 数据点实时流 ​

请求 ​

查询参数 ​

消息格式 ​

DataPoint 字段说明 ​

示例 ​

/logs — 日志实时流 ​

请求 ​

消息格式 ​

Event 字段说明 ​

日志级别（level） ​

示例 ​

/alerts/stream — 一等告警实时流 ​

/traffic — 流量统计实时流 ​

请求 ​

查询参数 ​

消息格式（默认，速率） ​

消息格式（mode=totals，兼容） ​

字段说明 ​

示例 ​

/memory — 内存监控实时流 ​

请求 ​

查询参数 ​

消息格式 ​

字段说明 ​

示例 ​

连接生命周期 ​

关闭码 ​

WebSocket 实时流

连接方式

鉴权

通用行为

/tags/stream — 数据点实时流

请求

查询参数

消息格式

DataPoint 字段说明

示例

/logs — 日志实时流

请求

消息格式

Event 字段说明

日志级别（level）

示例

/alerts/stream — 一等告警实时流

/traffic — 流量统计实时流

请求

查询参数

消息格式（默认，速率）

消息格式（`mode=totals`，兼容）

字段说明

示例

/memory — 内存监控实时流

请求

查询参数

消息格式

字段说明

示例

连接生命周期

关闭码