%%%------------------------------------------------------------------- %%% @author licheng5 %%% @copyright (C) 2021, %%% @doc %%% %%% @end %%% Created : 11. 1月 2021 上午12:17 %%%------------------------------------------------------------------- -module(ssl_channel). -author("licheng5"). -behaviour(ranch_protocol). -define(MAX_PACKET_ID, 16#FFFFFFFF). -define(INFLIGHT_TIMEOUT, 60000). %% API -export([pub/4, container_call/3, cancel_request_call/2, command/2, activate/2]). -export([start_link/3, stop/2]). %% gen_server callbacks -export([init/3, handle_call/3, handle_cast/2, handle_info/2, code_change/3, terminate/2]). -record(state, { transport, socket, uuid :: undefined | binary(), %% 用户进程id host_pid = undefined, %% 发送消息对应的id packet_id = 1 :: integer(), %% 请求响应的对应关系 inflight = #{} }). -record(inflight_request, { receiver_pid :: pid(), ref :: reference(), timer_ref :: reference() }). %% 向通道中写入消息 -spec pub(Pid :: pid(), Topic :: binary(), Qos :: integer(), Content :: binary()) -> no_return(). 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(). 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}). -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, {request_call, ReceiverPid, Ref, {container_request, Request}}), Ref. -spec cancel_request_call(Pid :: pid(), Ref :: reference()) -> ok. cancel_request_call(Pid, Ref) when is_pid(Pid), is_reference(Ref) -> gen_server:call(Pid, {cancel_request_call, Ref}). %% 关闭方法 -spec stop(Pid :: pid(), Reason :: any()) -> no_return(). stop(undefined, _Reason) -> ok; stop(Pid, Reason) when is_pid(Pid) -> gen_server:stop(Pid, Reason, 5000). %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% %% 逻辑处理方法 %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% start_link(Ref, Transport, Opts) -> {ok, proc_lib:spawn_link(?MODULE, init, [Ref, Transport, Opts])}. init(Ref, Transport, _Opts = []) -> ok = iot_log:set_metadata(), {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_request_call, Ref}, _From, State = #state{inflight = Inflight}) -> case take_inflight_by_ref(Ref, Inflight) of {ok, #inflight_request{timer_ref = TimerRef}, NInflight} -> erlang:cancel_timer(TimerRef), {reply, ok, State#state{inflight = NInflight}}; error -> {reply, ok, State} end; handle_call(_Request, _From, State) -> {reply, ok, State}. %% 发送消息, 基于pub/sub机制 handle_cast({pub, Topic, Qos, Content}, State = #state{transport = Transport, socket = Socket}) -> Packet = term_to_binary({message, {pub, #{topic => Topic, qos => Qos, content => Content}}}), Transport:send(Socket, Packet), {noreply, State}; %% 发送Command消息 handle_cast({command, Command}, State = #state{transport = Transport, socket = Socket}) -> Packet = term_to_binary({message, {auth_control, Command}}), Transport:send(Socket, Packet), {noreply, State}; %% 推送需要响应的请求 handle_cast({request_call, ReceiverPid, Ref, Body}, State = #state{transport = Transport, socket = Socket, packet_id = PacketId, inflight = Inflight}) -> Packet = term_to_binary({request, PacketId, Body}), TimerRef = erlang:start_timer(?INFLIGHT_TIMEOUT, self(), {request_timeout, PacketId}), Transport:send(Socket, Packet), RequestInfo = #inflight_request{receiver_pid = ReceiverPid, ref = Ref, timer_ref = TimerRef}, {noreply, State#state{ packet_id = inc_packet_id(PacketId), inflight = maps:put(PacketId, RequestInfo, Inflight) }}. handle_info({timeout, TimerRef, {request_timeout, PacketId}}, State = #state{inflight = Inflight}) -> case maps:get(PacketId, Inflight, undefined) of #inflight_request{ref = Ref, timer_ref = TimerRef} -> logger:warning("[ws_channel] request timeout, packet_id: ~p, ref: ~p", [PacketId, Ref]), {noreply, State#state{inflight = maps:remove(PacketId, Inflight)}}; _ -> {noreply, State} end; %% 关闭当前通道 handle_info({stop, Reason}, State) -> {stop, Reason, State}; %% 主机进程挂掉的时候，需要关闭掉链接 handle_info({'DOWN', _, process, HostPid, Reason}, State = #state{uuid = UUID, host_pid = HostPid}) -> logger:debug("[ws_channel] uuid: ~p, channel will close because host exited with reason: ~p", [UUID, Reason]), {stop, Reason, State}; handle_info({ssl, Socket, PacketBin}, State = #state{transport = Transport, socket = Socket, host_pid = HostPid, inflight = Inflight}) when is_binary(PacketBin) -> case binary_to_term(PacketBin, [safe]) of {request, PacketId, Body} -> handle_request_frame(PacketId, Body, Transport, Socket, State); {message, Body} -> handle_message_frame(Body, HostPid, State); {response, PacketId, Response} -> handle_response_frame(PacketId, Response, Inflight, State) end; handle_info({ssl_closed, Socket}, State = #state{socket = Socket}) -> logger:notice("[ssl_channel] ssl socket closed"), {stop, normal, State}; handle_info({ssl_error, Socket, Reason}, State = #state{socket = Socket}) -> logger:notice("[ssl_channel] ssl socket error: ~p", [Reason]), {stop, normal, State}; handle_info({ssl_passive, Socket}, State = #state{transport = Transport, socket = Socket}) -> ok = Transport:setopts(Socket, [{active, true}]), {noreply, State}; handle_info(Info, State) -> logger:warning("[ssl_channel] get a unknown message: ~p, state: ~p", [Info, State]), {noreply, State}. terminate(Reason, #state{}) -> logger:warning("[ssl_channel] stop with reason: ~p", [Reason]), ok. code_change(_OldVsn, State, _Extra) -> {ok, State}. %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% %%%% helper methods %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% -spec take_inflight_by_ref(reference(), map()) -> error | {ok, #inflight_request{}, map()}. take_inflight_by_ref(Ref, Inflight) -> maps:fold(fun(PacketId, Request = #inflight_request{ref = PacketRef}, Acc) -> case Acc of error when PacketRef =:= Ref -> {ok, Request, maps:remove(PacketId, Inflight)}; _ -> Acc end end, error, Inflight). -spec inc_packet_id(integer()) -> integer(). inc_packet_id(?MAX_PACKET_ID) -> 1; inc_packet_id(PacketId) when PacketId > 0, PacketId < ?MAX_PACKET_ID -> PacketId + 1. -spec handle_request_frame(non_neg_integer(), tuple(), module(), any(), #state{}) -> {noreply, #state{}} | {stop, term(), #state{}}. handle_request_frame(PacketId, {auth_request, #{uuid := UUID, token := Token, timestamp := Timestamp}}, Transport, Socket, State) -> logger:debug("[ws_channel] auth uuid: ~p", [UUID]), case auth(Token, UUID, Timestamp) of ok -> case iot_api_client:get_host_by_uuid(UUID) of undefined -> logger:warning("[ws_channel] uuid: ~p, token: ~p, host not found", [UUID, Token]), {stop, normal, State}; {ok, _} -> %% 尝试启动主机的服务进程 {ok, HostPid} = iot_host_sup:ensured_host_started(UUID), case iot_host:attach_channel(HostPid, self()) of ok -> erlang:monitor(process, HostPid), send_reply_frame(Transport, Socket, PacketId, {auth_response, {ok, <<"ok">>}}), {noreply, State#state{uuid = UUID, host_pid = HostPid}}; {denied, Reason} when is_binary(Reason) -> erlang:monitor(process, HostPid), send_reply_frame(Transport, Socket, PacketId, {auth_response, {error, 1, 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, PacketId, {auth_response, {error, 2, Reason}}), logger:debug("[ws_channel] uuid: ~p, attach channel get error: ~p, stop channel", [UUID, Reason]), {stop, Reason, State} end end; {error, Reason} -> send_reply_frame(Transport, Socket, PacketId, {auth_response, {error, 2, Reason}}), logger:warning("[ws_channel] uuid: ~p, token: ~p, auth failed, reason: ~p", [UUID, Token, Reason]), {stop, Reason, State} end; handle_request_frame(PacketId, {container_request, ContainerRequest}, _Transport, _Socket, State) -> logger:warning("[ws_channel] unsupported request message type: container_request, packet_id: ~p, request: ~p", [PacketId, ContainerRequest]), {stop, normal, State}; handle_request_frame(PacketId, Body, _Transport, _Socket, State) -> logger:warning("[ws_channel] unsupported request body, packet_id: ~p, body: ~p", [PacketId, Body]), {stop, normal, State}. -spec handle_message_frame(tuple(), undefined | pid(), #state{}) -> {noreply, #state{}}. handle_message_frame({data, #{route_key := RouteKey, metric := Metric}}, HostPid, State) when is_pid(HostPid) -> iot_host:handle(HostPid, {data, RouteKey, Metric}), {noreply, State}; handle_message_frame({task_event, Event}, HostPid, State) when is_pid(HostPid) -> handle_event_stream_frame(Event), {noreply, State}; handle_message_frame(Body, _HostPid, State) -> logger:warning("[ssl_channel] unsupported message body: ~p", [Body]), {noreply, State}. -spec handle_event_stream_frame(map()) -> any(). handle_event_stream_frame(#{task_id := TaskId, type := <<"close">>, stream := Reason}) -> iot_event_stream_observer:stream_close(TaskId, Reason); handle_event_stream_frame(#{task_id := TaskId, type := Type, stream := Stream}) -> logger:debug("[ssl_channel] get task_id: ~p, type: ~ts, stream: ~ts", [TaskId, Type, Stream]), iot_event_stream_observer:stream_data(TaskId, Type, Stream). -spec handle_response_frame(non_neg_integer(), tuple(), map(), #state{}) -> {noreply, #state{}}. handle_response_frame(PacketId, Reply, Inflight, State) when PacketId > 0 -> case maps:take(PacketId, Inflight) of error -> {noreply, State}; {#inflight_request{receiver_pid = ReceiverPid, ref = Ref, timer_ref = TimerRef}, NInflight} -> erlang:cancel_timer(TimerRef), case is_pid(ReceiverPid) andalso is_process_alive(ReceiverPid) of true -> ReceiverPid ! {request_reply, Ref, decode_reply(Reply)}; false -> logger:warning("[ws_channel] get reply message: ~p, packet_id: ~p, but receiver_pid is deaded", [Reply, PacketId]) end, {noreply, State#state{inflight = NInflight}} end; handle_response_frame(PacketId, Reply, _Inflight, State) -> logger:warning("[ws_channel] unexpected response frame, packet_id: ~p, reply: ~p", [PacketId, Reply]), {noreply, State}. -spec send_reply_frame(module(), any(), non_neg_integer(), tuple()) -> any(). send_reply_frame(Transport, Socket, PacketId, Reply) -> Packet = term_to_binary({response, PacketId, Reply}), Transport:send(Socket, Packet). -spec decode_reply({container_response, {ok, term()} | {error, integer(), binary()}} | tuple()) -> {ok, term()} | {error, integer(), binary()} | undefined. decode_reply({container_response, {ok, Result}}) -> {ok, Result}; decode_reply({container_response, {error, Code, Message}}) -> {error, Code, Message}; decode_reply(undefined) -> undefined; decode_reply(_Reply) -> undefined. %% 检测token是否是合法值 -spec auth(Token :: binary(), UUID :: binary(), Timestamp :: integer()) -> ok | {error, Reason :: binary()}. auth(Token, UUID, Timestamp) when is_binary(Token), is_binary(UUID), is_integer(Timestamp) -> %% 1分钟内有效 case iot_util:current_time() - Timestamp =< 60 of true -> case efka_client_store:auth(UUID, Token) of true -> ok; false -> {error, <<"invalid token">>} end; false -> {error, <<"invalid timestamp">>} end.