fxi

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,

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, <<?PUSH_STOP_CONTAINER:8, Servi
 
     {keep_state, State};
 
-%% config.json配置信息
+%% config.json配置信息, TODO 推送配置不是推送到服务，而是修改容器的配置文件
 handle_event(info, {server_async_call, PacketId, <<?PUSH_SERVICE_CONFIG:8, ConfigBin/binary>>}, ?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, <<?PUSH_SERVICE_CONFIG:8, Confi
             {keep_state, State#state{push_inflight = maps:put(Ref, PacketId, PushInflight)}}
     end;
 
-%% 收到需要回复的指令
-handle_event(info, {server_async_call, PacketId, <<?PUSH_INVOKE:8, InvokeBin/binary>>}, ?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, <<?PUSH_TASK_LOG:8, TaskLogBin/binary>>}, ?STATE_ACTIVATED, State = #state{transport_pid = TransportPid}) ->
     #fetch_task_log{task_id = TaskId} = message_pb:decode_msg(TaskLogBin, fetch_task_log),

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]).
+-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 | invoke | service_config | data | event | ping.
+-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() | invoke() | 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, <<Bin/binary, 10>>, 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, <<B1/binary, 18>>, TrUserData)
-                    end
-                end
-         end,
-    if F3 == undefined -> B2;
-       true ->
-           begin
-               TrF3 = id(F3, TrUserData),
-               if TrF3 =:= 0 -> B2;
-                  true -> e_varint(TrF3, <<B2/binary, 24>>, 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, <<Bytes:Len/binary, Rest2/binary>> = 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, <<Bytes:Len/binary, Rest2/binary>> = 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";

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, <<?PACKET_RESPONSE:8, 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]).

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;