%%%------------------------------------------------------------------- %%% @author licheng5 %%% @copyright (C) 2021, %%% @doc %%% %%% @end %%% Created : 11. 1月 2021 上午12:17 %%%------------------------------------------------------------------- -module(tcp_channel). -author("licheng5"). -include("protocol.hrl"). -include("message_pb.hrl"). -behaviour(ranch_protocol). -define(MAX_PACKET_ID, 16#FFFFFFFF). -define(INFLIGHT_TIMEOUT, 60000). %% API -export([pub/4, jsonrpc_call/3, cancel_jsonrpc_call/2, command/3]). -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 = #{} }). %% 向通道中写入消息 -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(), CommandType :: integer(), Command :: binary()) -> no_return(). command(Pid, CommandType, Command) when is_pid(Pid), is_integer(CommandType), is_binary(Command) -> gen_server:cast(Pid, {command, CommandType, Command}). %% 向通道中写入消息 -spec jsonrpc_call(Pid :: pid(), ReceiverPid :: pid(), Request :: {Method :: binary(), Params :: any()}) -> Ref :: reference(). jsonrpc_call(Pid, ReceiverPid, Request = {Method, _Params}) when is_pid(Pid), is_pid(ReceiverPid), is_binary(Method) -> Ref = make_ref(), gen_server:cast(Pid, {jsonrpc_call, ReceiverPid, Ref, Request}), Ref. -spec cancel_jsonrpc_call(Pid :: pid(), Ref :: reference()) -> ok. cancel_jsonrpc_call(Pid, Ref) when is_pid(Pid), is_reference(Ref) -> gen_server:call(Pid, {cancel_jsonrpc_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("[sdlan_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_jsonrpc_call, Ref}, _From, State = #state{inflight = Inflight}) -> case take_inflight_by_ref(Ref, Inflight) of {ok, _PacketId, {_ReceiverPid, _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}) -> Encoded = message_pb:encode_msg(#'Pub'{topic = Topic, qos = Qos, content = Content}), EncPub = <>, Transport:send(Socket, <>), {noreply, State}; %% 发送Command消息 handle_cast({command, CommandType, Command}, State = #state{transport = Transport, socket = Socket}) -> Encoded = message_pb:encode_msg(#'Command'{command_type = CommandType, command = Command}), EncCommand = <>, Transport:send(Socket, <>), {noreply, State}; %% 推送消息 handle_cast({jsonrpc_call, ReceiverPid, Ref, {Method, Params}}, State = #state{transport = Transport, socket = Socket, packet_id = PacketId, inflight = Inflight}) when is_binary(Method) -> case next_packet_id(PacketId, Inflight) of {ok, NPacketId, NextPacketId} -> Encoded = message_pb:encode_msg(#'JsonRpcRequest'{ method = Method, params = erlang:term_to_binary(Params) }), EncRequest = <>, TimerRef = erlang:start_timer(?INFLIGHT_TIMEOUT, self(), {jsonrpc_timeout, NPacketId}), Transport:send(Socket, <>), {noreply, State#state{ packet_id = NextPacketId, inflight = maps:put(NPacketId, {ReceiverPid, Ref, TimerRef}, Inflight) }}; {error, inflight_full} -> logger:warning("[ws_channel] uuid: ~p, inflight requests exhausted", [State#state.uuid]), {noreply, State} end. %% auth验证 handle_info({tcp, Socket, <>}, State = #state{transport = Transport, socket = Socket}) -> #'AuthRequest'{uuid = UUID, username = Username, token = Token, salt = Salt, timestamp = Timestamp} = message_pb:decode_msg(RequestBin, 'AuthRequest'), logger:debug("[ws_channel] auth uuid: ~p", [UUID]), case iot_auth:check(Username, Token, UUID, Salt, Timestamp) of true -> case iot_api_client:get_host_by_uuid(UUID) of undefined -> logger:warning("[ws_channel] uuid: ~p, user: ~p, host not found", [UUID, Username]), {stop, State}; {ok, _} -> %% 尝试启动主机的服务进程 {ok, HostPid} = iot_host_sup:ensured_host_started(UUID), case iot_host:attach_channel(HostPid, self()) of ok -> %% 建立到host的monitor erlang:monitor(process, HostPid), Encoded = message_pb:encode_msg(#'AuthReply'{code = 0, payload = <<"ok">>}, 'AuthReply'), AuthReplyBin = <>, Transport:send(Socket, <>), {noreply, State#state{uuid = UUID, host_pid = HostPid}}; {denied, Reason} when is_binary(Reason) -> erlang:monitor(process, HostPid), Encoded = message_pb:encode_msg(#'AuthReply'{code = 1, payload = Reason}, 'AuthReply'), AuthReplyBin = <>, Transport:send(Socket, <>), 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) -> Encoded = message_pb:encode_msg(#'AuthReply'{code = 2, payload = Reason}, 'AuthReply'), AuthReplyBin = <>, Transport:send(Socket, <>), logger:debug("[ws_channel] uuid: ~p, attach channel get error: ~p, stop channel", [UUID, Reason]), {stop, State} end end; false -> logger:warning("[ws_channel] uuid: ~p, user: ~p, auth failed", [UUID, Username]), {stop, State} end; handle_info({tcp, Socket, <>}, State = #state{socket = Socket}) -> logger:warning("[ws_channel] unsupported request message type: ~p", [MsgType]), {stop, State}; handle_info({tcp, Socket, <>}, State = #state{socket = Socket, host_pid = HostPid}) when is_pid(HostPid) -> CastMessage = message_pb:decode_msg(CastBin, 'Data'), case CastMessage of #'Data'{} = Data -> iot_host:handle(HostPid, {data, Data}) end, {noreply, State}; handle_info({tcp, Socket, <>}, State = #state{socket = Socket, host_pid = HostPid}) when is_pid(HostPid) -> CastMessage = message_pb:decode_msg(CastBin, 'TaskEventStream'), case CastMessage of #'TaskEventStream'{task_id = TaskId, type = Type0, stream = Reason0} when Type0 =:= <<"close">>; Type0 =:= [<<"close">>] -> iot_event_stream_observer:stream_close(TaskId, iolist_to_binary(Reason0)); #'TaskEventStream'{task_id = TaskId, type = Type, stream = Stream} -> TypeBin = iolist_to_binary(Type), StreamBin = iolist_to_binary(Stream), logger:debug("[tcp_channel] get task_id: ~p, type: ~ts, stream: ~ts", [TaskId, TypeBin, StreamBin]), iot_event_stream_observer:stream_data(TaskId, TypeBin, StreamBin) end, {noreply, State}; handle_info({tcp, Socket, <>}, State = #state{socket = Socket}) -> logger:warning("[tcp_channel] unsupported cast message type: ~p", [MsgType]), {noreply, State}; %handle_info({tcp, Socket, <>}, State = #state{socket = Socket, host_pid = HostPid}) when is_pid(HostPid) -> % Ping = message_pb:decode_msg(PingData, ping), % iot_host:handle(HostPid, {ping, Ping}), % {noreply, State}; %% 主机端的消息响应 handle_info({tcp, Socket, <>}, State = #state{socket = Socket, inflight = Inflight}) when PacketId > 0 -> RpcReply = message_pb:decode_msg(ReplyBin, 'JsonRpcReply'), case maps:take(PacketId, Inflight) of error -> {noreply, State}; {{ReceiverPid, Ref, TimerRef}, NInflight} -> erlang:cancel_timer(TimerRef), case is_pid(ReceiverPid) andalso is_process_alive(ReceiverPid) of true -> ReceiverPid ! {jsonrpc_reply, Ref, RpcReply}; false -> logger:warning("[ws_channel] get async_call_reply message: ~p, packet_id: ~p, but receiver_pid is deaded", [RpcReply, PacketId]) end, {noreply, State#state{inflight = NInflight}} end; handle_info({timeout, TimerRef, {jsonrpc_timeout, PacketId}}, State = #state{inflight = Inflight}) -> case maps:get(PacketId, Inflight, undefined) of {_ReceiverPid, Ref, TimerRef} -> logger:warning("[ws_channel] jsonrpc request timeout, packet_id: ~p, ref: ~p", [PacketId, Ref]), {noreply, State#state{inflight = maps:remove(PacketId, Inflight)}}; _ -> {noreply, State} end; handle_info({tcp_error, Sock, Reason}, State = #state{socket = Sock}) -> logger:notice("[sdlan_channel] tcp_error: ~p", [Reason]), {stop, normal, State}; handle_info({tcp_closed, Sock}, State = #state{socket = Sock}) -> logger:notice("[sdlan_channel] tcp_closed"), {stop, normal, 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, State}; handle_info(Info, State) -> logger:warning("[sdlan_channel] get a unknown message: ~p, channel will closed, state: ~p", [Info, State]), {noreply, State}. terminate(Reason, #state{}) -> logger:warning("[sdlan_channel] stop with reason: ~p", [Reason]), ok. code_change(_OldVsn, State, _Extra) -> {ok, State}. take_inflight_by_ref(Ref, Inflight) -> maps:fold(fun(PacketId, Value = {_ReceiverPid, PacketRef, _TimerRef}, Acc) -> case Acc of error when PacketRef =:= Ref -> {ok, PacketId, Value, maps:remove(PacketId, Inflight)}; _ -> Acc end end, error, Inflight). next_packet_id(_PacketId, Inflight) when map_size(Inflight) >= ?MAX_PACKET_ID -> {error, inflight_full}; next_packet_id(PacketId, Inflight) -> next_packet_id(PacketId, Inflight, PacketId, 0). next_packet_id(_PacketId, _Inflight, _StartPacketId, TryCount) when TryCount > ?MAX_PACKET_ID -> {error, inflight_full}; next_packet_id(PacketId, Inflight, StartPacketId, TryCount) -> case maps:is_key(PacketId, Inflight) of false -> {ok, PacketId, inc_packet_id(PacketId)}; true -> NPacketId = inc_packet_id(PacketId), case NPacketId =:= StartPacketId of true -> {error, inflight_full}; false -> next_packet_id(NPacketId, Inflight, StartPacketId, TryCount + 1) end end. inc_packet_id(?MAX_PACKET_ID) -> 1; inc_packet_id(PacketId) when PacketId > 0, PacketId < ?MAX_PACKET_ID -> PacketId + 1.