From e0b2da5e3f9671f64d25bd546a490f0045e0b794 Mon Sep 17 00:00:00 2001 From: anlicheng <244108715@qq.com> Date: Sat, 9 May 2026 13:04:25 +0800 Subject: [PATCH] =?UTF-8?q?=E7=A7=BB=E9=99=A4=E6=8E=89active=E9=9C=80?= =?UTF-8?q?=E8=A6=81efka=E9=85=8D=E5=90=88=E7=9A=84=E9=80=BB=E8=BE=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- HTTP_API_README.md | 2 +- include/iot.hrl | 4 ++ src/host/iot_host.erl | 77 +++++++++-------------------- src/transport/http/host_handler.erl | 2 - 4 files changed, 28 insertions(+), 57 deletions(-) diff --git a/HTTP_API_README.md b/HTTP_API_README.md index 11488b2..82d1166 100644 --- a/HTTP_API_README.md +++ b/HTTP_API_README.md @@ -1079,7 +1079,7 @@ json_error(ErrCode, ErrMessage) when is_integer(ErrCode), is_binary(ErrMessage) | 参数名 | 类型 | 必填 | 说明 | |--------|------|------|------| | uuid | binary (string) | ✅ | 主机唯一标识符 | -| auth | boolean | ✅ | `true` 激活, `false` 取消激活 | +| auth | boolean | ✅ | `true` 激活, `false` 取消激活。该操作只修改 iot 本地和持久化授权状态，不通知 efka；efka 连接可继续保持在线，数据是否处理由 iot_host 状态决定。 | #### 响应参数 | 字段 | 类型 | 说明 | diff --git a/include/iot.hrl b/include/iot.hrl index d307db7..0ea5bfd 100644 --- a/include/iot.hrl +++ b/include/iot.hrl @@ -13,6 +13,10 @@ -define(HOST_ONLINE, 1). -define(HOST_NOT_JOINED, -1). +%% 主机是否授权处理上报数据 +-define(HOST_DENIED, 0). +-define(HOST_AUTHORIZED, 1). + %% 设备是否在线状态 -define(DEVICE_OFFLINE, 0). -define(DEVICE_ONLINE, 1). diff --git a/src/host/iot_host.erl b/src/host/iot_host.erl index d85ffcf..954fc65 100644 --- a/src/host/iot_host.erl +++ b/src/host/iot_host.erl @@ -15,7 +15,6 @@ %% 心跳包检测时间间隔, 15分钟检测一次 -define(HEARTBEAT_INTERVAL, 900 * 1000). --define(AUTH_COMMAND_TIMEOUT, 10000). %% 状态 -define(STATE_DENIED, denied). @@ -82,13 +81,13 @@ get_status(Pid) when is_pid(Pid) -> %% 激活主机, true 表示激活； false表示关闭激活 -spec activate(Pid :: pid(), Auth :: boolean()) -> ok | {error, term()}. activate(Pid, Auth) when is_pid(Pid), is_boolean(Auth) -> - gen_statem:call(Pid, {activate, Auth}, ?AUTH_COMMAND_TIMEOUT + 1000). + gen_statem:call(Pid, {activate, Auth}). -spec get_metric(Pid :: pid()) -> {ok, MetricInfo :: map()}. get_metric(Pid) when is_pid(Pid) -> gen_statem:call(Pid, get_metric). --spec attach_channel(pid(), pid()) -> ok | {error, Reason :: binary()} | {denied, Reason :: binary()}. +-spec attach_channel(pid(), pid()) -> ok | {error, Reason :: binary()}. attach_channel(Pid, ChannelPid) when is_pid(Pid), is_pid(ChannelPid) -> gen_statem:call(Pid, {attach_channel, ChannelPid}). @@ -176,7 +175,7 @@ init([UUID]) -> %% 心跳检测机制 erlang:start_timer(?HEARTBEAT_INTERVAL, self(), heartbeat_ticker), - StateName = case AuthorizeStatus =:= 1 of + StateName = case AuthorizeStatus =:= ?HOST_AUTHORIZED of true -> ?STATE_ACTIVATED; false -> ?STATE_DENIED end, @@ -245,33 +244,16 @@ handle_event({call, From}, {pub, Topic, Qos, Content}, ?STATE_ACTIVATED, State = {keep_state, State, [{reply, From, {error, <<"主机离线，发送失败"/utf8>>}}]} end; -%% 激活主机 -handle_event({call, From}, {activate, true}, _, State = #state{uuid = UUID, channel_pid = ChannelPid}) -> - case is_pid(ChannelPid) of - true -> - logger:debug("[iot_host] uuid: ~p, activate: true", [UUID]), - start_auth_command(ChannelPid, true, From); - false -> - logger:debug("[iot_host] uuid: ~p, activate: true, no channel", [UUID]) - end, - case is_pid(ChannelPid) of - true -> - {next_state, ?STATE_ACTIVATED, State}; - false -> - {next_state, ?STATE_ACTIVATED, State, [{reply, From, ok}]} - end; - -%% 关闭授权 -handle_event({call, From}, {activate, false}, _, State = #state{uuid = UUID, channel_pid = ChannelPid}) -> - case is_pid(ChannelPid) of - true -> - logger:debug("[iot_host] uuid: ~p, activate: false", [UUID]), - start_auth_command(ChannelPid, false, From), - {keep_state, State}; - false -> - logger:debug("[iot_host] uuid: ~p, activate: false, no channel", [UUID]), - {next_state, ?STATE_DENIED, State#state{channel_pid = undefined, has_session = false}, [{reply, From, ok}]} - end; +%% 激活/关闭授权只修改 iot 本地授权状态；efka 连接保持在线，数据是否处理由 host 状态决定。 +handle_event({call, From}, {activate, Auth}, _, State = #state{uuid = UUID}) -> + NStateName = case Auth of + true -> + ?STATE_ACTIVATED; + false -> + ?STATE_DENIED + end, + logger:debug("[iot_host] uuid: ~p, state_name change to ~p", [UUID, NStateName]), + {next_state, NStateName, State, [{reply, From, ok}]}; %% 绑定channel handle_event({call, From}, {attach_channel, ChannelPid}, StateName, State = #state{uuid = UUID, channel_pid = OldChannelPid}) -> @@ -286,9 +268,11 @@ handle_event({call, From}, {attach_channel, ChannelPid}, StateName, State = #sta {keep_state, State#state{channel_pid = ChannelPid, has_session = true}, [{reply, From, ok}]}; %% 主机未激活 ?STATE_DENIED -> - logger:notice("[iot_host] attach_channel host_id uuid: ~p, channel: ~p, host inactivated", [UUID, ChannelPid]), + logger:notice("[iot_host] attach_channel host_id uuid: ~p, channel: ~p, host denied locally", [UUID, ChannelPid]), erlang:monitor(process, ChannelPid), - {keep_state, State#state{channel_pid = ChannelPid}, [{reply, From, {denied, <<"host inactivated">>}}]} + ChangeResult = iot_api_client:change_host_status(UUID, ?HOST_ONLINE), + logger:debug("[iot_host] host_id(attach_channel) uuid: ~p, denied but online, change status result: ~p", [UUID, ChangeResult]), + {keep_state, State#state{channel_pid = ChannelPid, has_session = true}, [{reply, From, ok}]} end; false -> logger:notice("[iot_host] attach_channel host_id uuid: ~p, old channel exists: ~p", [UUID, OldChannelPid]), @@ -302,6 +286,11 @@ handle_event(cast, {handle, {data, RouteKey, MetricBin}}, ?STATE_ACTIVATED, endpoint_subscription:publish(get_route_key(RouteKey), MetricBin), {keep_state, State}; +handle_event(cast, {handle, {data, RouteKey, MetricBin}}, StateName, + State = #state{uuid = UUID, has_session = true}) -> + logger:notice("[iot_host] host_uuid: ~p, state_name: ~p, metric_data route_key: ~p, metric: ~p, discard", [UUID, StateName, RouteKey, MetricBin]), + {keep_state, State}; + %% ping的数据是通过aes加密后的，因此需要在有会话的情况下才行 handle_event(cast, {handle, {ping, Metrics}}, ?STATE_ACTIVATED, State = #state{uuid = UUID, has_session = true}) -> logger:debug("[iot_host] ping host_id uuid: ~p, get ping: ~p", [UUID, Metrics]), @@ -311,16 +300,6 @@ handle_event(cast, {handle, {ping, Metrics}}, ?STATE_ACTIVATED, State = #state{u handle_event(cast, heartbeat, _, State = #state{heartbeat_counter = HeartbeatCounter}) -> {keep_state, State#state{heartbeat_counter = HeartbeatCounter + 1}}; -handle_event(info, {auth_command_result, From, true, _ChannelPid, ok}, _, State) -> - {keep_state, State, [{reply, From, ok}]}; -handle_event(info, {auth_command_result, From, true, _ChannelPid, {error, Reason}}, _, State) -> - {keep_state, State, [{reply, From, {error, Reason}}]}; -handle_event(info, {auth_command_result, From, false, ChannelPid, ok}, _, State) -> - ssl_channel:stop(ChannelPid, closed), - {next_state, ?STATE_DENIED, State#state{channel_pid = undefined, has_session = false}, [{reply, From, ok}]}; -handle_event(info, {auth_command_result, From, false, _ChannelPid, {error, Reason}}, _, State) -> - {keep_state, State, [{reply, From, {error, Reason}}]}; - %% 没有收到心跳包，主机下线, 设备状态不变 handle_event(info, {timeout, _, heartbeat_ticker}, _, State = #state{uuid = UUID, heartbeat_counter = 0, channel_pid = ChannelPid}) -> logger:warning("[iot_host] uuid: ~p, heartbeat lost, devices will unknown", [UUID]), @@ -373,16 +352,6 @@ code_change(_OldVsn, StateName, State = #state{}, _Extra) -> container_call(Pid, Request) when is_pid(Pid) -> gen_statem:call(Pid, {container_call, self(), Request}). --spec start_auth_command(pid(), boolean(), term()) -> pid(). -start_auth_command(ChannelPid, Auth, From) when is_pid(ChannelPid), is_boolean(Auth) -> - HostPid = self(), - spawn(fun() -> - Ref = make_ref(), - ok = ssl_channel:activate(ChannelPid, self(), Ref, Auth), - Result = iot_host:await_reply(HostPid, Ref, ?AUTH_COMMAND_TIMEOUT), - HostPid ! {auth_command_result, From, Auth, ChannelPid, Result} - end). - -spec get_route_key(binary()) -> binary(). get_route_key(<<"">>) -> <<"/">>; @@ -423,4 +392,4 @@ maybe_mark_host_offline(UUID) -> Other -> logger:warning("[iot_host] host: ~p, load status failed while marking offline: ~p", [UUID, Other]), ok - end. + end. \ No newline at end of file diff --git a/src/transport/http/host_handler.erl b/src/transport/http/host_handler.erl index 9c5e5ac..4329004 100644 --- a/src/transport/http/host_handler.erl +++ b/src/transport/http/host_handler.erl @@ -114,8 +114,6 @@ reason_to_binary(timeout) -> <<"timeout">>; reason_to_binary(invalid_response) -> <<"invalid response">>; -reason_to_binary({denied, Reason}) -> - reason_to_binary(Reason); reason_to_binary({failed, Reason}) -> reason_to_binary(Reason); reason_to_binary({channel_closed, Reason}) ->