From b1137bc1ad06665d380c12c51bf5caac5fd09605 Mon Sep 17 00:00:00 2001 From: anlicheng <244108715@qq.com> Date: Wed, 27 Aug 2025 16:10:38 +0800 Subject: [PATCH] fix channel --- apps/efka/src/gen_channel.erl | 22 ++++++ apps/efka/src/tcp_server/tcp_channel.erl | 74 ++++--------------- apps/efka/src/websocket_server/ws_channel.erl | 60 +++------------ 3 files changed, 49 insertions(+), 107 deletions(-) diff --git a/apps/efka/src/gen_channel.erl b/apps/efka/src/gen_channel.erl index 5071f1a..91adb55 100644 --- a/apps/efka/src/gen_channel.erl +++ b/apps/efka/src/gen_channel.erl @@ -10,6 +10,7 @@ -author("anlicheng"). -include("efka_service.hrl"). +-export([register/2]). -export([push_config/4, invoke/4, channel_reply/3]). -export([next_packet_id/1]). -export([json_result/2, json_error/3]). @@ -41,6 +42,27 @@ 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("[gen_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("[gen_channel] service_id: ~p, attach_channel get error: ~p", [ServiceId, Error]), + Reply = json_error(Id, -1, Error), + {error, Reply} + end + end. + %% 采用32位编码 -spec next_packet_id(PacketId :: integer()) -> NextPacketId :: integer(). next_packet_id(PacketId) when PacketId >= 4294967295 -> diff --git a/apps/efka/src/tcp_server/tcp_channel.erl b/apps/efka/src/tcp_server/tcp_channel.erl index f37ce4c..519d949 100644 --- a/apps/efka/src/tcp_server/tcp_channel.erl +++ b/apps/efka/src/tcp_server/tcp_channel.erl @@ -109,12 +109,7 @@ handle_info({invoke, Ref, ReceiverPid, Payload}, State = #state{socket = Socket, %% 处理micro-client:request => efka 主动的请求 handle_info({tcp, Socket, <>}, State = #state{socket = Socket}) -> Request = jiffy:decode(Data, [return_maps]), - case handle_request(Request, State) of - {ok, NewState} -> - {noreply, NewState}; - {stop, Reason, NewState} -> - {stop, Reason, NewState} - end; + handle_request(Request, State); %% 处理micro-client:response => efka 的响应 handle_info({tcp, Socket, <>}, State = #state{socket = Socket, inflight = Inflight}) -> @@ -125,34 +120,14 @@ handle_info({tcp, Socket, <>}, State = #state{s #{<<"id">> := Id, <<"error">> := #{<<"code">> := _Code, <<"message">> := Error}} -> {Id, {error, Error}} end, - case maps:take(PacketId, Inflight) of - error -> - lager:warning("[tcp_channel] get unknown publish response message: ~p, packet_id: ~p", [Resp, PacketId]), - {noreply, State}; - {{ReceiverPid, Ref}, NInflight} -> - case is_pid(ReceiverPid) andalso is_process_alive(ReceiverPid) of - true -> - ReceiverPid ! {channel_reply, Ref, Reply}; - false -> - lager:warning("[tcp_channel] get publish response message: ~p, packet_id: ~p, but receiver_pid is deaded", [Resp, PacketId]) - end, - {noreply, State#state{inflight = NInflight}} - end; + + NInflight = gen_channel:channel_reply(PacketId, Reply, Inflight), + {noreply, State#state{inflight = NInflight}}; %% 超时逻辑处理 handle_info({timeout, _, {pending_timeout, Id}}, State = #state{inflight = Inflight}) -> - case maps:take(Id, Inflight) of - error -> - {noreply, State}; - {{ReceiverPid, Ref}, NInflight} -> - case is_pid(ReceiverPid) andalso is_process_alive(ReceiverPid) of - true -> - ReceiverPid ! {channel_reply, Ref, {error, <<"timeout">>}}; - false -> - ok - end, - {noreply, State#state{inflight = NInflight}} - end; + NInflight = gen_channel:channel_reply(Id, {error, <<"timeout">>}, Inflight), + {noreply, State#state{inflight = NInflight}}; %% 订阅的消息 handle_info({topic_broadcast, Topic, Content}, State = #state{socket = Socket}) -> @@ -197,47 +172,30 @@ code_change(_OldVsn, State = #state{}, _Extra) -> %% 注册 handle_request(#{<<"id">> := Id, <<"method">> := <<"register">>, <<"params">> := #{<<"service_id">> := ServiceId}}, State = #state{socket = Socket}) -> - case efka_service:get_pid(ServiceId) of - undefined -> - lager:warning("[efka_tcp_channel] service_id: ~p, not running", [ServiceId]), - Packet = gen_channel:json_error(Id, -1, <<"service not running">>), - ok = gen_tcp:send(Socket, <>), + case gen_channel:register(Id, ServiceId) of + {error, Reply} -> + ok = gen_tcp:send(Socket, <>), {stop, normal, State}; - ServicePid when is_pid(ServicePid) -> - case efka_service:attach_channel(ServicePid, self()) of - ok -> - Packet = gen_channel:json_result(Id, <<"ok">>), - erlang:monitor(process, ServicePid), - - ok = gen_tcp:send(Socket, <>), - {ok, State#state{service_id = ServiceId, service_pid = ServicePid, is_registered = true}}; - {error, Error} -> - lager:warning("[efka_tcp_channel] service_id: ~p, attach_channel get error: ~p", [ServiceId, Error]), - Packet = gen_channel:json_error(Id, -1, Error), - ok = gen_tcp:send(Socket, <>), - {stop, normal, State} - end + {ok, Reply, ServicePid} -> + ok = gen_tcp:send(Socket, <>), + {noreply, State#state{service_id = ServiceId, service_pid = ServicePid, is_registered = true}} end; - %% 请求参数 handle_request(#{<<"id">> := Id, <<"method">> := <<"request_config">>}, State = #state{socket = Socket, service_pid = ServicePid, is_registered = true}) -> {ok, ConfigJson} = efka_service:request_config(ServicePid), Packet = gen_channel:json_result(Id, ConfigJson), ok = gen_tcp:send(Socket, <>), - {ok, State}; - + {noreply, 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}; - + {noreply, 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}; - + {noreply, State}; %% 订阅事件 handle_request(#{<<"id">> := 0, <<"method">> := <<"subscribe">>, <<"params">> := #{<<"topic">> := Topic}}, State = #state{is_registered = true}) -> efka_subscription:subscribe(Topic, self()), - {ok, State}. \ No newline at end of file + {noreply, State}. \ No newline at end of file diff --git a/apps/efka/src/websocket_server/ws_channel.erl b/apps/efka/src/websocket_server/ws_channel.erl index 2c25691..5b32a4e 100644 --- a/apps/efka/src/websocket_server/ws_channel.erl +++ b/apps/efka/src/websocket_server/ws_channel.erl @@ -45,26 +45,14 @@ websocket_handle({binary, <>}, State) -> %% 处理micro-client:response => efka 的响应 websocket_handle({binary, <>}, State = #state{inflight = Inflight}) -> - Resp = jiffy:decode(Data, [return_maps]), - {PacketId, Reply} = case Resp of + {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, - case maps:take(PacketId, Inflight) of - error -> - lager:warning("[tcp_channel] get unknown publish response message: ~p, packet_id: ~p", [Resp, PacketId]), - {ok, State}; - {{ReceiverPid, Ref}, NInflight} -> - case is_pid(ReceiverPid) andalso is_process_alive(ReceiverPid) of - true -> - ReceiverPid ! {channel_reply, Ref, Reply}; - false -> - lager:warning("[tcp_channel] get publish response message: ~p, packet_id: ~p, but receiver_pid is deaded", [Resp, PacketId]) - end, - {ok, State#state{inflight = NInflight}} - 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]), @@ -95,18 +83,8 @@ websocket_info({topic_broadcast, Topic, Content}, State = #state{}) -> %% 超时逻辑处理 websocket_info({timeout, _, {pending_timeout, Id}}, State = #state{inflight = Inflight}) -> - case maps:take(Id, Inflight) of - error -> - {ok, State}; - {{ReceiverPid, Ref}, NInflight} -> - case is_pid(ReceiverPid) andalso is_process_alive(ReceiverPid) of - true -> - ReceiverPid ! {channel_reply, Ref, {error, <<"timeout">>}}; - false -> - ok - end, - {ok, State#state{inflight = NInflight}} - end; + 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}) -> @@ -137,30 +115,14 @@ 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("[efka_tcp_channel] service_id: ~p, not running", [ServiceId]), - Packet = gen_channel:json_error(Id, -1, <<"service not running">>), - Reply = <>, + case gen_channel:register(Id, ServiceId) of + {error, Reply} -> delay_stop(10, normal), - {reply, {binary, Reply}, State}; - ServicePid when is_pid(ServicePid) -> - case efka_service:attach_channel(ServicePid, self()) of - ok -> - Packet = gen_channel:json_result(Id, <<"ok">>), - erlang:monitor(process, ServicePid), - Reply = <>, - - {reply, {binary, Reply}, State#state{service_id = ServiceId, service_pid = ServicePid, is_registered = true}}; - {error, Error} -> - lager:warning("[efka_tcp_channel] service_id: ~p, attach_channel get error: ~p", [ServiceId, Error]), - Packet = gen_channel:json_error(Id, -1, Error), - Reply = <>, - delay_stop(10, normal), - {reply, {binary, Reply}, State} - end + {reply, {binary, <>}, State}; + {ok, Reply, ServicePid} -> + delay_stop(10, normal), + {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),