fix
This commit is contained in:
parent
53d0ee9f3a
commit
71af786bce
@ -41,14 +41,7 @@ start_link() ->
|
||||
%% specifications.
|
||||
init([]) ->
|
||||
SupFlags = #{strategy => one_for_one, intensity => 1000, period => 3600},
|
||||
%% 简化逻辑，只启动需要运行的微服务
|
||||
{ok, Services} = service_model:get_running_services(),
|
||||
ServiceIds = lists:map(fun(#service{service_id = ServiceId}) -> ServiceId end, Services),
|
||||
lager:debug("[efka_service_sup] will start services: ~p", [ServiceIds]),
|
||||
|
||||
Specs = lists:map(fun(ServiceId) -> child_spec(ServiceId) end, Services),
|
||||
|
||||
{ok, {SupFlags, Specs}}.
|
||||
{ok, {SupFlags, []}}.
|
||||
|
||||
%%%===================================================================
|
||||
%%% Internal functions
|
||||
|
||||
@ -13,23 +13,15 @@
|
||||
%% API
|
||||
-export([init/2]).
|
||||
-export([websocket_init/1, websocket_handle/2, websocket_info/2, terminate/3]).
|
||||
-export([invoke/4]).
|
||||
|
||||
%% 最大的等待时间
|
||||
-define(PENDING_TIMEOUT, 10 * 1000).
|
||||
|
||||
-record(state, {
|
||||
packet_id = 1,
|
||||
container_pid :: undefined | pid(),
|
||||
is_registered = false :: boolean(),
|
||||
%% 请求响应的对应关系, #{packet_id => {ReceiverPid, Ref}}; 自身的inflight需要超时逻辑处理
|
||||
inflight = #{}
|
||||
service_pid :: undefined | pid(),
|
||||
is_registered = false :: boolean()
|
||||
}).
|
||||
|
||||
-spec invoke(ChannelPid :: pid(), Ref :: reference(), ReceiverPid :: pid(), Payload :: binary()) -> no_return().
|
||||
invoke(ChannelPid, Ref, ReceiverPid, Payload) when is_pid(ChannelPid), is_pid(ReceiverPid), is_binary(Payload), is_reference(Ref) ->
|
||||
ChannelPid ! {invoke, Ref, ReceiverPid, Payload}.
|
||||
|
||||
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
|
||||
%% 逻辑处理方法
|
||||
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
|
||||
@ -40,51 +32,26 @@ init(Req, Opts) ->
|
||||
websocket_init(_State) ->
|
||||
lager:debug("[ws_channel] get a new connection"),
|
||||
%% 初始状态为true
|
||||
{ok, #state{packet_id = 1}}.
|
||||
{ok, #state{}}.
|
||||
|
||||
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 = 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({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 = ws_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 = channel_reply(Id, {error, <<"timeout">>}, Inflight),
|
||||
{ok, State#state{inflight = NInflight}};
|
||||
|
||||
%% service进程关闭
|
||||
websocket_info({'DOWN', _Ref, process, ContainerPid, Reason}, State = #state{container_pid = ContainerPid}) ->
|
||||
websocket_info({'DOWN', _Ref, process, ContainerPid, Reason}, State = #state{service_pid = ContainerPid}) ->
|
||||
lager:debug("[tcp_channel] container_pid: ~p, exited: ~p", [ContainerPid, Reason]),
|
||||
{stop, State#state{container_pid = undefined}};
|
||||
{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]),
|
||||
@ -110,36 +77,27 @@ terminate(Reason, _Req, State) ->
|
||||
|
||||
%% 注册, 要建立程序和容器之间的关系
|
||||
handle_request(#{<<"id">> := Id, <<"method">> := <<"register">>, <<"params">> := #{<<"service_id">> := ServiceId}}, State) ->
|
||||
case efka_service:get_pid(ServiceId) of
|
||||
undefined ->
|
||||
lager:warning("[ws_channel] container_name: ~p, not running", [ContainerName]),
|
||||
Reply = json_error(Id, -1, <<"container not running">>),
|
||||
delay_stop(10, normal),
|
||||
{reply, {binary, <<?PACKET_RESPONSE:8, Reply/binary>>}, State};
|
||||
ContainerPid when is_pid(ContainerPid) ->
|
||||
case efka_container:attach_channel(ContainerPid, self(), MetaTag) of
|
||||
ok ->
|
||||
Reply = json_result(Id, <<"ok">>),
|
||||
erlang:monitor(process, ContainerPid),
|
||||
{ok, ServicePid} = efka_service_sup:start_service(ServiceId),
|
||||
case efka_service:attach_channel(ServicePid, self()) of
|
||||
ok ->
|
||||
Reply = json_result(Id, <<"ok">>),
|
||||
erlang:monitor(process, ServicePid),
|
||||
|
||||
{reply, {binary, <<?PACKET_RESPONSE:8, Reply/binary>>}, State#state{container_pid = ContainerPid, is_registered = true}};
|
||||
{error, Error} ->
|
||||
lager:warning("[ws_channel] container_name: ~p, attach_channel get error: ~p", [ContainerName, Error]),
|
||||
Reply = json_error(Id, -1, Error),
|
||||
delay_stop(10, normal),
|
||||
{reply, {binary, <<?PACKET_RESPONSE:8, Reply/binary>>}, State}
|
||||
end
|
||||
{reply, {binary, <<?PACKET_RESPONSE:8, Reply/binary>>}, State#state{service_pid = ServicePid, is_registered = true}};
|
||||
{error, Error} ->
|
||||
lager:warning("[ws_channel] service_id: ~p, attach_channel get error: ~p", [ServiceId, Error]),
|
||||
{stop, State}
|
||||
end;
|
||||
|
||||
%% 数据项
|
||||
handle_request(#{<<"id">> := 0, <<"method">> := <<"metric_data">>,
|
||||
<<"params">> := #{<<"device_uuid">> := DeviceUUID, <<"route_key">> := RouteKey, <<"metric">> := Metric}}, State = #state{container_pid = ContainerPid, is_registered = true}) ->
|
||||
efka_container:metric_data(ContainerPid, DeviceUUID, RouteKey, Metric),
|
||||
<<"params">> := #{<<"device_uuid">> := DeviceUUID, <<"route_key">> := RouteKey, <<"metric">> := Metric}}, State = #state{service_pid = ServicePid, is_registered = true}) ->
|
||||
efka_container:metric_data(ServicePid, DeviceUUID, RouteKey, Metric),
|
||||
{ok, State};
|
||||
|
||||
%% Event事件
|
||||
handle_request(#{<<"id">> := 0, <<"method">> := <<"event">>, <<"params">> := #{<<"event_type">> := EventType, <<"body">> := Body}}, State = #state{container_pid = ContainerPid, is_registered = true}) ->
|
||||
efka_container:send_event(ContainerPid, EventType, Body),
|
||||
handle_request(#{<<"id">> := 0, <<"method">> := <<"event">>, <<"params">> := #{<<"event_type">> := EventType, <<"body">> := Body}}, State = #state{service_pid = ServicePid, is_registered = true}) ->
|
||||
efka_container:send_event(ServicePid, EventType, Body),
|
||||
{ok, State};
|
||||
|
||||
%% 订阅事件
|
||||
@ -147,16 +105,6 @@ handle_request(#{<<"id">> := 0, <<"method">> := <<"subscribe">>, <<"params">> :=
|
||||
efka_subscription:subscribe(Topic, self()),
|
||||
{ok, State}.
|
||||
|
||||
delay_stop(Timeout, Reason) when is_integer(Timeout) ->
|
||||
erlang:start_timer(Timeout, self(), {stop, Reason}).
|
||||
|
||||
%% 采用32位编码
|
||||
-spec next_packet_id(PacketId :: integer()) -> NextPacketId :: integer().
|
||||
next_packet_id(PacketId) when PacketId >= 4294967295 ->
|
||||
1;
|
||||
next_packet_id(PacketId) ->
|
||||
PacketId + 1.
|
||||
|
||||
-spec json_result(Id :: integer(), Result :: term()) -> binary().
|
||||
json_result(Id, Result) when is_integer(Id) ->
|
||||
Response = #{
|
||||
@ -174,19 +122,4 @@ json_error(Id, Code, Message) when is_integer(Id), is_integer(Code), is_binary(M
|
||||
<<"message">> => Message
|
||||
}
|
||||
},
|
||||
jiffy:encode(Response, [force_utf8]).
|
||||
|
||||
%% 超时逻辑处理
|
||||
channel_reply(Id, Reply, Inflight) ->
|
||||
case maps:take(Id, Inflight) of
|
||||
error ->
|
||||
Inflight;
|
||||
{{ReceiverPid, Ref}, NInflight} ->
|
||||
case is_pid(ReceiverPid) andalso is_process_alive(ReceiverPid) of
|
||||
true ->
|
||||
ReceiverPid ! {channel_reply, Ref, Reply};
|
||||
false ->
|
||||
ok
|
||||
end,
|
||||
NInflight
|
||||
end.
|
||||
jiffy:encode(Response, [force_utf8]).
|
||||
Loading…
x
Reference in New Issue
Block a user