309 lines
14 KiB
Erlang
309 lines
14 KiB
Erlang
%%%-------------------------------------------------------------------
|
|
%%% @author licheng5
|
|
%%% @copyright (C) 2021, <COMPANY>
|
|
%%% @doc
|
|
%%%
|
|
%%% @end
|
|
%%% Created : 11. 1月 2021 上午12:17
|
|
%%%-------------------------------------------------------------------
|
|
-module(ssl_channel).
|
|
-author("licheng5").
|
|
-behaviour(ranch_protocol).
|
|
|
|
-define(INFLIGHT_TIMEOUT, 60000).
|
|
|
|
%% API
|
|
-export([pub/4, container_call/4, cancel_command_call/2, activate/4]).
|
|
|
|
-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,
|
|
|
|
%% iot 发起的 command 与 command_response 的对应关系
|
|
inflight = #{}
|
|
}).
|
|
|
|
-record(inflight_command, {
|
|
receiver_pid :: undefined | pid(),
|
|
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 activate(Pid :: pid(), ReceiverPid :: pid(), Ref :: reference(), Auth :: boolean()) -> ok.
|
|
activate(Pid, ReceiverPid, Ref, Auth) when is_pid(Pid), is_pid(ReceiverPid), is_reference(Ref), is_boolean(Auth) ->
|
|
Command = case Auth of true -> activate; false -> deactivate end,
|
|
gen_server:cast(Pid, {command_call, ReceiverPid, Ref, {auth, Command}}),
|
|
ok.
|
|
|
|
-spec container_call(Pid :: pid(), ReceiverPid :: pid(), Ref :: reference(), Request :: map()) -> ok.
|
|
container_call(Pid, ReceiverPid, Ref, Request) when is_pid(Pid), is_pid(ReceiverPid), is_reference(Ref), is_map(Request) ->
|
|
gen_server:cast(Pid, {command_call, ReceiverPid, Ref, {container, Request}}),
|
|
ok.
|
|
|
|
-spec cancel_command_call(Pid :: pid(), Ref :: reference()) -> ok.
|
|
cancel_command_call(Pid, Ref) when is_pid(Pid), is_reference(Ref) ->
|
|
gen_server:call(Pid, {cancel_command_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_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}}}),
|
|
Transport:send(Socket, Packet),
|
|
{noreply, State};
|
|
|
|
%% 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}),
|
|
Transport:send(Socket, Packet),
|
|
|
|
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)}}.
|
|
|
|
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({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) ->
|
|
try binary_to_term(PacketBin, [safe]) of
|
|
{request, Ref, Body} ->
|
|
handle_request_frame(Ref, Body, Transport, Socket, State);
|
|
{message, Body} ->
|
|
handle_message_frame(Body, HostPid, State);
|
|
{command_response, Ref, Response} ->
|
|
handle_command_response_frame(Ref, Response, Inflight, State);
|
|
Other ->
|
|
logger:warning("[ssl_channel] unsupported packet: ~p", [Other]),
|
|
{stop, bad_packet, State}
|
|
catch error:Error ->
|
|
logger:warning("[ssl_channel] binary_to_term get error: ~p", [Error]),
|
|
{stop, bad_packet, 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{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
|
|
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
|
|
|
|
-spec handle_request_frame(reference(), tuple(), module(), any(), #state{}) -> {noreply, #state{}} | {stop, term(), #state{}}.
|
|
handle_request_frame(Ref, {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, Ref, {auth_response, ok}),
|
|
{noreply, State#state{uuid = UUID, host_pid = HostPid}};
|
|
{denied, Reason} when is_binary(Reason) ->
|
|
erlang:monitor(process, HostPid),
|
|
send_reply_frame(Transport, Socket, Ref, {auth_response, {error, {denied, 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, Ref, {auth_response, {error, {failed, 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, Ref, {auth_response, {error, {failed, Reason}}}),
|
|
logger:warning("[ws_channel] uuid: ~p, token: ~p, auth failed, reason: ~p", [UUID, Token, Reason]),
|
|
{stop, Reason, State}
|
|
end;
|
|
handle_request_frame(Ref, {container, ContainerCommand}, _Transport, _Socket, State) ->
|
|
logger:warning("[ws_channel] unsupported request message type: container, ref: ~p, command: ~p", [Ref, ContainerCommand]),
|
|
{stop, normal, State};
|
|
handle_request_frame(Ref, Body, _Transport, _Socket, State) ->
|
|
logger:warning("[ws_channel] unsupported request body, ref: ~p, body: ~p", [Ref, 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_command_response_frame(reference(), tuple(), map(), #state{}) ->
|
|
{noreply, #state{}}.
|
|
handle_command_response_frame(Ref, Reply, Inflight, State) when is_reference(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, _Inflight, State) ->
|
|
logger:warning("[ws_channel] unexpected command_response frame, ref: ~p, reply: ~p", [Ref, Reply]),
|
|
{noreply, State}.
|
|
|
|
-spec send_reply_frame(module(), any(), reference(), tuple()) -> any().
|
|
send_reply_frame(Transport, Socket, Ref, Reply) ->
|
|
Packet = term_to_binary({response, Ref, Reply}),
|
|
Transport:send(Socket, Packet).
|
|
|
|
-spec decode_command_response({container, ok | {ok, term()} | {error, 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({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()}.
|
|
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.
|