From 512059422ff29e1f256c79327f7cc29b35c34d99 Mon Sep 17 00:00:00 2001 From: anlicheng <244108715@qq.com> Date: Mon, 11 May 2026 14:44:27 +0800 Subject: [PATCH] fix docs --- docs/service_websocket_protocol.md | 439 +++++++++++++++++++++++++++++ 1 file changed, 439 insertions(+) create mode 100644 docs/service_websocket_protocol.md diff --git a/docs/service_websocket_protocol.md b/docs/service_websocket_protocol.md new file mode 100644 index 0000000..888fba6 --- /dev/null +++ b/docs/service_websocket_protocol.md @@ -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 | 十六进制 | 方向 | 语义 | 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` 当前基于列表扫描匹配，订阅规模很大时需要后续优化为索引结构。