移除掉active需要efka配合的逻辑

This commit is contained in:
anlicheng 2026-05-09 13:04:25 +08:00
parent 8ee8d54c26
commit e0b2da5e3f
4 changed files with 28 additions and 57 deletions

View File

@ -1079,7 +1079,7 @@ json_error(ErrCode, ErrMessage) when is_integer(ErrCode), is_binary(ErrMessage)
| 参数名 | 类型 | 必填 | 说明 | | 参数名 | 类型 | 必填 | 说明 |
|--------|------|------|------| |--------|------|------|------|
| uuid | binary (string) | ✅ | 主机唯一标识符 | | uuid | binary (string) | ✅ | 主机唯一标识符 |
| auth | boolean | ✅ | `true` 激活, `false` 取消激活 | | auth | boolean | ✅ | `true` 激活, `false` 取消激活。该操作只修改 iot 本地和持久化授权状态,不通知 efkaefka 连接可继续保持在线,数据是否处理由 iot_host 状态决定。 |
#### 响应参数 #### 响应参数
| 字段 | 类型 | 说明 | | 字段 | 类型 | 说明 |

View File

@ -13,6 +13,10 @@
-define(HOST_ONLINE, 1). -define(HOST_ONLINE, 1).
-define(HOST_NOT_JOINED, -1). -define(HOST_NOT_JOINED, -1).
%%
-define(HOST_DENIED, 0).
-define(HOST_AUTHORIZED, 1).
%% 线 %% 线
-define(DEVICE_OFFLINE, 0). -define(DEVICE_OFFLINE, 0).
-define(DEVICE_ONLINE, 1). -define(DEVICE_ONLINE, 1).

View File

@ -15,7 +15,6 @@
%% , 15 %% , 15
-define(HEARTBEAT_INTERVAL, 900 * 1000). -define(HEARTBEAT_INTERVAL, 900 * 1000).
-define(AUTH_COMMAND_TIMEOUT, 10000).
%% %%
-define(STATE_DENIED, denied). -define(STATE_DENIED, denied).
@ -82,13 +81,13 @@ get_status(Pid) when is_pid(Pid) ->
%% , true false表示关闭激活 %% , true false表示关闭激活
-spec activate(Pid :: pid(), Auth :: boolean()) -> ok | {error, term()}. -spec activate(Pid :: pid(), Auth :: boolean()) -> ok | {error, term()}.
activate(Pid, Auth) when is_pid(Pid), is_boolean(Auth) -> 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()}. -spec get_metric(Pid :: pid()) -> {ok, MetricInfo :: map()}.
get_metric(Pid) when is_pid(Pid) -> get_metric(Pid) when is_pid(Pid) ->
gen_statem:call(Pid, get_metric). 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) -> attach_channel(Pid, ChannelPid) when is_pid(Pid), is_pid(ChannelPid) ->
gen_statem:call(Pid, {attach_channel, ChannelPid}). gen_statem:call(Pid, {attach_channel, ChannelPid}).
@ -176,7 +175,7 @@ init([UUID]) ->
%% %%
erlang:start_timer(?HEARTBEAT_INTERVAL, self(), heartbeat_ticker), erlang:start_timer(?HEARTBEAT_INTERVAL, self(), heartbeat_ticker),
StateName = case AuthorizeStatus =:= 1 of StateName = case AuthorizeStatus =:= ?HOST_AUTHORIZED of
true -> ?STATE_ACTIVATED; true -> ?STATE_ACTIVATED;
false -> ?STATE_DENIED false -> ?STATE_DENIED
end, end,
@ -245,33 +244,16 @@ handle_event({call, From}, {pub, Topic, Qos, Content}, ?STATE_ACTIVATED, State =
{keep_state, State, [{reply, From, {error, <<"主机离线,发送失败"/utf8>>}}]} {keep_state, State, [{reply, From, {error, <<"主机离线,发送失败"/utf8>>}}]}
end; end;
%% %% / iot efka 线 host
handle_event({call, From}, {activate, true}, _, State = #state{uuid = UUID, channel_pid = ChannelPid}) -> handle_event({call, From}, {activate, Auth}, _, State = #state{uuid = UUID}) ->
case is_pid(ChannelPid) of NStateName = case Auth of
true -> true ->
logger:debug("[iot_host] uuid: ~p, activate: true", [UUID]), ?STATE_ACTIVATED;
start_auth_command(ChannelPid, true, From); false ->
false -> ?STATE_DENIED
logger:debug("[iot_host] uuid: ~p, activate: true, no channel", [UUID]) end,
end, logger:debug("[iot_host] uuid: ~p, state_name change to ~p", [UUID, NStateName]),
case is_pid(ChannelPid) of {next_state, NStateName, State, [{reply, From, ok}]};
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;
%% channel %% channel
handle_event({call, From}, {attach_channel, ChannelPid}, StateName, State = #state{uuid = UUID, channel_pid = OldChannelPid}) -> 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}]}; {keep_state, State#state{channel_pid = ChannelPid, has_session = true}, [{reply, From, ok}]};
%% %%
?STATE_DENIED -> ?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), 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; end;
false -> false ->
logger:notice("[iot_host] attach_channel host_id uuid: ~p, old channel exists: ~p", [UUID, OldChannelPid]), 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), endpoint_subscription:publish(get_route_key(RouteKey), MetricBin),
{keep_state, State}; {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加密后的 %% ping的数据是通过aes加密后的
handle_event(cast, {handle, {ping, Metrics}}, ?STATE_ACTIVATED, State = #state{uuid = UUID, has_session = true}) -> 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]), 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}) -> handle_event(cast, heartbeat, _, State = #state{heartbeat_counter = HeartbeatCounter}) ->
{keep_state, State#state{heartbeat_counter = HeartbeatCounter + 1}}; {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}) -> 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]), 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) -> container_call(Pid, Request) when is_pid(Pid) ->
gen_statem:call(Pid, {container_call, self(), Request}). 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(). -spec get_route_key(binary()) -> binary().
get_route_key(<<"">>) -> get_route_key(<<"">>) ->
<<"/">>; <<"/">>;
@ -423,4 +392,4 @@ maybe_mark_host_offline(UUID) ->
Other -> Other ->
logger:warning("[iot_host] host: ~p, load status failed while marking offline: ~p", [UUID, Other]), logger:warning("[iot_host] host: ~p, load status failed while marking offline: ~p", [UUID, Other]),
ok ok
end. end.

View File

@ -114,8 +114,6 @@ reason_to_binary(timeout) ->
<<"timeout">>; <<"timeout">>;
reason_to_binary(invalid_response) -> reason_to_binary(invalid_response) ->
<<"invalid response">>; <<"invalid response">>;
reason_to_binary({denied, Reason}) ->
reason_to_binary(Reason);
reason_to_binary({failed, Reason}) -> reason_to_binary({failed, Reason}) ->
reason_to_binary(Reason); reason_to_binary(Reason);
reason_to_binary({channel_closed, Reason}) -> reason_to_binary({channel_closed, Reason}) ->