324 lines
15 KiB
Erlang
324 lines
15 KiB
Erlang
%%%-------------------------------------------------------------------
|
|
%%% @author licheng5
|
|
%%% @copyright (C) 2021, <COMPANY>
|
|
%%% @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 = #{}
|
|
}).
|
|
|
|
-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(), 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 :: binary()}) -> Ref :: reference().
|
|
jsonrpc_call(Pid, ReceiverPid, Request = {Method, Params}) when is_pid(Pid), is_pid(ReceiverPid), is_binary(Method), is_binary(Params) ->
|
|
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, #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}) ->
|
|
Encoded = message_pb:encode_msg(#'CastFrame'{
|
|
body = {pub, #'Pub'{topic = Topic, qos = Qos, content = Content}}
|
|
}),
|
|
Transport:send(Socket, <<?FRAME_CAST, Encoded/binary>>),
|
|
{noreply, State};
|
|
|
|
%% 发送Command消息
|
|
handle_cast({command, CommandType, Command}, State = #state{transport = Transport, socket = Socket}) ->
|
|
Encoded = message_pb:encode_msg(#'CastFrame'{
|
|
body = {command, #'Command'{command_type = CommandType, command = Command}}
|
|
}),
|
|
Transport:send(Socket, <<?FRAME_CAST, Encoded/binary>>),
|
|
{noreply, State};
|
|
|
|
%% 推送消息
|
|
handle_cast({jsonrpc_call, ReceiverPid, Ref, {Method, Params}}, State = #state{transport = Transport, socket = Socket, packet_id = PacketId, inflight = Inflight}) ->
|
|
case next_packet_id(PacketId, Inflight) of
|
|
{ok, NPacketId, NextPacketId} ->
|
|
Encoded = message_pb:encode_msg(#'RequestFrame'{
|
|
packet_id = NPacketId,
|
|
body = {jsonrpc_request, #'JsonRpcRequest'{method = Method, params = Params}}
|
|
}),
|
|
TimerRef = erlang:start_timer(?INFLIGHT_TIMEOUT, self(), {jsonrpc_timeout, NPacketId}),
|
|
Transport:send(Socket, <<?FRAME_REQUEST, Encoded/binary>>),
|
|
|
|
RequestInfo = #inflight_request{receiver_pid = ReceiverPid, ref = Ref, timer_ref = TimerRef},
|
|
{noreply, State#state{packet_id = NextPacketId, inflight = maps:put(NPacketId, RequestInfo, Inflight)}};
|
|
{error, inflight_full} ->
|
|
logger:warning("[ws_channel] uuid: ~p, inflight requests exhausted", [State#state.uuid]),
|
|
{noreply, State}
|
|
end.
|
|
|
|
%% auth验证
|
|
handle_info({tcp, Socket, <<?FRAME_REQUEST, FrameBin/binary>>}, State = #state{transport = Transport, socket = Socket}) ->
|
|
RequestFrame = message_pb:decode_msg(FrameBin, 'RequestFrame'),
|
|
handle_request_frame(RequestFrame, Transport, Socket, State);
|
|
|
|
handle_info({tcp, Socket, <<?FRAME_CAST, FrameBin/binary>>}, State = #state{socket = Socket, host_pid = HostPid}) ->
|
|
CastFrame = message_pb:decode_msg(FrameBin, 'CastFrame'),
|
|
handle_cast_frame(CastFrame, HostPid, State);
|
|
|
|
%% 主机端的消息响应
|
|
handle_info({tcp, Socket, <<?FRAME_RESPONSE, FrameBin/binary>>}, State = #state{socket = Socket, inflight = Inflight}) ->
|
|
ResponseFrame = message_pb:decode_msg(FrameBin, 'ResponseFrame'),
|
|
handle_response_frame(ResponseFrame, Inflight, State);
|
|
|
|
handle_info({timeout, TimerRef, {jsonrpc_timeout, PacketId}}, State = #state{inflight = Inflight}) ->
|
|
case maps:get(PacketId, Inflight, undefined) of
|
|
#inflight_request{ref = Ref, timer_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}.
|
|
|
|
|
|
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
|
|
%%%% 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 next_packet_id(integer(), map()) ->
|
|
{ok, integer(), integer()} | {error, inflight_full}.
|
|
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).
|
|
|
|
-spec next_packet_id(integer(), map(), integer(), non_neg_integer()) ->
|
|
{ok, integer(), integer()} | {error, inflight_full}.
|
|
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.
|
|
|
|
-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(message_pb:'RequestFrame'(), module(), any(), #state{}) ->
|
|
{noreply, #state{}} | {stop, #state{}}.
|
|
handle_request_frame(#'RequestFrame'{packet_id = PacketId,
|
|
body = {auth_request, #'AuthRequest'{uuid = UUID, username = Username, token = Token, salt = Salt, timestamp = Timestamp}}},
|
|
Transport, Socket, State) ->
|
|
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 ->
|
|
erlang:monitor(process, HostPid),
|
|
send_response_frame(Transport, Socket, PacketId,
|
|
{auth_reply, #'AuthReply'{code = 0, payload = <<"ok">>}}),
|
|
{noreply, State#state{uuid = UUID, host_pid = HostPid}};
|
|
{denied, Reason} when is_binary(Reason) ->
|
|
erlang:monitor(process, HostPid),
|
|
send_response_frame(Transport, Socket, PacketId,
|
|
{auth_reply, #'AuthReply'{code = 1, payload = 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_response_frame(Transport, Socket, PacketId,
|
|
{auth_reply, #'AuthReply'{code = 2, payload = Reason}}),
|
|
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_request_frame(#'RequestFrame'{packet_id = PacketId, body = {jsonrpc_request, RpcRequest}}, _Transport, _Socket, State) ->
|
|
logger:warning("[ws_channel] unsupported request message type: jsonrpc_request, packet_id: ~p, request: ~p", [PacketId, RpcRequest]),
|
|
{stop, State};
|
|
handle_request_frame(#'RequestFrame'{packet_id = PacketId, body = undefined}, _Transport, _Socket, State) ->
|
|
logger:warning("[ws_channel] empty request frame, packet_id: ~p", [PacketId]),
|
|
{stop, State}.
|
|
|
|
-spec handle_cast_frame(message_pb:'CastFrame'(), undefined | pid(), #state{}) ->
|
|
{noreply, #state{}}.
|
|
handle_cast_frame(#'CastFrame'{body = {data, Data}}, HostPid, State) when is_pid(HostPid) ->
|
|
iot_host:handle(HostPid, {data, Data}),
|
|
{noreply, State};
|
|
handle_cast_frame(#'CastFrame'{body = {event_stream, CastMessage}}, HostPid, State) when is_pid(HostPid) ->
|
|
handle_event_stream_frame(CastMessage),
|
|
{noreply, State};
|
|
handle_cast_frame(#'CastFrame'{body = Body}, _HostPid, State) ->
|
|
logger:warning("[tcp_channel] unsupported cast message type: command, body: ~p", [Body]),
|
|
{noreply, State}.
|
|
|
|
-spec handle_event_stream_frame(message_pb:'TaskEventStream'()) -> any().
|
|
handle_event_stream_frame(#'TaskEventStream'{task_id = TaskId, type = Type0, stream = Reason0}) when Type0 =:= <<"close">> ->
|
|
iot_event_stream_observer:stream_close(TaskId, iolist_to_binary(Reason0));
|
|
handle_event_stream_frame(#'TaskEventStream'{task_id = TaskId, type = Type, stream = Stream}) ->
|
|
logger:debug("[tcp_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(message_pb:'ResponseFrame'(), map(), #state{}) ->
|
|
{noreply, #state{}}.
|
|
handle_response_frame(#'ResponseFrame'{packet_id = PacketId, body = {jsonrpc_reply, RpcReply}}, 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 ! {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_response_frame(#'ResponseFrame'{packet_id = PacketId, body = {auth_reply, AuthReply}}, _Inflight, State) ->
|
|
logger:warning("[ws_channel] unexpected auth_reply response, packet_id: ~p, body: ~p", [PacketId, AuthReply]),
|
|
{noreply, State};
|
|
handle_response_frame(#'ResponseFrame'{packet_id = PacketId, body = undefined}, _Inflight, State) ->
|
|
logger:warning("[ws_channel] empty response frame, packet_id: ~p", [PacketId]),
|
|
{noreply, State}.
|
|
|
|
-spec send_response_frame(module(), any(), non_neg_integer(), tuple()) -> any().
|
|
send_response_frame(Transport, Socket, PacketId, Body) ->
|
|
Encoded = message_pb:encode_msg(#'ResponseFrame'{packet_id = PacketId, body = Body}),
|
|
Transport:send(Socket, <<?FRAME_RESPONSE, Encoded/binary>>).
|