%%%------------------------------------------------------------------- %%% @author licheng5 %%% @copyright (C) 2021, %%% @doc %%% %%% @end %%% Created : 11. 1月 2021 上午12:17 %%%------------------------------------------------------------------- -module(ssl_channel). -author("licheng5"). -behaviour(ranch_protocol). -define(INFLIGHT_TIMEOUT, 60000). -define(SSL_IDLE_TIMEOUT, 120000). %% API -export([pub/4, container_call/4, cancel_command_call/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(), is_authed = false :: boolean(), %% 用户进程id host_pid = undefined, %% iot 发起的 command 与 command_response 的对应关系 inflight = #{}, idle_timer_ref :: undefined | reference() }). -record(inflight_command, { receiver_pid :: undefined | pid(), timer_ref :: reference() }). -type request_ref() :: binary(). %% 向通道中写入消息 -spec pub(Pid :: pid(), Topic :: binary(), Qos :: integer(), Content :: binary()) -> ok. 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 container_call(Pid :: pid(), ReceiverPid :: pid(), Ref :: request_ref(), Request :: map()) -> ok. container_call(Pid, ReceiverPid, Ref, Request) when is_pid(Pid), is_pid(ReceiverPid), is_binary(Ref), is_map(Request) -> gen_server:cast(Pid, {command_call, ReceiverPid, Ref, {container, Request}}), ok. -spec cancel_command_call(Pid :: pid(), Ref :: request_ref()) -> ok. cancel_command_call(Pid, Ref) when is_pid(Pid), is_binary(Ref) -> gen_server:call(Pid, {cancel_command_call, Ref}). %% 关闭方法 -spec stop(Pid :: undefined | pid(), Reason :: any()) -> ok. 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}]), IdleTimerRef = start_idle_timer(), gen_server:enter_loop(?MODULE, [], #state{transport = Transport, socket = Socket, idle_timer_ref = IdleTimerRef}). handle_call({cancel_command_call, Ref}, _From, State = #state{inflight = Inflight}) -> case maps:take(Ref, Inflight) of {#inflight_command{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}}}), case Transport:send(Socket, Packet) of ok -> {noreply, State}; {error, Reason} -> logger:warning("[ssl_channel] send pub failed, reason: ~p", [Reason]), {stop, {send_failed, Reason}, State} end; %% 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, encode_command_body(Body)}), case Transport:send(Socket, Packet) of ok -> TimerRef = erlang:start_timer(?INFLIGHT_TIMEOUT, self(), {command_timeout, Ref}), CommandInfo = #inflight_command{receiver_pid = ReceiverPid, timer_ref = TimerRef}, {noreply, State#state{inflight = maps:put(Ref, CommandInfo, Inflight)}}; {error, Reason} -> logger:warning("[ssl_channel] send command failed, ref: ~p, reason: ~p", [Ref, Reason]), deliver_command_error(ReceiverPid, Ref, {send_failed, Reason}), {stop, {send_failed, Reason}, State} end. handle_info({timeout, TimerRef, {command_timeout, Ref}}, State = #state{inflight = Inflight}) -> case maps:get(Ref, Inflight, undefined) of #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} end; handle_info({timeout, TimerRef, ssl_idle_timeout}, State = #state{idle_timer_ref = TimerRef}) -> logger:notice("[ssl_channel] ssl channel idle timeout"), {stop, ssl_idle_timeout, State#state{idle_timer_ref = undefined}}; handle_info({timeout, _TimerRef, ssl_idle_timeout}, State) -> {noreply, State}; %% 关闭当前通道 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{socket = Socket}) when is_binary(PacketBin) -> State1 = reset_idle_timer(State), try binary_to_term(PacketBin, [safe]) of {<<"request">>, Ref, Body} -> handle_request_frame(Ref, Body, State1); {<<"message">>, Body} -> handle_message_frame(Body, State1); {<<"command_response">>, Ref, Response} -> handle_command_response_frame(Ref, Response, State1); Other -> logger:warning("[ssl_channel] unsupported packet: ~p", [Other]), {stop, bad_packet, State1} catch error:Error -> logger:warning("[ssl_channel] binary_to_term get error: ~p, packet_size: ~p", [Error, byte_size(PacketBin)]), {stop, bad_packet, State1} 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{inflight = Inflight, transport = Transport, socket = Socket, idle_timer_ref = IdleTimerRef}) -> cancel_timer(IdleTimerRef), maps:foreach(fun(Ref, CommandInfo) -> reply_command_closed(Ref, CommandInfo, Reason) end, Inflight), Transport:close(Socket), logger:warning("[ssl_channel] stop with reason: ~p", [Reason]), ok. code_change(_OldVsn, State, _Extra) -> {ok, State}. %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% %%%% helper methods %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% -spec handle_request_frame(request_ref(), tuple(), #state{}) -> {noreply, #state{}} | {stop, term(), #state{}}. handle_request_frame(Ref, {<<"auth_request">>, #{<<"uuid">> := UUID}}, State = #state{is_authed = true}) -> logger:warning("[ws_channel] repeated auth request, ref: ~p, uuid: ~p, close channel", [Ref, UUID]), {stop, repeated_auth, State}; handle_request_frame(Ref, {<<"auth_request">>, #{<<"uuid">> := UUID, <<"token">> := Token, <<"timestamp">> := Timestamp}}, State = #state{transport = Transport, socket = Socket}) -> maybe ok ?= auth(Token, UUID, Timestamp), {ok, HostPid} ?= iot_host:lookup_pid(UUID), ok ?= iot_host:attach_channel(HostPid, self()), erlang:monitor(process, HostPid), ok = send_reply_frame(Transport, Socket, Ref, {<<"auth_response">>, <<"ok">>}), logger:debug("[ws_channel] auth uuid: ~p", [UUID]), {noreply, State#state{uuid = UUID, is_authed = true, host_pid = HostPid}} else {error, Reason} -> logger:warning("[ws_channel] uuid: ~p, auth failed with reason: ~p", [UUID, Reason]), case send_reply_frame(Transport, Socket, Ref, {<<"auth_response">>, {<<"error">>, {<<"failed">>, Reason}}}) of ok -> {stop, Reason, State}; {error, SendReason} -> logger:warning("[ws_channel] uuid: ~p, send auth failed response failed: ~p", [UUID, SendReason]), {stop, {send_failed, SendReason}, State} end end; handle_request_frame(Ref, {<<"container">>, ContainerCommand}, State) -> logger:warning("[ws_channel] unsupported request message type: container, ref: ~p, command: ~p", [Ref, ContainerCommand]), {stop, normal, State}; handle_request_frame(Ref, Body, State) -> logger:warning("[ws_channel] unsupported request body, ref: ~p, body: ~p", [Ref, Body]), {stop, normal, State}. -spec handle_message_frame(term(), #state{}) -> {noreply, #state{}} | {stop, term(), #state{}}. handle_message_frame(<<"ping">>, State = #state{transport = Transport, socket = Socket}) -> Packet = term_to_binary({<<"message">>, <<"pong">>}), case Transport:send(Socket, Packet) of ok -> {noreply, State}; {error, Reason} -> logger:warning("[ssl_channel] send pong failed, reason: ~p", [Reason]), {stop, {send_failed, Reason}, State} end; handle_message_frame({<<"data">>, #{<<"route_key">> := RouteKey, <<"metric">> := Metric}}, State = #state{host_pid = HostPid}) when is_pid(HostPid) -> iot_host:handle(HostPid, {data, RouteKey, Metric}), {noreply, State}; handle_message_frame({<<"task_event">>, Event}, State = #state{uuid = UUID, host_pid = HostPid}) when is_binary(UUID), is_pid(HostPid) -> handle_event_stream_frame(UUID, Event), {noreply, State}; handle_message_frame(Body, State) -> logger:warning("[ssl_channel] unsupported message body: ~p", [Body]), {noreply, State}. -spec handle_event_stream_frame(binary(), map()) -> ok. handle_event_stream_frame(UUID, #{<<"task_id">> := TaskId, <<"type">> := <<"close">>, <<"stream">> := Reason}) -> logger:debug("[ssl_channel] get uuid: ~p, task_id: ~p, close with reason: ~p", [UUID, TaskId, Reason]), iot_container_task_sup:close(UUID, TaskId, Reason); handle_event_stream_frame(UUID, #{<<"task_id">> := TaskId, <<"type">> := Type, <<"stream">> := Stream}) -> logger:debug("[ssl_channel] get uuid: ~p, task_id: ~p, type: ~ts, stream: ~ts", [UUID, TaskId, Type, Stream]), iot_container_task_sup:stream(UUID, TaskId, Type, Stream); handle_event_stream_frame(UUID, Event) -> logger:warning("[ssl_channel] invalid task_event, uuid: ~p, event: ~p", [UUID, Event]), ok. -spec handle_command_response_frame(request_ref(), tuple(), #state{}) -> {noreply, #state{}}. handle_command_response_frame(Ref, Reply, State = #state{inflight = Inflight}) when is_binary(Ref) -> case maps:take(Ref, Inflight) of error -> {noreply, State}; {#inflight_command{receiver_pid = ReceiverPid, timer_ref = TimerRef}, NInflight} -> erlang:cancel_timer(TimerRef), deliver_command_response(ReceiverPid, Ref, Reply), {noreply, State#state{inflight = NInflight}} end; handle_command_response_frame(Ref, Reply, State) -> logger:warning("[ws_channel] unexpected command_response frame, ref: ~p, reply: ~p", [Ref, Reply]), {noreply, State}. -spec send_reply_frame(module(), any(), request_ref(), tuple()) -> ok | {error, term()}. send_reply_frame(Transport, Socket, Ref, Reply) -> Packet = term_to_binary({<<"response">>, Ref, Reply}), Transport:send(Socket, Packet). -spec decode_command_response({binary(), term()} | tuple()) -> ok | {ok, term()} | {error, term()}. decode_command_response({<<"container">>, <<"ok">>}) -> ok; decode_command_response({<<"container">>, {<<"ok">>, Result}}) -> {ok, Result}; decode_command_response({<<"container">>, {<<"error">>, Reason}}) -> {error, Reason}; decode_command_response(_Reply) -> {error, invalid_response}. -spec deliver_command_response(undefined | pid(), request_ref(), 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 deliver_command_error(undefined | pid(), request_ref(), term()) -> ok. deliver_command_error(ReceiverPid, Ref, Reason) when is_pid(ReceiverPid) -> ReceiverPid ! {command_reply, Ref, {error, Reason}}, ok; deliver_command_error(_ReceiverPid, _Ref, _Reason) -> ok. -spec reply_command_timeout(undefined | pid(), request_ref()) -> 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(request_ref(), #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()}. auth(Token, UUID, Timestamp) when is_binary(Token), is_binary(UUID), is_integer(Timestamp) -> %% 1分钟内有效 Now = iot_util:current_time(), case Timestamp =< Now andalso Now - Timestamp =< 60 of true -> case efka_client_store:auth(UUID, Token) of true -> ok; false -> {error, <<"invalid token">>} end; false -> {error, <<"invalid timestamp">>} end. -spec encode_command_body({container, map()}) -> {binary(), map()}. encode_command_body({container, Request}) -> {<<"container">>, safe_term(Request)}. -spec safe_term(term()) -> term(). safe_term(true) -> true; safe_term(false) -> false; safe_term(undefined) -> undefined; safe_term(Value) when is_atom(Value) -> atom_to_binary(Value, utf8); safe_term(Value) when is_map(Value) -> maps:from_list([{safe_term(K), safe_term(V)} || {K, V} <- maps:to_list(Value)]); safe_term(Value) when is_list(Value) -> [safe_term(Item) || Item <- Value]; safe_term(Value) when is_tuple(Value) -> list_to_tuple([safe_term(Item) || Item <- tuple_to_list(Value)]); safe_term(Value) -> Value. -spec start_idle_timer() -> reference(). start_idle_timer() -> erlang:start_timer(?SSL_IDLE_TIMEOUT, self(), ssl_idle_timeout). -spec reset_idle_timer(#state{}) -> #state{}. reset_idle_timer(State = #state{idle_timer_ref = TimerRef}) -> cancel_timer(TimerRef), State#state{idle_timer_ref = start_idle_timer()}. -spec cancel_timer(undefined | reference()) -> ok. cancel_timer(undefined) -> ok; cancel_timer(TimerRef) -> _ = erlang:cancel_timer(TimerRef), ok.