diff --git a/src/transport/tcp/ssl_channel.erl b/src/transport/tcp/ssl_channel.erl index e2e0eba..c0cc313 100644 --- a/src/transport/tcp/ssl_channel.erl +++ b/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,20 +41,18 @@ 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 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(). +-spec activate(Pid :: pid(), Auth :: boolean()) -> Ref :: reference(). 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) -> Ref = make_ref(), - gen_server:cast(Pid, {command_call, ReceiverPid, Ref, {container, Request}}), + gen_server:cast(Pid, {command_call, ReceiverPid, Ref, {container, Request}}), Ref. -spec cancel_command_call(Pid :: pid(), Ref :: reference()) -> ok. @@ -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 - 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, + deliver_command_response(ReceiverPid, Ref, Reply), {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()}.