440 lines
12 KiB
Markdown
440 lines
12 KiB
Markdown
# EFKA 与 Service WebSocket 交互协议
|
||
|
||
本文档描述部署在 `efka` 主机上的 service 与 `efka` 之间的 WebSocket 通讯逻辑。当前接入实现位于：
|
||
|
||
- [service_channel.erl](/usr/local/code/cloudkit/efka/apps/efka/src/service/service_channel.erl)
|
||
- [service.proto](/usr/local/code/cloudkit/efka/apps/efka/proto/service.proto)
|
||
- [efka_subscription.erl](/usr/local/code/cloudkit/efka/apps/efka/src/service/efka_subscription.erl)
|
||
- [efka_service.erl](/usr/local/code/cloudkit/efka/apps/efka/src/service/efka_service.erl)
|
||
|
||
## 1. 连接入口
|
||
|
||
`efka` 启动时会创建 Cowboy WebSocket server，监听配置来自 `efka.config` 的 `websocket_server`：
|
||
|
||
```erlang
|
||
{websocket_server, [
|
||
{port, 18080},
|
||
{acceptors, 10},
|
||
{max_connections, 1024},
|
||
{backlog, 1024}
|
||
]}
|
||
```
|
||
|
||
WebSocket 路由固定为：
|
||
|
||
```http
|
||
GET /ws
|
||
```
|
||
|
||
连接建立后，每个 WebSocket 连接对应一个 `service_channel` 进程。
|
||
|
||
当前协议没有独立的认证字段。service 必须先发送 `register` request 完成注册；未注册前只能处理 `register`，其它 request 会返回 `invalid request`，cast 会被忽略。
|
||
|
||
Cowboy WebSocket ping 帧由 `service_channel` 直接回复 pong：
|
||
|
||
```erlang
|
||
websocket_handle(ping, State) -> {reply, pong, State}
|
||
```
|
||
|
||
## 2. 二进制帧格式
|
||
|
||
service 与 `efka` 之间使用 WebSocket binary frame。每个业务 frame 的第 1 个字节是自定义帧类型，剩余字节是 protobuf payload。
|
||
|
||
```text
|
||
<frame_type:1 byte><protobuf_payload:N bytes>
|
||
```
|
||
|
||
帧类型：
|
||
|
||
| frame_type | 十六进制 | 方向 | 语义 | protobuf 类型 |
|
||
| --- | --- | --- | --- | --- |
|
||
| `REQUEST` | `0x01` | service -> efka | 请求，需要 efka 回复 | `ServiceRequest` |
|
||
| `REPLY` | `0x02` | efka -> service | request 的响应 | `ServiceReply` |
|
||
| `CAST` | `0x03` | 双向 | 单向消息，不要求回复 | `ServiceCast` |
|
||
|
||
`service_channel` 当前只处理 service 发来的 `REQUEST` 和 `CAST`。收到未知格式时只记录日志，不主动发送错误响应。
|
||
|
||
## 3. Protobuf 定义
|
||
|
||
协议定义来自 `apps/efka/proto/service.proto`：
|
||
|
||
```protobuf
|
||
syntax = "proto3";
|
||
|
||
message ServiceRequest {
|
||
uint32 packet_id = 1;
|
||
|
||
message Register {
|
||
string service_id = 1;
|
||
}
|
||
|
||
message Subscribe {
|
||
string topic = 1;
|
||
}
|
||
|
||
oneof request {
|
||
Register register = 10;
|
||
Subscribe subscribe = 11;
|
||
}
|
||
}
|
||
|
||
message ServiceReply {
|
||
uint32 packet_id = 1;
|
||
|
||
message Error {
|
||
int32 code = 1;
|
||
string message = 2;
|
||
}
|
||
|
||
oneof reply {
|
||
bytes result = 10;
|
||
Error error = 11;
|
||
}
|
||
}
|
||
|
||
message ServiceCast {
|
||
message MetricData {
|
||
bytes route_key = 1;
|
||
bytes metric = 2;
|
||
}
|
||
|
||
message TopicEvent {
|
||
string topic = 1;
|
||
bytes content = 2;
|
||
}
|
||
|
||
oneof body {
|
||
TopicEvent topic_event = 10;
|
||
MetricData metric_data = 11;
|
||
}
|
||
}
|
||
```
|
||
|
||
注意：
|
||
|
||
- `ServiceRequest.packet_id` 由 service 生成，用于匹配 request 和 reply。
|
||
- `ServiceReply.packet_id` 必须等于原 request 的 `packet_id`。
|
||
- `ServiceReply.result` 当前成功值一般为 `<<"ok">>`。
|
||
- `ServiceReply.error.code` 当前失败时统一使用 `-1`。
|
||
- `MetricData.route_key` 和 `MetricData.metric` 是 bytes。
|
||
- `TopicEvent.topic` 是 string，`TopicEvent.content` 是 bytes。
|
||
|
||
## 4. request/reply 语义
|
||
|
||
### 4.1 register
|
||
|
||
service 建立 WebSocket 后应先发送注册请求：
|
||
|
||
```protobuf
|
||
ServiceRequest {
|
||
packet_id: 1
|
||
register {
|
||
service_id: "service-a"
|
||
}
|
||
}
|
||
```
|
||
|
||
WebSocket binary frame：
|
||
|
||
```text
|
||
0x01 ++ protobuf(ServiceRequest)
|
||
```
|
||
|
||
`efka` 处理逻辑：
|
||
|
||
1. `service_channel` 解码 `ServiceRequest`。
|
||
2. 调用 `efka_service_sup:start_service(ServiceId)` 启动或复用 service 进程。
|
||
3. 调用 `efka_service:attach_channel(ServicePid, ChannelPid)` 绑定当前 WebSocket channel。
|
||
4. 通过 `service_model:insert/1` 写入 service 记录：
|
||
|
||
```erlang
|
||
#service{
|
||
service_id = ServiceId,
|
||
container_name = <<>>,
|
||
status = ?SERVICE_RUNNING,
|
||
meta_data = #{},
|
||
create_ts = efka_util:timestamp(),
|
||
update_ts = efka_util:timestamp()
|
||
}
|
||
```
|
||
|
||
如果该 `service_id` 是新记录，`status` 会写入 `?SERVICE_RUNNING`。如果该 `service_id` 已存在，当前 `service_model:insert/1` 只更新 `meta_data`、`container_name` 和 `update_ts`，不会覆盖已有 `status`。
|
||
|
||
5. `service_channel` 状态更新为：
|
||
|
||
```erlang
|
||
#state{
|
||
service_id = ServiceId,
|
||
service_pid = ServicePid,
|
||
is_registered = true
|
||
}
|
||
```
|
||
|
||
成功响应：
|
||
|
||
```protobuf
|
||
ServiceReply {
|
||
packet_id: 1
|
||
result: "ok"
|
||
}
|
||
```
|
||
|
||
WebSocket binary frame：
|
||
|
||
```text
|
||
0x02 ++ protobuf(ServiceReply)
|
||
```
|
||
|
||
失败响应：
|
||
|
||
```protobuf
|
||
ServiceReply {
|
||
packet_id: 1
|
||
error {
|
||
code: -1
|
||
message: "attach channel failed"
|
||
}
|
||
}
|
||
```
|
||
|
||
常见失败场景：
|
||
|
||
- 同一个 `service_id` 已有存活 channel，`efka_service:attach_channel/2` 返回 `{error, <<"channel exists">>}`。
|
||
- 当前实现对外统一返回 `"attach channel failed"`，详细原因只写日志。
|
||
|
||
### 4.2 subscribe
|
||
|
||
注册成功后，service 可以订阅 topic：
|
||
|
||
```protobuf
|
||
ServiceRequest {
|
||
packet_id: 2
|
||
subscribe {
|
||
topic: "device/*/event"
|
||
}
|
||
}
|
||
```
|
||
|
||
`efka` 处理逻辑：
|
||
|
||
1. 要求当前 channel 已注册，即 `is_registered = true`。
|
||
2. 调用 `efka_subscription:subscribe(Topic, ChannelPid)`。
|
||
3. 订阅成功后，把 topic 加入 `service_channel` 的 `subscribed_topics` 集合。
|
||
|
||
成功响应：
|
||
|
||
```protobuf
|
||
ServiceReply {
|
||
packet_id: 2
|
||
result: "ok"
|
||
}
|
||
```
|
||
|
||
失败响应：
|
||
|
||
```protobuf
|
||
ServiceReply {
|
||
packet_id: 2
|
||
error {
|
||
code: -1
|
||
message: "invalid topic name"
|
||
}
|
||
}
|
||
```
|
||
|
||
未注册时发送 subscribe 会走通用错误：
|
||
|
||
```protobuf
|
||
ServiceReply {
|
||
packet_id: 2
|
||
error {
|
||
code: -1
|
||
message: "invalid request"
|
||
}
|
||
}
|
||
```
|
||
|
||
## 5. cast 语义
|
||
|
||
### 5.1 service -> efka: metric_data
|
||
|
||
service 通过 `CAST` 上报指标数据：
|
||
|
||
```protobuf
|
||
ServiceCast {
|
||
metric_data {
|
||
route_key: "device/a/temp"
|
||
metric: "{\"value\":23.5}"
|
||
}
|
||
}
|
||
```
|
||
|
||
WebSocket binary frame：
|
||
|
||
```text
|
||
0x03 ++ protobuf(ServiceCast)
|
||
```
|
||
|
||
`efka` 处理逻辑：
|
||
|
||
1. `service_channel` 解码 `ServiceCast`。
|
||
2. 要求当前 channel 已注册。
|
||
3. 调用：
|
||
|
||
```erlang
|
||
efka_service:metric_data(ServicePid, RouteKey, Metric)
|
||
```
|
||
|
||
4. `efka_service` 再调用：
|
||
|
||
```erlang
|
||
efka_client:metric_data(RouteKey, Metric)
|
||
```
|
||
|
||
5. `efka_client` 通过 efka 与 iot 的 TLS 长连接把数据上报给 iot：
|
||
|
||
```erlang
|
||
{<<"message">>, {<<"data">>, #{
|
||
<<"route_key">> => RouteKey,
|
||
<<"metric">> => Metric
|
||
}}}
|
||
```
|
||
|
||
`metric_data` 是单向消息，service 不会收到 reply。未注册或未知 cast body 会被忽略。
|
||
|
||
### 5.2 efka -> service: topic_event
|
||
|
||
当 `iot` 通过 efka 与 iot 的 TLS channel 下发 pub 消息到 `efka` 时，`efka_client` 会调用：
|
||
|
||
```erlang
|
||
efka_subscription:publish(Topic, Qos, Content)
|
||
```
|
||
|
||
`efka_subscription` 根据 topic 匹配已订阅的 `service_channel`。匹配成功后向 channel 进程发送：
|
||
|
||
```erlang
|
||
{topic_broadcast, Topic, Content}
|
||
```
|
||
|
||
`service_channel` 再推送 WebSocket `CAST` 给 service：
|
||
|
||
```protobuf
|
||
ServiceCast {
|
||
topic_event {
|
||
topic: "device/a/event"
|
||
content: "payload bytes"
|
||
}
|
||
}
|
||
```
|
||
|
||
WebSocket binary frame：
|
||
|
||
```text
|
||
0x03 ++ protobuf(ServiceCast)
|
||
```
|
||
|
||
`topic_event` 是单向消息，不要求 service 回复。
|
||
|
||
## 6. topic 匹配规则
|
||
|
||
`efka_subscription` 当前使用 `/` 拆分 topic：
|
||
|
||
```erlang
|
||
binary:split(Topic, <<$/>>, [global])
|
||
```
|
||
|
||
支持两种通配符：
|
||
|
||
| 通配符 | 语义 |
|
||
| --- | --- |
|
||
| `*` | 单级匹配，只匹配一个 segment |
|
||
| `+` | 多级匹配，只能出现在最后一段，且至少匹配一个剩余 segment |
|
||
|
||
示例：
|
||
|
||
| 订阅 topic | 发布 topic | 是否匹配 |
|
||
| --- | --- | --- |
|
||
| `device/a/temp` | `device/a/temp` | 是 |
|
||
| `device/*/temp` | `device/a/temp` | 是 |
|
||
| `device/*/temp` | `device/a/b/temp` | 否 |
|
||
| `device/+` | `device/a` | 是 |
|
||
| `device/+` | `device/a/temp` | 是 |
|
||
| `device/+` | `device` | 否 |
|
||
|
||
同一个 `service_channel` 订阅多个 topic 时，同一次 publish 最多投递一次。
|
||
|
||
## 7. QoS 与遗留消息
|
||
|
||
`efka_subscription:publish(Topic, Qos, Content)` 中的 `Qos` 影响没有订阅者时的行为：
|
||
|
||
- `Qos = 0`：没有匹配订阅者时直接丢弃。
|
||
- `Qos /= 0`：没有匹配订阅者时暂存到 `remaining_messages`。
|
||
|
||
当后续有 service 订阅能匹配这些 topic 时，`efka_subscription` 会把对应遗留消息推送给该 service，并从 `remaining_messages` 移除。
|
||
|
||
当前 `remaining_messages` 存在内存中，不持久化；`efka` 重启后会丢失。
|
||
|
||
## 8. 生命周期与清理
|
||
|
||
### WebSocket channel 关闭
|
||
|
||
`service_channel:terminate/3` 会执行：
|
||
|
||
1. 调用 `efka_subscription:unsubscribe_all(ChannelPid)` 清理该 channel 的所有订阅。
|
||
2. 如果 channel 已完成 register，则调用：
|
||
|
||
```erlang
|
||
service_model:change_status(ServiceId, ?SERVICE_STOPPED)
|
||
```
|
||
|
||
也就是把 service 状态更新为停止。
|
||
|
||
### service 进程退出
|
||
|
||
`service_channel` monitor 绑定的 `efka_service` 进程。如果 service 进程退出，channel 会停止：
|
||
|
||
```erlang
|
||
websocket_info({'DOWN', _Ref, process, ServicePid, Reason}, State) ->
|
||
{stop, State#state{service_pid = undefined}}
|
||
```
|
||
|
||
`efka_service` 同时也 monitor channel。channel 退出时，`efka_service` 会清空自身的 `channel_pid`。
|
||
|
||
## 9. 完整交互流程
|
||
|
||
典型流程：
|
||
|
||
```text
|
||
service efka/service_channel efka_service efka_subscription
|
||
| | | |
|
||
|-- WebSocket connect /ws ------------>| | |
|
||
| | | |
|
||
|-- REQUEST register ----------------->| | |
|
||
| |-- start_service(service_id) ->| |
|
||
| |-- attach_channel ----------->| |
|
||
|<------------- REPLY result="ok" -----| | |
|
||
| | | |
|
||
|-- REQUEST subscribe ---------------->| | |
|
||
| |---------------- subscribe ------------------------------>|
|
||
|<------------- REPLY result="ok" -----| | |
|
||
| | | |
|
||
|-- CAST metric_data ----------------->| | |
|
||
| |-- metric_data -------------->| |
|
||
| | |-- efka_client:metric_data -> iot
|
||
| | | |
|
||
|<------------- CAST topic_event ------|<--------------- topic_broadcast -----------------------|
|
||
| | | |
|
||
|-- WebSocket close ------------------>| | |
|
||
| |-- unsubscribe_all -------------------------------------->|
|
||
| |-- change_status(service_id, stopped)
|
||
```
|
||
|
||
## 10. 当前限制
|
||
|
||
- 没有认证字段，service 只通过 `service_id` 注册。
|
||
- `register` 可重复连接同一个 `service_id`，但同一时间只允许一个活跃 channel 绑定到 `efka_service`。
|
||
- request 只支持 `register` 和 `subscribe`。
|
||
- cast 只支持 `metric_data` 和 `topic_event`。
|
||
- `subscribed_topics` 目前只在 `service_channel` 内记录，关闭时实际清理由 `efka_subscription:unsubscribe_all/1` 完成。
|
||
- `efka_subscription` 当前基于列表扫描匹配，订阅规模很大时需要后续优化为索引结构。
|