From 7e4611d16b7d67a4299b47e041dfe53803a4aa0d Mon Sep 17 00:00:00 2001 From: anlicheng <244108715@qq.com> Date: Tue, 16 Sep 2025 19:05:46 +0800 Subject: [PATCH] fix channel --- apps/efka/src/docker/efka_container.erl | 6 +- .../ws_channel.erl | 86 +++++++------------ .../ws_server.erl | 0 3 files changed, 32 insertions(+), 60 deletions(-) rename apps/efka/src/{websocket_server => websocket}/ws_channel.erl (67%) rename apps/efka/src/{websocket_server => websocket}/ws_server.erl (100%) diff --git a/apps/efka/src/docker/efka_container.erl b/apps/efka/src/docker/efka_container.erl index 35acdf8..f1e1c1d 100644 --- a/apps/efka/src/docker/efka_container.erl +++ b/apps/efka/src/docker/efka_container.erl @@ -69,9 +69,9 @@ metric_data(Pid, DeviceUUID, RouteKey, Metric) when is_pid(Pid), is_binary(Devic send_event(Pid, EventType, Params) when is_pid(Pid), is_integer(EventType), is_binary(Params) -> gen_server:cast(Pid, {send_event, EventType, Params}). --spec attach_channel(Pid :: pid(), ChannelPid :: pid(), Meta :: binary()) -> ok | {error, Reason :: binary()}. -attach_channel(Pid, ChannelPid, Meta) when is_pid(Pid), is_pid(ChannelPid), is_binary(Meta) -> - gen_server:call(Pid, {attach_channel, ChannelPid, Meta}). +-spec attach_channel(Pid :: pid(), ChannelPid :: pid(), MetaTag :: binary()) -> ok | {error, Reason :: binary()}. +attach_channel(Pid, ChannelPid, MetaTag) when is_pid(Pid), is_pid(ChannelPid), is_binary(MetaTag) -> + gen_server:call(Pid, {attach_channel, ChannelPid, MetaTag}). %% @doc Spawns the server and registers the local name (unique) -spec(start_link(Name :: atom(), ContainerId :: binary()) -> diff --git a/apps/efka/src/websocket_server/ws_channel.erl b/apps/efka/src/websocket/ws_channel.erl similarity index 67% rename from apps/efka/src/websocket_server/ws_channel.erl rename to apps/efka/src/websocket/ws_channel.erl index eeb2800..66fa33d 100644 --- a/apps/efka/src/websocket_server/ws_channel.erl +++ b/apps/efka/src/websocket/ws_channel.erl @@ -13,25 +13,20 @@ %% API -export([init/2]). -export([websocket_init/1, websocket_handle/2, websocket_info/2, terminate/3]). --export([push_config/4, invoke/4, channel_reply/3, register/2]). +-export([invoke/4, channel_reply/3]). %% 最大的等待时间 -define(PENDING_TIMEOUT, 10 * 1000). -record(state, { packet_id = 1, - service_id :: undefined | binary(), - service_pid :: undefined | pid(), + container_pid :: undefined | pid(), is_registered = false :: boolean(), %% 请求响应的对应关系, #{packet_id => {ReceiverPid, Ref}}; 自身的inflight需要超时逻辑处理 inflight = #{} }). --spec push_config(ChannelPid :: pid(), Ref :: reference(), ReceiverPid :: pid(), ConfigJson :: binary()) -> no_return(). -push_config(ChannelPid, Ref, ReceiverPid, ConfigJson) when is_pid(ChannelPid), is_pid(ReceiverPid), is_binary(ConfigJson), is_reference(Ref) -> - ChannelPid ! {push_config, Ref, ReceiverPid, ConfigJson}. - -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}. @@ -51,27 +46,6 @@ channel_reply(Id, Reply, Inflight) -> NInflight end. -%% 注册 --spec register(Id :: integer(), ServiceId :: binary()) -> {error, Reply :: binary()} | {ok, Reply :: binary(), ServicePid :: pid()}. -register(Id, ServiceId) when is_integer(Id), is_binary(ServiceId) -> - case efka_service:get_pid(ServiceId) of - undefined -> - lager:warning("[ws_channel] service_id: ~p, not running", [ServiceId]), - Reply = json_error(Id, -1, <<"service not running">>), - {error, Reply}; - ServicePid when is_pid(ServicePid) -> - case efka_service:attach_channel(ServicePid, self()) of - ok -> - Reply = json_result(Id, <<"ok">>), - erlang:monitor(process, ServicePid), - {ok, Reply, ServicePid}; - {error, Error} -> - lager:warning("[ws_channel] service_id: ~p, attach_channel get error: ~p", [ServiceId, Error]), - Reply = json_error(Id, -1, Error), - {error, Reply} - end - end. - %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% %% 逻辑处理方法 %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% @@ -103,14 +77,6 @@ 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 = <>, - 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({invoke, Ref, ReceiverPid, Payload}, State = #state{packet_id = PacketId, inflight = Inflight}) -> PushConfig = #{<<"id">> => PacketId, <<"method">> => <<"invoke">>, <<"params">> => #{<<"payload">> => Payload}}, @@ -132,9 +98,9 @@ websocket_info({timeout, _, {pending_timeout, Id}}, State = #state{inflight = In {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({'DOWN', _Ref, process, ContainerPid, Reason}, State = #state{container_pid = ContainerPid}) -> + lager:debug("[tcp_channel] container_pid: ~p, exited: ~p", [ContainerPid, Reason]), + {stop, State#state{container_pid = undefined}}; websocket_info({timeout, _, {stop, Reason}}, State) -> lager:debug("[ws_channel] the channel will be closed with reason: ~p", [Reason]), @@ -158,32 +124,38 @@ terminate(Reason, _Req, State) -> %% helper methods %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% -%% 注册 -handle_request(#{<<"id">> := Id, <<"method">> := <<"register">>, <<"params">> := #{<<"service_id">> := ServiceId}}, State) -> - case ws_channel:register(Id, ServiceId) of - {error, Reply} -> +%% 注册, 要建立程序和容器之间的关系 +handle_request(#{<<"id">> := Id, <<"method">> := <<"register">>, <<"params">> := #{<<"container_name">> := ContainerName, <<"meta_tag">> := MetaTag}}, State) -> + case efka_container:get_pid(ContainerName) 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}; - {ok, Reply, ServicePid} -> - delay_stop(10, normal), - {reply, {binary, <>}, State#state{service_id = ServiceId, service_pid = ServicePid, is_registered = true}} + {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), + + {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 end; -%% 请求参数 -handle_request(#{<<"id">> := Id, <<"method">> := <<"request_config">>}, State = #state{service_pid = ServicePid, is_registered = true}) -> - {ok, ConfigJson} = efka_service:request_config(ServicePid), - Packet = ws_channel:json_result(Id, ConfigJson), - Reply = <>, - {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), + <<"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), {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), +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), {ok, State}; %% 订阅事件 diff --git a/apps/efka/src/websocket_server/ws_server.erl b/apps/efka/src/websocket/ws_server.erl similarity index 100% rename from apps/efka/src/websocket_server/ws_server.erl rename to apps/efka/src/websocket/ws_server.erl