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

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 状态决定。 |
 
 #### 响应参数
 | 字段 | 类型 | 说明 |

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).

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.

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}) ->

src/transport/tcp/ssl_channel.erl

@@ -13,7 +13,7 @@
 -define(INFLIGHT_TIMEOUT, 60000).
 
 %% API
--export([pub/4, container_call/4, cancel_command_call/2, activate/4]).
+-export([pub/4, container_call/4, cancel_command_call/2]).
 
 -export([start_link/3, stop/2]).
 %% gen_server callbacks
@@ -41,12 +41,6 @@
 pub(Pid, Topic, Qos, Content) when is_pid(Pid), is_binary(Topic), is_integer(Qos), is_binary(Content) ->
     gen_server:cast(Pid, {pub, Topic, Qos, Content}).
 
--spec activate(Pid :: pid(), ReceiverPid :: pid(), Ref :: reference(), Auth :: boolean()) -> ok.
-activate(Pid, ReceiverPid, Ref, Auth) when is_pid(Pid), is_pid(ReceiverPid), is_reference(Ref), is_boolean(Auth) ->
-    Command = case Auth of true -> activate; false -> deactivate end,
-    gen_server:cast(Pid, {command_call, ReceiverPid, Ref, {auth, Command}}),
-    ok.
-
 -spec container_call(Pid :: pid(), ReceiverPid :: pid(), Ref :: reference(), Request :: map()) -> ok.
 container_call(Pid, ReceiverPid, Ref, Request) when is_pid(Pid), is_pid(ReceiverPid), is_reference(Ref), is_map(Request) ->
     gen_server:cast(Pid, {command_call, ReceiverPid, Ref,  {container, Request}}),
@@ -75,7 +69,6 @@ init(Ref, Transport, _Opts = []) ->
     {ok, Socket} = ranch:handshake(Ref),
     logger:debug("[ssl_channel] get a new connection: ~p", [Socket]),
     Transport:setopts(Socket, [binary, {active, true}, {packet, 4}]),
-    % erlang:start_timer(?PING_TICKER, self(), ping_ticker),
     gen_server:enter_loop(?MODULE, [], #state{transport = Transport, socket = Socket}).
 
 handle_call({cancel_command_call, Ref}, _From, State = #state{inflight = Inflight}) ->
@@ -187,11 +180,6 @@ handle_request_frame(Ref, {auth_request, #{uuid := UUID, token := Token, timesta
                             erlang:monitor(process, HostPid),
                             send_reply_frame(Transport, Socket, Ref, {auth_response, ok}),
                             {noreply, State#state{uuid = UUID, host_pid = HostPid}};
-                        {denied, Reason} when is_binary(Reason) ->
-                            erlang:monitor(process, HostPid),
-                            send_reply_frame(Transport, Socket, Ref, {auth_response, {error, {denied, Reason}}}),
-                            logger:debug("[ws_channel] uuid: ~p, attach channel get error: ~p, stop channel", [UUID, Reason]),
-                            {noreply, State#state{uuid = UUID, host_pid = HostPid}};
                         {error, Reason} when is_binary(Reason) ->
                             send_reply_frame(Transport, Socket, Ref, {auth_response, {error, {failed, Reason}}}),
                             logger:debug("[ws_channel] uuid: ~p, attach channel get error: ~p, stop channel", [UUID, Reason]),
@@ -257,10 +245,6 @@ decode_command_response({container, {ok, Result}}) ->
     {ok, Result};
 decode_command_response({container, {error, Reason}}) ->
     {error, Reason};
-decode_command_response({auth, ok}) ->
-    ok;
-decode_command_response({auth, {error, Reason}}) ->
-    {error, Reason};
 decode_command_response(_Reply) ->
     {error, invalid_response}.