From 71af786bce67d2fb28dc85dec5223710f2e0a1f3 Mon Sep 17 00:00:00 2001 From: anlicheng <244108715@qq.com> Date: Wed, 17 Sep 2025 11:53:27 +0800 Subject: [PATCH] fix --- apps/efka/src/efka_service_sup.erl | 9 +-- apps/efka/src/websocket/ws_channel.erl | 105 +++++-------------------- 2 files changed, 20 insertions(+), 94 deletions(-) diff --git a/apps/efka/src/efka_service_sup.erl b/apps/efka/src/efka_service_sup.erl index bb4b937..495b899 100644 --- a/apps/efka/src/efka_service_sup.erl +++ b/apps/efka/src/efka_service_sup.erl @@ -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 diff --git a/apps/efka/src/websocket/ws_channel.erl b/apps/efka/src/websocket/ws_channel.erl index 1f94880..54bbd96 100644 --- a/apps/efka/src/websocket/ws_channel.erl +++ b/apps/efka/src/websocket/ws_channel.erl @@ -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, <>}, State) -> Request = jiffy:decode(Data, [return_maps]), handle_request(Request, State); -%% 处理micro-client:response => efka 的响应 -websocket_handle({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 = <>, - 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 = <>, {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, <>}, 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, <>}, 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, <>}, State} - end + {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. \ No newline at end of file + jiffy:encode(Response, [force_utf8]). \ No newline at end of file