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` 取消激活。该操作只修改 iot 本地和持久化授权状态，不通知 efka；efka 连接可继续保持在线，数据是否处理由 iot_host 状态决定。 |
+| auth | boolean | ✅ | `true` 激活, `false` 取消激活 |
 
 #### 响应参数
 | 字段 | 类型 | 说明 |

include/iot.hrl

@@ -13,10 +13,6 @@
 -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,6 +15,7 @@
 
 %% 心跳包检测时间间隔, 15分钟检测一次
 -define(HEARTBEAT_INTERVAL, 900 * 1000).
+-define(AUTH_COMMAND_TIMEOUT, 10000).
 
 %% 状态
 -define(STATE_DENIED, denied).
@@ -81,13 +82,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}).
+    gen_statem:call(Pid, {activate, Auth}, ?AUTH_COMMAND_TIMEOUT + 1000).
 
 -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()}.
+-spec attach_channel(pid(), pid()) -> ok | {error, Reason :: binary()} | {denied, Reason :: binary()}.
 attach_channel(Pid, ChannelPid) when is_pid(Pid), is_pid(ChannelPid) ->
     gen_statem:call(Pid, {attach_channel, ChannelPid}).
 
@@ -175,7 +176,7 @@ init([UUID]) ->
             %% 心跳检测机制
             erlang:start_timer(?HEARTBEAT_INTERVAL, self(), heartbeat_ticker),
 
-            StateName = case AuthorizeStatus =:= ?HOST_AUTHORIZED of
+            StateName = case AuthorizeStatus =:= 1 of
                             true  -> ?STATE_ACTIVATED;
                             false -> ?STATE_DENIED
                         end,
@@ -244,16 +245,33 @@ handle_event({call, From}, {pub, Topic, Qos, Content}, ?STATE_ACTIVATED, State =
             {keep_state, State, [{reply, From, {error, <<"主机离线，发送失败"/utf8>>}}]}
     end;
 
-%% 激活/关闭授权只修改 iot 本地授权状态；efka 连接保持在线，数据是否处理由 host 状态决定。
-handle_event({call, From}, {activate, Auth}, _, State = #state{uuid = UUID}) ->
-    NStateName = case Auth of
+%% 激活主机
+handle_event({call, From}, {activate, true}, _, State = #state{uuid = UUID, channel_pid = ChannelPid}) ->
+    case is_pid(ChannelPid) of
         true ->
-                         ?STATE_ACTIVATED;
+            logger:debug("[iot_host] uuid: ~p, activate: true", [UUID]),
+            start_auth_command(ChannelPid, true, From);
         false ->
-                         ?STATE_DENIED
+            logger:debug("[iot_host] uuid: ~p, activate: true, no channel", [UUID])
     end,
-    logger:debug("[iot_host] uuid: ~p, state_name change to ~p", [UUID, NStateName]),
-    {next_state, NStateName, State, [{reply, From, ok}]};
+    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;
 
 %% 绑定channel
 handle_event({call, From}, {attach_channel, ChannelPid}, StateName, State = #state{uuid = UUID, channel_pid = OldChannelPid}) ->
@@ -268,11 +286,9 @@ 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 denied locally", [UUID, ChannelPid]),
+                    logger:notice("[iot_host] attach_channel host_id uuid: ~p, channel: ~p, host inactivated", [UUID, ChannelPid]),
                     erlang:monitor(process, ChannelPid),
-                    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}]}
+                    {keep_state, State#state{channel_pid = ChannelPid}, [{reply, From, {denied, <<"host inactivated">>}}]}
             end;
         false ->
             logger:notice("[iot_host] attach_channel host_id uuid: ~p, old channel exists: ~p", [UUID, OldChannelPid]),
@@ -286,11 +302,6 @@ 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]),
@@ -300,6 +311,16 @@ 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]),
@@ -352,6 +373,16 @@ 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(<<"">>) ->
     <<"/">>;

src/transport/http/host_handler.erl

@@ -114,6 +114,8 @@ 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]).
+-export([pub/4, container_call/4, cancel_command_call/2, activate/4]).
 
 -export([start_link/3, stop/2]).
 %% gen_server callbacks
@@ -41,6 +41,12 @@
 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}}),
@@ -69,6 +75,7 @@ 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}) ->
@@ -180,6 +187,11 @@ 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]),
@@ -245,6 +257,10 @@ 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}.