diff --git a/docs/efka_iot_protocol.md b/docs/efka_iot_protocol.md index c83e0a3..8e7cba1 100644 --- a/docs/efka_iot_protocol.md +++ b/docs/efka_iot_protocol.md @@ -219,6 +219,22 @@ GET /event_stream?uuid=&task_id= `iot` 会为每个 `{UUID, TaskId}` 维护一个独立的任务进程,用于缓存最近的部署日志、支持多个 SSE listener,并在收到 close 事件后结束事件流。 +### efka -> iot: ping + +```erlang +{<<"message">>, <<"ping">>} +``` + +用于 TLS 长连接的应用层保活。`efka` 在鉴权成功后周期发送,当前发送间隔为 30 秒。 + +`iot` 收到后回复: + +```erlang +{<<"message">>, <<"pong">>} +``` + +`ping/pong` 只表示 TLS 连接仍可读写,不参与 host online/offline 判定。host 上下线仍由 UDP 心跳和 `iot_host` 本地连接状态共同维护。 + ### iot -> efka: pub ```erlang @@ -235,6 +251,7 @@ GET /event_stream?uuid=&task_id= - `efka` 鉴权超时时间:5 秒。 - `iot` command inflight 超时时间:60 秒。 +- `iot` SSL channel 空闲超时时间:120 秒。120 秒内没有收到任何 TLS 包,包括 `ping`、业务 `message`、`request`、`command_response`,`iot` 会主动关闭该连接。 - `iot` 管理多个 `efka` 时,每个连接有独立 `ssl_channel` 和独立 inflight 表。 - command 超时后,`iot` 删除 inflight 记录;之后如果迟到的 `command_response` 到达,会被视为未预期响应。 diff --git a/src/transport/efka_client.erl b/src/transport/efka_client.erl index 0e9f1c5..7bee5f0 100644 --- a/src/transport/efka_client.erl +++ b/src/transport/efka_client.erl @@ -28,11 +28,13 @@ -define(STATE_AUTH, auth). %% 激活状态下 -define(STATE_ACTIVATED, activated). +-define(SSL_PING_INTERVAL, 30000). -record(state, { socket :: undefined | ssl:sslsocket(), %% 保存当前auth请求的ref,用来建立auth请求和响应的对应关系 auth_ref = undefined :: undefined | binary(), + ping_timer_ref = undefined :: undefined | reference(), dropped_message_count = 0 :: non_neg_integer() }). @@ -143,6 +145,20 @@ handle_event(state_timeout, auth_timeout, ?STATE_AUTH, State = #state{socket = S schedule_reconnect(), {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined}}; +handle_event(info, {timeout, TimerRef, ssl_ping}, ?STATE_ACTIVATED, State = #state{socket = Socket, ping_timer_ref = TimerRef}) -> + Packet = term_to_binary({<<"message">>, <<"ping">>}), + case ssl:send(Socket, Packet) of + ok -> + {keep_state, schedule_ssl_ping(State)}; + {error, Reason} -> + logger:warning("[efka_client] send ssl ping failed, reason: ~p", [Reason]), + disconnect(Socket), + schedule_reconnect(), + {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined, ping_timer_ref = undefined}} + end; +handle_event(info, {timeout, _TimerRef, ssl_ping}, _StateName, State) -> + {keep_state, State}; + %% 将缓存中的数据推送到服务器端 handle_event(info, flush_cache, ?STATE_ACTIVATED, State = #state{socket = Socket}) -> case efka_client_cache:fetch_next() of @@ -165,19 +181,22 @@ handle_event(info, {ssl, Socket, PacketBin}, _, State = #state{socket = Socket}) error:Error -> logger:warning("[efka_client] binary_to_term get error: ~p, packet_size: ~p", [Error, byte_size(PacketBin)]), disconnect(Socket), + cancel_ssl_ping(State), schedule_reconnect(), - {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined}} + {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined, ping_timer_ref = undefined}} end; handle_event(info, {ssl_error, Socket, Reason}, _, State = #state{}) -> logger:debug("[efka_client] ssl error: ~p", [Reason]), disconnect(Socket), + cancel_ssl_ping(State), schedule_reconnect(), - {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined}}; + {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined, ping_timer_ref = undefined}}; handle_event(info, {ssl_closed, Socket}, _, State = #state{}) -> logger:debug("[efka_client] ssl closed"), disconnect(Socket), + cancel_ssl_ping(State), schedule_reconnect(), - {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined}}; + {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined, ping_timer_ref = undefined}}; %%% 处理内部消息,ssl收到的消息会先 binary_to_term,再由这里按协议结构模式匹配 @@ -193,7 +212,8 @@ handle_event(internal, {<<"command">>, Ref, {<<"container">>, Request}}, _StateN %% 处理response handle_event(internal, {<<"response">>, AuthRef, {<<"auth_response">>, <<"ok">>}}, ?STATE_AUTH, State = #state{auth_ref = AuthRef}) -> logger:debug("[efka_client] auth success"), - {next_state, ?STATE_ACTIVATED, State#state{auth_ref = undefined}, [{next_event, info, flush_cache}]}; + State1 = schedule_ssl_ping(State#state{auth_ref = undefined}), + {next_state, ?STATE_ACTIVATED, State1, [{next_event, info, flush_cache}]}; handle_event(internal, {<<"response">>, AuthRef, {<<"auth_response">>, {<<"error">>, Reason}}}, ?STATE_AUTH, State = #state{socket = Socket, auth_ref = AuthRef}) -> logger:debug("[efka_client] auth failed, reason: ~p", [Reason]), disconnect(Socket), @@ -207,6 +227,8 @@ handle_event(internal, {<<"command_response">>, _Ref, Reply}, StateName, State) {keep_state, State}; %% 处理Pub/Sub机制 +handle_event(internal, {<<"message">>, <<"pong">>}, ?STATE_ACTIVATED, State) -> + {keep_state, State}; handle_event(internal, {<<"message">>, {<<"pub">>, #{<<"topic">> := Topic, <<"qos">> := Qos, <<"content">> := Content}}}, ?STATE_ACTIVATED, State) -> logger:debug("[efka_client] get pub topic: ~p, qos: ~p, content: ~p", [Topic, Qos, Content]), efka_subscription:publish(Topic, Qos, Content), @@ -254,7 +276,8 @@ handle_container_command(Ref, Request, Socket) -> ok. -spec terminate(term(), atom(), #state{}) -> ok. -terminate(Reason, _StateName, _State = #state{socket = Socket}) -> +terminate(Reason, _StateName, State = #state{socket = Socket}) -> + cancel_ssl_ping(State), disconnect(Socket), efka_client_cache:close(), logger:notice("[efka_client] terminate with reason: ~p", [Reason]), @@ -305,6 +328,22 @@ disconnect(Socket) -> schedule_reconnect() -> erlang:start_timer(5000, self(), create_transport). +-spec schedule_ssl_ping(#state{}) -> #state{}. +schedule_ssl_ping(State = #state{ping_timer_ref = TimerRef}) -> + cancel_timer(TimerRef), + State#state{ping_timer_ref = erlang:start_timer(?SSL_PING_INTERVAL, self(), ssl_ping)}. + +-spec cancel_ssl_ping(#state{}) -> ok. +cancel_ssl_ping(#state{ping_timer_ref = TimerRef}) -> + cancel_timer(TimerRef). + +-spec cancel_timer(undefined | reference()) -> ok. +cancel_timer(undefined) -> + ok; +cancel_timer(TimerRef) -> + _ = erlang:cancel_timer(TimerRef), + ok. + -spec request_ref() -> binary(). request_ref() -> crypto:strong_rand_bytes(16).