From 8e4d66e9d88216d5a5f3d322258eb8f4d8b7e9dd Mon Sep 17 00:00:00 2001 From: anlicheng <244108715@qq.com> Date: Wed, 17 Sep 2025 12:02:03 +0800 Subject: [PATCH] fxi --- apps/efka/include/message_pb.hrl | 9 -- apps/efka/src/efka_remote_agent.erl | 34 +----- apps/efka/src/proto/message_pb.erl | 151 +------------------------ apps/efka/src/websocket/ws_channel.erl | 16 +-- message_pb.proto | 7 -- 5 files changed, 11 insertions(+), 206 deletions(-) diff --git a/apps/efka/include/message_pb.hrl b/apps/efka/include/message_pb.hrl index e084d3c..bb8d00e 100644 --- a/apps/efka/include/message_pb.hrl +++ b/apps/efka/include/message_pb.hrl @@ -59,15 +59,6 @@ }). -endif. --ifndef('INVOKE_PB_H'). --define('INVOKE_PB_H', true). --record(invoke, - {service_id = <<>> :: unicode:chardata() | undefined, % = 1, optional - payload = <<>> :: iodata() | undefined, % = 2, optional - timeout = 0 :: non_neg_integer() | undefined % = 3, optional, 32 bits - }). --endif. - -ifndef('SERVICE_CONFIG_PB_H'). -define('SERVICE_CONFIG_PB_H', true). -record(service_config, diff --git a/apps/efka/src/efka_remote_agent.erl b/apps/efka/src/efka_remote_agent.erl index f4875be..bae6601 100644 --- a/apps/efka/src/efka_remote_agent.erl +++ b/apps/efka/src/efka_remote_agent.erl @@ -16,7 +16,7 @@ %% API -export([start_link/0]). --export([metric_data/4, event/3, ping/13, request_service_config/2, await_reply/2]). +-export([metric_data/4, event/3, ping/13, await_reply/2]). %% gen_statem callbacks -export([init/1, handle_event/4, terminate/3, code_change/4, callback_mode/0]). @@ -57,11 +57,6 @@ event(ServiceId, EventType, Params) when is_binary(ServiceId), is_integer(EventT ping(AdCode, BootTime, Province, City, EfkaVersion, KernelArch, Ips, CpuCore, CpuLoad, CpuTemperature, Disk, Memory, Interfaces) -> gen_statem:cast(?SERVER, {ping, AdCode, BootTime, Province, City, EfkaVersion, KernelArch, Ips, CpuCore, CpuLoad, CpuTemperature, Disk, Memory, Interfaces}). -%% 请求微服务的配置 --spec request_service_config(ReceiverPid :: pid(), ServiceId :: binary()) -> {ok, Ref :: reference()} | {error, Reason :: term()}. -request_service_config(ReceiverPid, ServiceId) when is_binary(ServiceId) -> - gen_statem:call(?SERVER, {request_service_config, ReceiverPid, ServiceId}). - %% 等待消息的回复 -spec await_reply(Ref :: reference(), Timeout :: timeout()) -> {ok, Reply :: binary()} | {error, timeout}. await_reply(Ref, Timeout) when is_reference(Ref), is_integer(Timeout) -> @@ -100,12 +95,6 @@ callback_mode() -> %% @doc If callback_mode is handle_event_function, then whenever a %% gen_statem receives an event from call/2, cast/2, or as a normal %% process message, this function is called. -handle_event({call, From}, {request_service_config, ReceiverPid, ServiceId}, ?STATE_ACTIVATED, State = #state{transport_pid = TransportPid, request_inflight = RequestInflight}) -> - Ref = efka_transport:request(TransportPid, ?METHOD_REQUEST_SERVICE_CONFIG, ServiceId), - {keep_state, State#state{request_inflight = maps:put(Ref, ReceiverPid, RequestInflight)}, [{reply, From, {ok, Ref}}]}; - -handle_event({call, From}, {request_service_config, _ReceiverPid, _ServiceId}, _, State) -> - {keep_state, State, [{reply, From, {error, <<"transport is not alive">>}}]}; %% 异步发送数据, 连接存在时候直接发送；否则缓存到mnesia handle_event(cast, {metric_data, ServiceId, DeviceUUID, RouteKey, Metric}, ?STATE_ACTIVATED, State = #state{transport_pid = TransportPid}) -> @@ -280,7 +269,7 @@ handle_event(info, {server_async_call, PacketId, <>}, ?STATE_ACTIVATED, State = #state{transport_pid = TransportPid, push_inflight = PushInflight}) -> #service_config{service_id = ServiceId, config_json = ConfigJson, timeout = Timeout} = message_pb:decode_msg(ConfigBin, push_service_config), @@ -299,25 +288,6 @@ handle_event(info, {server_async_call, PacketId, <>}, ?STATE_ACTIVATED, State = #state{push_inflight = PushInflight, transport_pid = TransportPid}) -> - #invoke{service_id = ServiceId, payload = Payload, timeout = Timeout} = message_pb:decode_msg(InvokeBin, invoke), - %% 消息发送到订阅系统 - case efka_service:get_pid(ServiceId) of - undefined -> - Reply = #async_call_reply{code = 0, message = <<"micro_service not run">>}, - efka_transport:async_call_reply(TransportPid, PacketId, message_pb:encode_msg(Reply)), - - {keep_state, State}; - ServicePid when is_pid(ServicePid) -> - Ref = make_ref(), - efka_service:invoke(ServicePid, Ref, Payload), - %% 处理超时逻辑 - erlang:start_timer(Timeout, self(), {request_timeout, Ref}), - - {keep_state, State#state{push_inflight = maps:put(Ref, PacketId, PushInflight)}} - end; - %% 处理task_log handle_event(info, {server_async_call, PacketId, <>}, ?STATE_ACTIVATED, State = #state{transport_pid = TransportPid}) -> #fetch_task_log{task_id = TaskId} = message_pb:decode_msg(TaskLogBin, fetch_task_log), diff --git a/apps/efka/src/proto/message_pb.erl b/apps/efka/src/proto/message_pb.erl index f24fabc..883a085 100644 --- a/apps/efka/src/proto/message_pb.erl +++ b/apps/efka/src/proto/message_pb.erl @@ -67,8 +67,6 @@ -type fetch_task_log() :: #fetch_task_log{}. --type invoke() :: #invoke{}. - -type service_config() :: #service_config{}. -type data() :: #data{}. @@ -77,9 +75,9 @@ -type ping() :: #ping{}. --export_type(['auth_request'/0, 'auth_reply'/0, 'pub'/0, 'async_call_reply'/0, 'deploy'/0, 'fetch_task_log'/0, 'invoke'/0, 'service_config'/0, 'data'/0, 'event'/0, 'ping'/0]). --type '$msg_name'() :: auth_request | auth_reply | pub | async_call_reply | deploy | fetch_task_log | invoke | service_config | data | event | ping. --type '$msg'() :: auth_request() | auth_reply() | pub() | async_call_reply() | deploy() | fetch_task_log() | invoke() | service_config() | data() | event() | ping(). +-export_type(['auth_request'/0, 'auth_reply'/0, 'pub'/0, 'async_call_reply'/0, 'deploy'/0, 'fetch_task_log'/0, 'service_config'/0, 'data'/0, 'event'/0, 'ping'/0]). +-type '$msg_name'() :: auth_request | auth_reply | pub | async_call_reply | deploy | fetch_task_log | service_config | data | event | ping. +-type '$msg'() :: auth_request() | auth_reply() | pub() | async_call_reply() | deploy() | fetch_task_log() | service_config() | data() | event() | ping(). -export_type(['$msg_name'/0, '$msg'/0]). -if(?OTP_RELEASE >= 24). @@ -112,7 +110,6 @@ encode_msg(Msg, MsgName, Opts) -> async_call_reply -> encode_msg_async_call_reply(id(Msg, TrUserData), TrUserData); deploy -> encode_msg_deploy(id(Msg, TrUserData), TrUserData); fetch_task_log -> encode_msg_fetch_task_log(id(Msg, TrUserData), TrUserData); - invoke -> encode_msg_invoke(id(Msg, TrUserData), TrUserData); service_config -> encode_msg_service_config(id(Msg, TrUserData), TrUserData); data -> encode_msg_data(id(Msg, TrUserData), TrUserData); event -> encode_msg_event(id(Msg, TrUserData), TrUserData); @@ -295,40 +292,6 @@ encode_msg_fetch_task_log(#fetch_task_log{task_id = F1}, Bin, TrUserData) -> end end. -encode_msg_invoke(Msg, TrUserData) -> encode_msg_invoke(Msg, <<>>, TrUserData). - - -encode_msg_invoke(#invoke{service_id = F1, payload = F2, timeout = F3}, Bin, TrUserData) -> - B1 = if F1 == undefined -> Bin; - true -> - begin - TrF1 = id(F1, TrUserData), - case is_empty_string(TrF1) of - true -> Bin; - false -> e_type_string(TrF1, <>, TrUserData) - end - end - end, - B2 = if F2 == undefined -> B1; - true -> - begin - TrF2 = id(F2, TrUserData), - case iolist_size(TrF2) of - 0 -> B1; - _ -> e_type_bytes(TrF2, <>, TrUserData) - end - end - end, - if F3 == undefined -> B2; - true -> - begin - TrF3 = id(F3, TrUserData), - if TrF3 =:= 0 -> B2; - true -> e_varint(TrF3, <>, TrUserData) - end - end - end. - encode_msg_service_config(Msg, TrUserData) -> encode_msg_service_config(Msg, <<>>, TrUserData). @@ -735,7 +698,6 @@ decode_msg_2_doit(pub, Bin, TrUserData) -> id(decode_msg_pub(Bin, TrUserData), T decode_msg_2_doit(async_call_reply, Bin, TrUserData) -> id(decode_msg_async_call_reply(Bin, TrUserData), TrUserData); decode_msg_2_doit(deploy, Bin, TrUserData) -> id(decode_msg_deploy(Bin, TrUserData), TrUserData); decode_msg_2_doit(fetch_task_log, Bin, TrUserData) -> id(decode_msg_fetch_task_log(Bin, TrUserData), TrUserData); -decode_msg_2_doit(invoke, Bin, TrUserData) -> id(decode_msg_invoke(Bin, TrUserData), TrUserData); decode_msg_2_doit(service_config, Bin, TrUserData) -> id(decode_msg_service_config(Bin, TrUserData), TrUserData); decode_msg_2_doit(data, Bin, TrUserData) -> id(decode_msg_data(Bin, TrUserData), TrUserData); decode_msg_2_doit(event, Bin, TrUserData) -> id(decode_msg_event(Bin, TrUserData), TrUserData); @@ -1070,64 +1032,6 @@ skip_32_fetch_task_log(<<_:32, Rest/binary>>, Z1, Z2, F, F@_1, TrUserData) -> df skip_64_fetch_task_log(<<_:64, Rest/binary>>, Z1, Z2, F, F@_1, TrUserData) -> dfp_read_field_def_fetch_task_log(Rest, Z1, Z2, F, F@_1, TrUserData). -decode_msg_invoke(Bin, TrUserData) -> dfp_read_field_def_invoke(Bin, 0, 0, 0, id(<<>>, TrUserData), id(<<>>, TrUserData), id(0, TrUserData), TrUserData). - -dfp_read_field_def_invoke(<<10, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData) -> d_field_invoke_service_id(Rest, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData); -dfp_read_field_def_invoke(<<18, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData) -> d_field_invoke_payload(Rest, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData); -dfp_read_field_def_invoke(<<24, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData) -> d_field_invoke_timeout(Rest, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData); -dfp_read_field_def_invoke(<<>>, 0, 0, _, F@_1, F@_2, F@_3, _) -> #invoke{service_id = F@_1, payload = F@_2, timeout = F@_3}; -dfp_read_field_def_invoke(Other, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData) -> dg_read_field_def_invoke(Other, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData). - -dg_read_field_def_invoke(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, F@_3, TrUserData) when N < 32 - 7 -> dg_read_field_def_invoke(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, F@_3, TrUserData); -dg_read_field_def_invoke(<<0:1, X:7, Rest/binary>>, N, Acc, _, F@_1, F@_2, F@_3, TrUserData) -> - Key = X bsl N + Acc, - case Key of - 10 -> d_field_invoke_service_id(Rest, 0, 0, 0, F@_1, F@_2, F@_3, TrUserData); - 18 -> d_field_invoke_payload(Rest, 0, 0, 0, F@_1, F@_2, F@_3, TrUserData); - 24 -> d_field_invoke_timeout(Rest, 0, 0, 0, F@_1, F@_2, F@_3, TrUserData); - _ -> - case Key band 7 of - 0 -> skip_varint_invoke(Rest, 0, 0, Key bsr 3, F@_1, F@_2, F@_3, TrUserData); - 1 -> skip_64_invoke(Rest, 0, 0, Key bsr 3, F@_1, F@_2, F@_3, TrUserData); - 2 -> skip_length_delimited_invoke(Rest, 0, 0, Key bsr 3, F@_1, F@_2, F@_3, TrUserData); - 3 -> skip_group_invoke(Rest, 0, 0, Key bsr 3, F@_1, F@_2, F@_3, TrUserData); - 5 -> skip_32_invoke(Rest, 0, 0, Key bsr 3, F@_1, F@_2, F@_3, TrUserData) - end - end; -dg_read_field_def_invoke(<<>>, 0, 0, _, F@_1, F@_2, F@_3, _) -> #invoke{service_id = F@_1, payload = F@_2, timeout = F@_3}. - -d_field_invoke_service_id(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, F@_3, TrUserData) when N < 57 -> d_field_invoke_service_id(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, F@_3, TrUserData); -d_field_invoke_service_id(<<0:1, X:7, Rest/binary>>, N, Acc, F, _, F@_2, F@_3, TrUserData) -> - {NewFValue, RestF} = begin Len = X bsl N + Acc, <> = Rest, Bytes2 = binary:copy(Bytes), {id(Bytes2, TrUserData), Rest2} end, - dfp_read_field_def_invoke(RestF, 0, 0, F, NewFValue, F@_2, F@_3, TrUserData). - -d_field_invoke_payload(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, F@_3, TrUserData) when N < 57 -> d_field_invoke_payload(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, F@_3, TrUserData); -d_field_invoke_payload(<<0:1, X:7, Rest/binary>>, N, Acc, F, F@_1, _, F@_3, TrUserData) -> - {NewFValue, RestF} = begin Len = X bsl N + Acc, <> = Rest, Bytes2 = binary:copy(Bytes), {id(Bytes2, TrUserData), Rest2} end, - dfp_read_field_def_invoke(RestF, 0, 0, F, F@_1, NewFValue, F@_3, TrUserData). - -d_field_invoke_timeout(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, F@_3, TrUserData) when N < 57 -> d_field_invoke_timeout(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, F@_3, TrUserData); -d_field_invoke_timeout(<<0:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, _, TrUserData) -> - {NewFValue, RestF} = {id((X bsl N + Acc) band 4294967295, TrUserData), Rest}, - dfp_read_field_def_invoke(RestF, 0, 0, F, F@_1, F@_2, NewFValue, TrUserData). - -skip_varint_invoke(<<1:1, _:7, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData) -> skip_varint_invoke(Rest, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData); -skip_varint_invoke(<<0:1, _:7, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData) -> dfp_read_field_def_invoke(Rest, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData). - -skip_length_delimited_invoke(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, F@_3, TrUserData) when N < 57 -> skip_length_delimited_invoke(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, F@_3, TrUserData); -skip_length_delimited_invoke(<<0:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, F@_3, TrUserData) -> - Length = X bsl N + Acc, - <<_:Length/binary, Rest2/binary>> = Rest, - dfp_read_field_def_invoke(Rest2, 0, 0, F, F@_1, F@_2, F@_3, TrUserData). - -skip_group_invoke(Bin, _, Z2, FNum, F@_1, F@_2, F@_3, TrUserData) -> - {_, Rest} = read_group(Bin, FNum), - dfp_read_field_def_invoke(Rest, 0, Z2, FNum, F@_1, F@_2, F@_3, TrUserData). - -skip_32_invoke(<<_:32, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData) -> dfp_read_field_def_invoke(Rest, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData). - -skip_64_invoke(<<_:64, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData) -> dfp_read_field_def_invoke(Rest, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData). - decode_msg_service_config(Bin, TrUserData) -> dfp_read_field_def_service_config(Bin, 0, 0, 0, id(<<>>, TrUserData), id(<<>>, TrUserData), id(0, TrUserData), TrUserData). dfp_read_field_def_service_config(<<10, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData) -> d_field_service_config_service_id(Rest, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData); @@ -1601,7 +1505,6 @@ merge_msgs(Prev, New, MsgName, Opts) -> async_call_reply -> merge_msg_async_call_reply(Prev, New, TrUserData); deploy -> merge_msg_deploy(Prev, New, TrUserData); fetch_task_log -> merge_msg_fetch_task_log(Prev, New, TrUserData); - invoke -> merge_msg_invoke(Prev, New, TrUserData); service_config -> merge_msg_service_config(Prev, New, TrUserData); data -> merge_msg_data(Prev, New, TrUserData); event -> merge_msg_event(Prev, New, TrUserData); @@ -1686,21 +1589,6 @@ merge_msg_fetch_task_log(#fetch_task_log{task_id = PFtask_id}, #fetch_task_log{t true -> NFtask_id end}. --compile({nowarn_unused_function,merge_msg_invoke/3}). -merge_msg_invoke(#invoke{service_id = PFservice_id, payload = PFpayload, timeout = PFtimeout}, #invoke{service_id = NFservice_id, payload = NFpayload, timeout = NFtimeout}, _) -> - #invoke{service_id = - if NFservice_id =:= undefined -> PFservice_id; - true -> NFservice_id - end, - payload = - if NFpayload =:= undefined -> PFpayload; - true -> NFpayload - end, - timeout = - if NFtimeout =:= undefined -> PFtimeout; - true -> NFtimeout - end}. - -compile({nowarn_unused_function,merge_msg_service_config/3}). merge_msg_service_config(#service_config{service_id = PFservice_id, config_json = PFconfig_json, timeout = PFtimeout}, #service_config{service_id = NFservice_id, config_json = NFconfig_json, timeout = NFtimeout}, _) -> #service_config{service_id = @@ -1829,7 +1717,6 @@ verify_msg(Msg, MsgName, Opts) -> async_call_reply -> v_msg_async_call_reply(Msg, [MsgName], TrUserData); deploy -> v_msg_deploy(Msg, [MsgName], TrUserData); fetch_task_log -> v_msg_fetch_task_log(Msg, [MsgName], TrUserData); - invoke -> v_msg_invoke(Msg, [MsgName], TrUserData); service_config -> v_msg_service_config(Msg, [MsgName], TrUserData); data -> v_msg_data(Msg, [MsgName], TrUserData); event -> v_msg_event(Msg, [MsgName], TrUserData); @@ -1919,21 +1806,6 @@ v_msg_fetch_task_log(#fetch_task_log{task_id = F1}, Path, TrUserData) -> ok; v_msg_fetch_task_log(X, Path, _TrUserData) -> mk_type_error({expected_msg, fetch_task_log}, X, Path). --compile({nowarn_unused_function,v_msg_invoke/3}). --dialyzer({nowarn_function,v_msg_invoke/3}). -v_msg_invoke(#invoke{service_id = F1, payload = F2, timeout = F3}, Path, TrUserData) -> - if F1 == undefined -> ok; - true -> v_type_string(F1, [service_id | Path], TrUserData) - end, - if F2 == undefined -> ok; - true -> v_type_bytes(F2, [payload | Path], TrUserData) - end, - if F3 == undefined -> ok; - true -> v_type_uint32(F3, [timeout | Path], TrUserData) - end, - ok; -v_msg_invoke(X, Path, _TrUserData) -> mk_type_error({expected_msg, invoke}, X, Path). - -compile({nowarn_unused_function,v_msg_service_config/3}). -dialyzer({nowarn_function,v_msg_service_config/3}). v_msg_service_config(#service_config{service_id = F1, config_json = F2, timeout = F3}, Path, TrUserData) -> @@ -2123,10 +1995,6 @@ get_msg_defs() -> #field{name = message, fnum = 3, rnum = 4, type = string, occurrence = optional, opts = []}]}, {{msg, deploy}, [#field{name = task_id, fnum = 1, rnum = 2, type = uint32, occurrence = optional, opts = []}, #field{name = config, fnum = 2, rnum = 3, type = string, occurrence = optional, opts = []}]}, {{msg, fetch_task_log}, [#field{name = task_id, fnum = 1, rnum = 2, type = uint32, occurrence = optional, opts = []}]}, - {{msg, invoke}, - [#field{name = service_id, fnum = 1, rnum = 2, type = string, occurrence = optional, opts = []}, - #field{name = payload, fnum = 2, rnum = 3, type = bytes, occurrence = optional, opts = []}, - #field{name = timeout, fnum = 3, rnum = 4, type = uint32, occurrence = optional, opts = []}]}, {{msg, service_config}, [#field{name = service_id, fnum = 1, rnum = 2, type = string, occurrence = optional, opts = []}, #field{name = config_json, fnum = 2, rnum = 3, type = string, occurrence = optional, opts = []}, @@ -2156,13 +2024,13 @@ get_msg_defs() -> #field{name = interfaces, fnum = 13, rnum = 14, type = string, occurrence = optional, opts = []}]}]. -get_msg_names() -> [auth_request, auth_reply, pub, async_call_reply, deploy, fetch_task_log, invoke, service_config, data, event, ping]. +get_msg_names() -> [auth_request, auth_reply, pub, async_call_reply, deploy, fetch_task_log, service_config, data, event, ping]. get_group_names() -> []. -get_msg_or_group_names() -> [auth_request, auth_reply, pub, async_call_reply, deploy, fetch_task_log, invoke, service_config, data, event, ping]. +get_msg_or_group_names() -> [auth_request, auth_reply, pub, async_call_reply, deploy, fetch_task_log, service_config, data, event, ping]. get_enum_names() -> []. @@ -2193,10 +2061,6 @@ find_msg_def(async_call_reply) -> #field{name = message, fnum = 3, rnum = 4, type = string, occurrence = optional, opts = []}]; find_msg_def(deploy) -> [#field{name = task_id, fnum = 1, rnum = 2, type = uint32, occurrence = optional, opts = []}, #field{name = config, fnum = 2, rnum = 3, type = string, occurrence = optional, opts = []}]; find_msg_def(fetch_task_log) -> [#field{name = task_id, fnum = 1, rnum = 2, type = uint32, occurrence = optional, opts = []}]; -find_msg_def(invoke) -> - [#field{name = service_id, fnum = 1, rnum = 2, type = string, occurrence = optional, opts = []}, - #field{name = payload, fnum = 2, rnum = 3, type = bytes, occurrence = optional, opts = []}, - #field{name = timeout, fnum = 3, rnum = 4, type = uint32, occurrence = optional, opts = []}]; find_msg_def(service_config) -> [#field{name = service_id, fnum = 1, rnum = 2, type = string, occurrence = optional, opts = []}, #field{name = config_json, fnum = 2, rnum = 3, type = string, occurrence = optional, opts = []}, @@ -2288,7 +2152,6 @@ fqbin_to_msg_name(<<"Pub">>) -> pub; fqbin_to_msg_name(<<"AsyncCallReply">>) -> async_call_reply; fqbin_to_msg_name(<<"Deploy">>) -> deploy; fqbin_to_msg_name(<<"FetchTaskLog">>) -> fetch_task_log; -fqbin_to_msg_name(<<"Invoke">>) -> invoke; fqbin_to_msg_name(<<"ServiceConfig">>) -> service_config; fqbin_to_msg_name(<<"Data">>) -> data; fqbin_to_msg_name(<<"Event">>) -> event; @@ -2302,7 +2165,6 @@ msg_name_to_fqbin(pub) -> <<"Pub">>; msg_name_to_fqbin(async_call_reply) -> <<"AsyncCallReply">>; msg_name_to_fqbin(deploy) -> <<"Deploy">>; msg_name_to_fqbin(fetch_task_log) -> <<"FetchTaskLog">>; -msg_name_to_fqbin(invoke) -> <<"Invoke">>; msg_name_to_fqbin(service_config) -> <<"ServiceConfig">>; msg_name_to_fqbin(data) -> <<"Data">>; msg_name_to_fqbin(event) -> <<"Event">>; @@ -2345,7 +2207,7 @@ get_all_source_basenames() -> ["message_pb.proto"]. get_all_proto_names() -> ["message_pb"]. -get_msg_containment("message_pb") -> [async_call_reply, auth_reply, auth_request, data, deploy, event, fetch_task_log, invoke, ping, pub, service_config]; +get_msg_containment("message_pb") -> [async_call_reply, auth_reply, auth_request, data, deploy, event, fetch_task_log, ping, pub, service_config]; get_msg_containment(P) -> error({gpb_error, {badproto, P}}). @@ -2369,7 +2231,6 @@ get_proto_by_msg_name_as_fqbin(<<"Data">>) -> "message_pb"; get_proto_by_msg_name_as_fqbin(<<"Pub">>) -> "message_pb"; get_proto_by_msg_name_as_fqbin(<<"Event">>) -> "message_pb"; get_proto_by_msg_name_as_fqbin(<<"AuthRequest">>) -> "message_pb"; -get_proto_by_msg_name_as_fqbin(<<"Invoke">>) -> "message_pb"; get_proto_by_msg_name_as_fqbin(<<"ServiceConfig">>) -> "message_pb"; get_proto_by_msg_name_as_fqbin(<<"Ping">>) -> "message_pb"; get_proto_by_msg_name_as_fqbin(<<"FetchTaskLog">>) -> "message_pb"; diff --git a/apps/efka/src/websocket/ws_channel.erl b/apps/efka/src/websocket/ws_channel.erl index 54bbd96..aaed115 100644 --- a/apps/efka/src/websocket/ws_channel.erl +++ b/apps/efka/src/websocket/ws_channel.erl @@ -101,9 +101,10 @@ handle_request(#{<<"id">> := 0, <<"method">> := <<"event">>, <<"params">> := #{< {ok, State}; %% 订阅事件 -handle_request(#{<<"id">> := 0, <<"method">> := <<"subscribe">>, <<"params">> := #{<<"topic">> := Topic}}, State = #state{is_registered = true}) -> +handle_request(#{<<"id">> := Id, <<"method">> := <<"subscribe">>, <<"params">> := #{<<"topic">> := Topic}}, State = #state{is_registered = true}) -> efka_subscription:subscribe(Topic, self()), - {ok, State}. + Reply = json_result(Id, <<"ok">>), + {reply, {binary, <>}, State}. -spec json_result(Id :: integer(), Result :: term()) -> binary(). json_result(Id, Result) when is_integer(Id) -> @@ -111,15 +112,4 @@ json_result(Id, Result) when is_integer(Id) -> <<"id">> => Id, <<"result">> => Result }, - jiffy:encode(Response, [force_utf8]). - --spec json_error(Id :: integer(), Code :: integer(), Message :: binary()) -> binary(). -json_error(Id, Code, Message) when is_integer(Id), is_integer(Code), is_binary(Message) -> - Response = #{ - <<"id">> => Id, - <<"error">> => #{ - <<"code">> => Code, - <<"message">> => Message - } - }, jiffy:encode(Response, [force_utf8]). \ No newline at end of file diff --git a/message_pb.proto b/message_pb.proto index e66838d..8a317d6 100644 --- a/message_pb.proto +++ b/message_pb.proto @@ -43,13 +43,6 @@ message FetchTaskLog { uint32 task_id = 1; } -// 需要响应, 云端主动发起的调用; 提供给用户 -message Invoke { - string service_id = 1; - bytes payload = 2; - uint32 timeout = 3; -} - // 参数配置 message ServiceConfig { string service_id = 1;