150 lines
6.7 KiB
Erlang
150 lines
6.7 KiB
Erlang
%%%-------------------------------------------------------------------
|
|
%%% @author licheng5
|
|
%%% @copyright (C) 2021, <COMPANY>
|
|
%%% @doc
|
|
%%%
|
|
%%% @end
|
|
%%% Created : 11. 1月 2021 上午12:17
|
|
%%%-------------------------------------------------------------------
|
|
-module(ws_channel).
|
|
-author("licheng5").
|
|
-include("efka_service.hrl").
|
|
|
|
%% API
|
|
-export([init/2]).
|
|
-export([websocket_init/1, websocket_handle/2, websocket_info/2, terminate/3]).
|
|
|
|
%% 最大的等待时间
|
|
-define(PENDING_TIMEOUT, 10 * 1000).
|
|
|
|
-record(state, {
|
|
packet_id = 1,
|
|
service_id :: undefined | binary(),
|
|
service_pid :: undefined | pid(),
|
|
is_registered = false :: boolean(),
|
|
|
|
%% 请求响应的对应关系, #{packet_id => {ReceiverPid, Ref}}; 自身的inflight需要超时逻辑处理
|
|
inflight = #{}
|
|
}).
|
|
|
|
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
|
|
%% 逻辑处理方法
|
|
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
|
|
|
|
init(Req, Opts) ->
|
|
{cowboy_websocket, Req, Opts}.
|
|
|
|
websocket_init(_State) ->
|
|
lager:debug("[ws_channel] get a new connection"),
|
|
%% 初始状态为true
|
|
{ok, #state{packet_id = 1}}.
|
|
|
|
websocket_handle({binary, <<?PACKET_REQUEST, Data/binary>>}, State) ->
|
|
Request = jiffy:decode(Data, [return_maps]),
|
|
handle_request(Request, State);
|
|
|
|
%% 处理micro-client:response => efka 的响应
|
|
websocket_handle({binary, <<?PACKET_RESPONSE:8, Data/binary>>}, State = #state{inflight = Inflight}) ->
|
|
{PacketId, Reply} = case jiffy:decode(Data, [return_maps]) of
|
|
#{<<"id">> := Id, <<"result">> := Result} ->
|
|
{Id, {ok, Result}};
|
|
#{<<"id">> := Id, <<"error">> := #{<<"code">> := _Code, <<"message">> := Error}} ->
|
|
{Id, {error, Error}}
|
|
end,
|
|
NInflight = gen_channel:channel_reply(PacketId, Reply, Inflight),
|
|
{ok, State#state{inflight = NInflight}};
|
|
|
|
websocket_handle(Info, State) ->
|
|
lager:error("[ws_channel] get a unknown message: ~p, channel will closed", [Info]),
|
|
{stop, State}.
|
|
|
|
websocket_info({push_config, Ref, ReceiverPid, ConfigJson}, State = #state{packet_id = PacketId, inflight = Inflight}) ->
|
|
PushConfig = #{<<"id">> => PacketId, <<"method">> => <<"push_config">>, <<"params">> => #{<<"config">> => ConfigJson}},
|
|
Packet = jiffy:encode(PushConfig, [force_utf8]),
|
|
Reply = <<?PACKET_PUSH:8, Packet/binary>>,
|
|
erlang:start_timer(?PENDING_TIMEOUT, self(), {pending_timeout, PacketId}),
|
|
|
|
{reply, {binary, Reply}, State#state{packet_id = gen_channel:next_packet_id(PacketId), inflight = maps:put(PacketId, {ReceiverPid, Ref}, Inflight)}};
|
|
|
|
%% 远程调用
|
|
websocket_info({invoke, Ref, ReceiverPid, Payload}, State = #state{packet_id = PacketId, inflight = Inflight}) ->
|
|
PushConfig = #{<<"id">> => PacketId, <<"method">> => <<"invoke">>, <<"params">> => #{<<"payload">> => Payload}},
|
|
Packet = jiffy:encode(PushConfig, [force_utf8]),
|
|
Reply = <<?PACKET_PUSH:8, Packet/binary>>,
|
|
erlang:start_timer(?PENDING_TIMEOUT, self(), {pending_timeout, PacketId}),
|
|
|
|
{reply, {binary, Reply}, State#state{packet_id = gen_channel:next_packet_id(PacketId), inflight = maps:put(PacketId, {ReceiverPid, Ref}, Inflight)}};
|
|
|
|
%% 订阅的消息
|
|
websocket_info({topic_broadcast, Topic, Content}, State = #state{}) ->
|
|
Packet = jiffy:encode(#{<<"topic">> => Topic, <<"content">> => Content}, [force_utf8]),
|
|
Reply = <<?PACKET_PUB:8, Packet/binary>>,
|
|
{reply, {binary, Reply}, State};
|
|
|
|
%% 超时逻辑处理
|
|
websocket_info({timeout, _, {pending_timeout, Id}}, State = #state{inflight = Inflight}) ->
|
|
NInflight = gen_channel:channel_reply(Id, {error, <<"timeout">>}, Inflight),
|
|
{ok, State#state{inflight = NInflight}};
|
|
|
|
%% service进程关闭
|
|
websocket_info({'DOWN', _Ref, process, ServicePid, Reason}, State = #state{service_pid = ServicePid}) ->
|
|
lager:debug("[tcp_channel] service_pid: ~p, exited: ~p", [ServicePid, Reason]),
|
|
{stop, State#state{service_pid = undefined}};
|
|
|
|
websocket_info({timeout, _, {stop, Reason}}, State) ->
|
|
lager:debug("[ws_channel] the channel will be closed with reason: ~p", [Reason]),
|
|
{stop, State};
|
|
%% 处理关闭信号
|
|
websocket_info({stop, Reason}, State) ->
|
|
lager:debug("[ws_channel] the channel will be closed with reason: ~p", [Reason]),
|
|
{stop, State};
|
|
|
|
%% 处理其他未知消息
|
|
websocket_info(Info, State) ->
|
|
lager:debug("[ws_channel] channel get unknown info: ~p", [Info]),
|
|
{ok, State}.
|
|
|
|
%% 进程关闭事件
|
|
terminate(Reason, _Req, State) ->
|
|
lager:debug("[ws_channel] channel close with reason: ~p, state is: ~p", [Reason, State]),
|
|
ok.
|
|
|
|
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
|
|
%% helper methods
|
|
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
|
|
|
|
%% 注册
|
|
handle_request(#{<<"id">> := Id, <<"method">> := <<"register">>, <<"params">> := #{<<"service_id">> := ServiceId}}, State) ->
|
|
case gen_channel:register(Id, ServiceId) of
|
|
{error, Reply} ->
|
|
delay_stop(10, normal),
|
|
{reply, {binary, <<?PACKET_RESPONSE:8, Reply/binary>>}, State};
|
|
{ok, Reply, ServicePid} ->
|
|
delay_stop(10, normal),
|
|
{reply, {binary, <<?PACKET_RESPONSE:8, Reply/binary>>}, State#state{service_id = ServiceId, service_pid = ServicePid, is_registered = true}}
|
|
end;
|
|
%% 请求参数
|
|
handle_request(#{<<"id">> := Id, <<"method">> := <<"request_config">>}, State = #state{service_pid = ServicePid, is_registered = true}) ->
|
|
{ok, ConfigJson} = efka_service:request_config(ServicePid),
|
|
Packet = gen_channel:json_result(Id, ConfigJson),
|
|
Reply = <<?PACKET_RESPONSE:8, Packet/binary>>,
|
|
{reply, {binary, Reply}, State};
|
|
|
|
%% 数据项
|
|
handle_request(#{<<"id">> := 0, <<"method">> := <<"metric_data">>,
|
|
<<"params">> := #{<<"device_uuid">> := DeviceUUID, <<"route_key">> := RouteKey, <<"metric">> := Metric}}, State = #state{service_pid = ServicePid, is_registered = true}) ->
|
|
efka_service:metric_data(ServicePid, DeviceUUID, RouteKey, Metric),
|
|
{ok, State};
|
|
|
|
%% Event事件
|
|
handle_request(#{<<"id">> := 0, <<"method">> := <<"event">>, <<"params">> := #{<<"event_type">> := EventType, <<"body">> := Body}}, State = #state{service_pid = ServicePid, is_registered = true}) ->
|
|
efka_service:send_event(ServicePid, EventType, Body),
|
|
{ok, State};
|
|
|
|
%% 订阅事件
|
|
handle_request(#{<<"id">> := 0, <<"method">> := <<"subscribe">>, <<"params">> := #{<<"topic">> := Topic}}, State = #state{is_registered = true}) ->
|
|
efka_subscription:subscribe(Topic, self()),
|
|
{ok, State}.
|
|
|
|
delay_stop(Timeout, Reason) when is_integer(Timeout) ->
|
|
erlang:start_timer(Timeout, self(), {stop, Reason}). |