fix ssl_channel

src/transport/tcp/ssl_channel.erl

@@ -13,7 +13,7 @@
 -define(INFLIGHT_TIMEOUT, 60000).
 
 %% API
--export([pub/4, container_call/3, cancel_command_call/2, command/2, activate/2]).
+-export([pub/4, container_call/3, cancel_command_call/2, activate/2]).
 
 -export([start_link/3, stop/2]).
 %% gen_server callbacks
@@ -41,15 +41,13 @@
 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(), Auth :: boolean()) -> Ref :: reference().
--spec command(Pid :: pid(), Command :: activate | deactivate) -> no_return().
-command(Pid, Command) when is_pid(Pid), (Command =:= activate orelse Command =:= deactivate) ->
-    gen_server:cast(Pid, {command, Command}).
-
--spec activate(Pid :: pid(), Auth :: boolean()) -> no_return().
 activate(Pid, Auth) when is_pid(Pid), is_boolean(Auth) ->
     Command = case Auth of true -> activate; false -> deactivate end,
-    gen_server:cast(Pid, {command, Command}).
+    Ref = make_ref(),
+    ReceiverPid = self(),
+    gen_server:cast(Pid, {command_call, ReceiverPid, Ref, {auth, Command}}),
+    Ref.
 
 -spec container_call(Pid :: pid(), ReceiverPid :: pid(), Request :: map()) -> Ref :: reference().
 container_call(Pid, ReceiverPid, Request) when is_pid(Pid), is_pid(ReceiverPid), is_map(Request) ->
@@ -100,17 +98,6 @@ handle_cast({pub, Topic, Qos, Content}, State = #state{transport = Transport, so
     Transport:send(Socket, Packet),
     {noreply, State};
 
-%% 发送Command消息
-handle_cast({command, Command}, State = #state{transport = Transport, socket = Socket, inflight = Inflight}) ->
-    Ref = make_ref(),
-    Packet = term_to_binary({command, Ref, {auth, Command}}),
-    Transport:send(Socket, Packet),
-
-    TimerRef = erlang:start_timer(?INFLIGHT_TIMEOUT, self(), {command_timeout, Ref}),
-    CommandInfo = #inflight_command{receiver_pid = undefined, timer_ref = TimerRef},
-
-    {noreply, State#state{inflight = maps:put(Ref, CommandInfo, Inflight)}};
-
 %% iot 请求 efka 时使用 command/command_response 语义。
 handle_cast({command_call, ReceiverPid, Ref, Body}, State = #state{transport = Transport, socket = Socket, inflight = Inflight}) ->
     Packet = term_to_binary({command, Ref, Body}),
@@ -123,8 +110,9 @@ handle_cast({command_call, ReceiverPid, Ref, Body}, State = #state{transport = T
 
 handle_info({timeout, TimerRef, {command_timeout, Ref}}, State = #state{inflight = Inflight}) ->
     case maps:get(Ref, Inflight, undefined) of
-        #inflight_command{timer_ref = TimerRef} ->
+        #inflight_command{receiver_pid = ReceiverPid, timer_ref = TimerRef} ->
             logger:warning("[ws_channel] command timeout, ref: ~p", [Ref]),
+            reply_command_timeout(ReceiverPid, Ref),
             {noreply, State#state{inflight = maps:remove(Ref, Inflight)}};
         _ ->
             {noreply, State}
@@ -171,14 +159,14 @@ handle_info(Info, State) ->
     logger:warning("[ssl_channel] get a unknown message: ~p, state: ~p", [Info, State]),
     {noreply, State}.
 
-terminate(Reason, #state{}) ->
+terminate(Reason, #state{inflight = Inflight}) ->
+    maps:foreach(fun(Ref, CommandInfo) -> reply_command_closed(Ref, CommandInfo, Reason) end, Inflight),
     logger:warning("[ssl_channel] stop with reason: ~p", [Reason]),
     ok.
 
 code_change(_OldVsn, State, _Extra) ->
     {ok, State}.
 
-
 %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
 %%%% helper methods
 %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
@@ -252,17 +240,7 @@ handle_command_response_frame(Ref, Reply, Inflight, State) when is_reference(Ref
             {noreply, State};
         {#inflight_command{receiver_pid = ReceiverPid, timer_ref = TimerRef}, NInflight} ->
             erlang:cancel_timer(TimerRef),
-            case ReceiverPid of
+            deliver_command_response(ReceiverPid, Ref, Reply),
-                undefined ->
-                    ok;
-                _ when is_pid(ReceiverPid) ->
-                    case is_process_alive(ReceiverPid) of
-                        true ->
-                            ReceiverPid ! {command_reply, Ref, decode_command_response(Reply)};
-                        false ->
-                            logger:warning("[ws_channel] get command_response: ~p, ref: ~p, but receiver_pid is deaded", [Reply, Ref])
-                    end
-            end,
             {noreply, State#state{inflight = NInflight}}
     end;
 handle_command_response_frame(Ref, Reply, _Inflight, State) ->
@@ -282,9 +260,38 @@ 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}.
 
+-spec deliver_command_response(undefined | pid(), reference(), tuple()) -> ok.
+deliver_command_response(undefined, _Ref, _Reply) ->
+    ok;
+deliver_command_response(ReceiverPid, Ref, Reply) when is_pid(ReceiverPid) ->
+    case is_process_alive(ReceiverPid) of
+        true ->
+            ReceiverPid ! {command_reply, Ref, decode_command_response(Reply)};
+        false ->
+            logger:warning("[ws_channel] get command_response: ~p, ref: ~p, but receiver_pid is deaded", [Reply, Ref])
+    end.
+
+-spec reply_command_timeout(undefined | pid(), reference()) -> ok.
+reply_command_timeout(ReceiverPid, Ref) when is_pid(ReceiverPid) ->
+    ReceiverPid ! {command_reply, Ref, {error, timeout}},
+    ok;
+reply_command_timeout(_ReceiverPid, _Ref) ->
+    ok.
+
+-spec reply_command_closed(reference(), #inflight_command{}, term()) -> ok.
+reply_command_closed(Ref, #inflight_command{receiver_pid = ReceiverPid}, Reason) when is_pid(ReceiverPid) ->
+    ReceiverPid ! {command_reply, Ref, {error, {channel_closed, Reason}}},
+    ok;
+reply_command_closed(_Ref, _CommandInfo, _Reason) ->
+    ok.
+
 %% 检测token是否是合法值
 -spec auth(Token :: binary(), UUID :: binary(), Timestamp :: integer()) ->
     ok | {error, Reason :: binary()}.