anlicheng/ekfa
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix docs

Browse Source
This commit is contained in:
anlicheng 2026-05-11 14:44:27 +08:00
parent 47d9973854
commit 512059422f
1 changed files with 439 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

439
docs/service_websocket_protocol.md Normal file
View File

@ -0,0 +1,439 @@
# EFKA 与 Service WebSocket 交互协议
本文档描述部署在 `efka` 主机上的 service 与 `efka` 之间的 WebSocket 通讯逻辑。当前接入实现位于：
- [service_channel.erl](/usr/local/code/cloudkit/efka/src/transport/service_channel.erl)
- [service.proto](/usr/local/code/cloudkit/efka/proto/service.proto)
- [efka_subscription.erl](/usr/local/code/cloudkit/efka/src/efka_subscription.erl)
- [efka_service.erl](/usr/local/code/cloudkit/efka/src/micro_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 定义
协议定义来自 `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` 当前基于列表扫描匹配，订阅规模很大时需要后续优化为索引结构。