fix channel

2025-08-27 16:10:38 +08:00 · 2025-08-27 16:10:38 +08:00 · b1137bc1ad
commit b1137bc1ad
parent 63add6a2e6
3 changed files with 49 additions and 107 deletions
--- 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 ->
--- 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, <<?PACKET_REQUEST:8, Data/binary>>}, 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, <<?PACKET_RESPONSE:8, Data/binary>>}, State = #state{socket = Socket, inflight = Inflight}) ->
@ -125,34 +120,14 @@ handle_info({tcp, Socket, <<?PACKET_RESPONSE:8, Data/binary>>}, 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, <<?PACKET_RESPONSE:8, Packet/binary>>),
+    case gen_channel:register(Id, ServiceId) of
+        {error, Reply} ->
+            ok = gen_tcp:send(Socket, <<?PACKET_RESPONSE:8, Reply/binary>>),
            {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, <<?PACKET_RESPONSE:8, Packet/binary>>),
-                    {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, <<?PACKET_RESPONSE:8, Packet/binary>>),
-                    {stop, normal, State}
-            end
+        {ok, Reply, ServicePid} ->
+            ok = gen_tcp:send(Socket, <<?PACKET_RESPONSE:8, Reply/binary>>),
+            {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, <<?PACKET_RESPONSE:8, Packet/binary>>),
-    {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}.
+    {noreply, State}.
--- 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, <<?PACKET_REQUEST, Data/binary>>}, State) ->

 %% 处理micro-client:response => efka 的响应
 websocket_handle({binary, <<?PACKET_RESPONSE:8, Data/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 = <<?PACKET_RESPONSE:8, Packet/binary>>,
+    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 = <<?PACKET_RESPONSE:8, Packet/binary>>,
-
-                    {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 = <<?PACKET_RESPONSE:8, Packet/binary>>,
-                    delay_stop(10, normal),
-                    {reply, {binary, Reply}, State}
-            end
+            {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),