fix

apps/efka/include/efka.hrl

@@ -31,6 +31,9 @@
 %% ai识别的事件上报
 -define(METHOD_AI_EVENT, 16#08).
 
+%% 部署微服务
+-define(METHOD_DEPLOY, 16#10).
+
 %% 事件类型
 -define(EVENT_DEVICE, 16#01).
 %% 主机的相关事件

apps/efka/include/message_pb.hrl

@@ -35,6 +35,22 @@
         }).
 -endif.
 
+-ifndef('DEPLOY_PB_H').
+-define('DEPLOY_PB_H', true).
+-record(deploy,
+        {from = <<>>            :: unicode:chardata() | undefined, % = 1, optional
+         service_id = <<>>      :: unicode:chardata() | undefined % = 2, optional
+        }).
+-endif.
+
+-ifndef('DEPLOY_REPLY_PB_H').
+-define('DEPLOY_REPLY_PB_H', true).
+-record(deploy_reply,
+        {is_ok = false          :: boolean() | 0 | 1 | undefined, % = 1, optional
+         message = <<>>         :: unicode:chardata() | undefined % = 2, optional
+        }).
+-endif.
+
 -ifndef('TOPIC_MESSAGE_PB_H').
 -define('TOPIC_MESSAGE_PB_H', true).
 -record(topic_message,

apps/efka/src/efka_agent.erl

@@ -198,6 +198,20 @@ handle_info({auth_reply, {error, Reason}}, State = #state{transport_pid = Transp
 %% 云端服务器推送了消息
 %% 激活消息
 
+%% 微服务部署
+handle_info({server_push_message, PacketId, <<?METHOD_DEPLOY:8, DeployBin/binary>>}, State = #state{transport_pid = TransportPid}) ->
+    #deploy{from = From, service_id = ServiceId} = message_pb:decode_msg(DeployBin, deploy),
+
+    Reply = case efka_inetd:deploy(ServiceId, From) of
+                ok ->
+                    #deploy_reply{is_ok = true, message = <<"">>};
+                {error, Reason} ->
+                    #deploy_reply{is_ok = false, message = Reason}
+            end,
+    efka_transport:response(TransportPid, PacketId, message_pb:encode_msg(Reply)),
+
+    {noreply, State#state{status = ?STATE_RESTRICTED}};
+
 %% TODO
 handle_info({server_push_message, <<8:8, ActivatePush/binary>>}, State = #state{transport_pid = TransportPid, status = Status}) ->
     #activate_push{auth = Auth} = message_pb:decode_msg(ActivatePush, activate_push),

apps/efka/src/efka_inetd.erl

@@ -15,7 +15,7 @@
 
 %% API
 -export([start_link/0]).
--export([init_server/2]).
+-export([deploy/2]).
 -export([ensure_dirs/3, check_lock/1]).
 
 %% gen_server callbacks
@@ -31,8 +31,8 @@
 %%% API
 %%%===================================================================
 
-init_server(ServerId, From) when is_binary(ServerId), is_binary(From) ->
+deploy(ServerId, From) when is_binary(ServerId), is_binary(From) ->
-    gen_server:call(?SERVER, {init_server, ServerId, From}).
+    gen_server:call(?SERVER, {deploy, ServerId, From}).
 
 %% @doc Spawns the server and registers the local name (unique)
 -spec(start_link() ->
@@ -63,21 +63,27 @@ init([]) ->
     {noreply, NewState :: #state{}, timeout() | hibernate} |
     {stop, Reason :: term(), Reply :: term(), NewState :: #state{}} |
     {stop, Reason :: term(), NewState :: #state{}}).
-handle_call({init_server, ServerId, From}, _From, State = #state{root_dir = RootDir}) ->
+handle_call({deploy, ServerId, From}, _From, State = #state{root_dir = RootDir}) ->
     %% 创建目录
     {ok, ServerRootDir} = ensure_dirs(RootDir, ServerId, From),
     case check_lock(ServerRootDir) of
         true ->
-            ok;
+            {reply, ok, State};
         false ->
             DownloadUrl = make_download_url(From),
+            case check_download_url(DownloadUrl) of
+                ok ->
+                    {reply, ok, State};
+                {error, Reason} ->
+                    lager:debug("[efka_inetd] check_download_url: ~p, get error: ~p", [DownloadUrl, Reason]),
+                    {reply, {error, <<"download url error">>}, State}
+            end,
+
             {ok, TaskPid} = efka_downloader:start_link(),
             efka_downloader:download(TaskPid, DownloadUrl, ServerRootDir),
 
             ok
-    end,
+    end;
-
-    {reply, ok, State};
 
 handle_call(_Request, _From, State = #state{}) ->
     {reply, ok, State}.
@@ -144,9 +150,23 @@ check_lock(DirName) when is_list(DirName) ->
     FileName = DirName ++ ".efka.lock",
     filelib:is_file(FileName).
 
-%% 下载文件
+%% 通过head请求先判定下载地址是否正确
-download_file(From) when is_binary(From) ->
+check_download_url(Url) when is_binary(Url) ->
-    ok.
+    check_download_url(binary_to_list(Url));
+check_download_url(Url) when is_list(Url) ->
+    SslOpts = [
+        {ssl, [
+            % 完全禁用证书验证
+            {verify, verify_none}
+        ]}
+    ],
+
+    case httpc:request(head, {Url, []}, SslOpts, [{sync, true}]) of
+        {ok, {{_, 200, "OK"}, Headers, _}} ->
+            ok;
+        {error, Reason} ->
+            {error, Reason}
+    end.
 
 %%  From的格式: dianbiao:1.0
 -spec make_download_url(From :: binary()) -> string().

apps/efka/src/proto/message_pb.erl

@@ -61,6 +61,10 @@
 
 -type activate_push() :: #activate_push{}.
 
+-type deploy() :: #deploy{}.
+
+-type deploy_reply() :: #deploy_reply{}.
+
 -type topic_message() :: #topic_message{}.
 
 -type session_reply() :: #session_reply{}.
@@ -79,9 +83,9 @@
 
 -type ai_event() :: #ai_event{}.
 
--export_type(['auth_request'/0, 'auth_reply'/0, 'activate_push'/0, 'topic_message'/0, 'session_reply'/0, 'data'/0, 'ping'/0, 'service_inform'/0, 'feedback_step'/0, 'feedback_result'/0, 'event'/0, 'ai_event'/0]).
+-export_type(['auth_request'/0, 'auth_reply'/0, 'activate_push'/0, 'deploy'/0, 'deploy_reply'/0, 'topic_message'/0, 'session_reply'/0, 'data'/0, 'ping'/0, 'service_inform'/0, 'feedback_step'/0, 'feedback_result'/0, 'event'/0, 'ai_event'/0]).
--type '$msg_name'() :: auth_request | auth_reply | activate_push | topic_message | session_reply | data | ping | service_inform | feedback_step | feedback_result | event | ai_event.
+-type '$msg_name'() :: auth_request | auth_reply | activate_push | deploy | deploy_reply | topic_message | session_reply | data | ping | service_inform | feedback_step | feedback_result | event | ai_event.
--type '$msg'() :: auth_request() | auth_reply() | activate_push() | topic_message() | session_reply() | data() | ping() | service_inform() | feedback_step() | feedback_result() | event() | ai_event().
+-type '$msg'() :: auth_request() | auth_reply() | activate_push() | deploy() | deploy_reply() | topic_message() | session_reply() | data() | ping() | service_inform() | feedback_step() | feedback_result() | event() | ai_event().
 -export_type(['$msg_name'/0, '$msg'/0]).
 
 -record('map<string,string>',{key, value}).
@@ -112,6 +116,8 @@ encode_msg(Msg, MsgName, Opts) ->
         auth_request -> encode_msg_auth_request(id(Msg, TrUserData), TrUserData);
         auth_reply -> encode_msg_auth_reply(id(Msg, TrUserData), TrUserData);
         activate_push -> encode_msg_activate_push(id(Msg, TrUserData), TrUserData);
+        deploy -> encode_msg_deploy(id(Msg, TrUserData), TrUserData);
+        deploy_reply -> encode_msg_deploy_reply(id(Msg, TrUserData), TrUserData);
         topic_message -> encode_msg_topic_message(id(Msg, TrUserData), TrUserData);
         session_reply -> encode_msg_session_reply(id(Msg, TrUserData), TrUserData);
         data -> encode_msg_data(id(Msg, TrUserData), TrUserData);
@@ -226,6 +232,55 @@ encode_msg_activate_push(#activate_push{auth = F1}, Bin, TrUserData) ->
            end
     end.
 
+encode_msg_deploy(Msg, TrUserData) -> encode_msg_deploy(Msg, <<>>, TrUserData).
+
+
+encode_msg_deploy(#deploy{from = F1, service_id = F2}, 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,
+    if F2 == undefined -> B1;
+       true ->
+           begin
+               TrF2 = id(F2, TrUserData),
+               case is_empty_string(TrF2) of
+                   true -> B1;
+                   false -> e_type_string(TrF2, <<B1/binary, 18>>, TrUserData)
+               end
+           end
+    end.
+
+encode_msg_deploy_reply(Msg, TrUserData) -> encode_msg_deploy_reply(Msg, <<>>, TrUserData).
+
+
+encode_msg_deploy_reply(#deploy_reply{is_ok = F1, message = F2}, Bin, TrUserData) ->
+    B1 = if F1 == undefined -> Bin;
+            true ->
+                begin
+                    TrF1 = id(F1, TrUserData),
+                    if TrF1 =:= false -> Bin;
+                       true -> e_type_bool(TrF1, <<Bin/binary, 8>>, TrUserData)
+                    end
+                end
+         end,
+    if F2 == undefined -> B1;
+       true ->
+           begin
+               TrF2 = id(F2, TrUserData),
+               case is_empty_string(TrF2) of
+                   true -> B1;
+                   false -> e_type_string(TrF2, <<B1/binary, 18>>, TrUserData)
+               end
+           end
+    end.
+
 encode_msg_topic_message(Msg, TrUserData) -> encode_msg_topic_message(Msg, <<>>, TrUserData).
 
 
@@ -818,6 +873,8 @@ decode_msg_1_catch(Bin, MsgName, TrUserData) ->
 decode_msg_2_doit(auth_request, Bin, TrUserData) -> id(decode_msg_auth_request(Bin, TrUserData), TrUserData);
 decode_msg_2_doit(auth_reply, Bin, TrUserData) -> id(decode_msg_auth_reply(Bin, TrUserData), TrUserData);
 decode_msg_2_doit(activate_push, Bin, TrUserData) -> id(decode_msg_activate_push(Bin, TrUserData), TrUserData);
+decode_msg_2_doit(deploy, Bin, TrUserData) -> id(decode_msg_deploy(Bin, TrUserData), TrUserData);
+decode_msg_2_doit(deploy_reply, Bin, TrUserData) -> id(decode_msg_deploy_reply(Bin, TrUserData), TrUserData);
 decode_msg_2_doit(topic_message, Bin, TrUserData) -> id(decode_msg_topic_message(Bin, TrUserData), TrUserData);
 decode_msg_2_doit(session_reply, Bin, TrUserData) -> id(decode_msg_session_reply(Bin, TrUserData), TrUserData);
 decode_msg_2_doit(data, Bin, TrUserData) -> id(decode_msg_data(Bin, TrUserData), TrUserData);
@@ -1004,6 +1061,108 @@ skip_32_activate_push(<<_:32, Rest/binary>>, Z1, Z2, F, F@_1, TrUserData) -> dfp
 
 skip_64_activate_push(<<_:64, Rest/binary>>, Z1, Z2, F, F@_1, TrUserData) -> dfp_read_field_def_activate_push(Rest, Z1, Z2, F, F@_1, TrUserData).
 
+decode_msg_deploy(Bin, TrUserData) -> dfp_read_field_def_deploy(Bin, 0, 0, 0, id(<<>>, TrUserData), id(<<>>, TrUserData), TrUserData).
+
+dfp_read_field_def_deploy(<<10, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, TrUserData) -> d_field_deploy_from(Rest, Z1, Z2, F, F@_1, F@_2, TrUserData);
+dfp_read_field_def_deploy(<<18, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, TrUserData) -> d_field_deploy_service_id(Rest, Z1, Z2, F, F@_1, F@_2, TrUserData);
+dfp_read_field_def_deploy(<<>>, 0, 0, _, F@_1, F@_2, _) -> #deploy{from = F@_1, service_id = F@_2};
+dfp_read_field_def_deploy(Other, Z1, Z2, F, F@_1, F@_2, TrUserData) -> dg_read_field_def_deploy(Other, Z1, Z2, F, F@_1, F@_2, TrUserData).
+
+dg_read_field_def_deploy(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, TrUserData) when N < 32 - 7 -> dg_read_field_def_deploy(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, TrUserData);
+dg_read_field_def_deploy(<<0:1, X:7, Rest/binary>>, N, Acc, _, F@_1, F@_2, TrUserData) ->
+    Key = X bsl N + Acc,
+    case Key of
+        10 -> d_field_deploy_from(Rest, 0, 0, 0, F@_1, F@_2, TrUserData);
+        18 -> d_field_deploy_service_id(Rest, 0, 0, 0, F@_1, F@_2, TrUserData);
+        _ ->
+            case Key band 7 of
+                0 -> skip_varint_deploy(Rest, 0, 0, Key bsr 3, F@_1, F@_2, TrUserData);
+                1 -> skip_64_deploy(Rest, 0, 0, Key bsr 3, F@_1, F@_2, TrUserData);
+                2 -> skip_length_delimited_deploy(Rest, 0, 0, Key bsr 3, F@_1, F@_2, TrUserData);
+                3 -> skip_group_deploy(Rest, 0, 0, Key bsr 3, F@_1, F@_2, TrUserData);
+                5 -> skip_32_deploy(Rest, 0, 0, Key bsr 3, F@_1, F@_2, TrUserData)
+            end
+    end;
+dg_read_field_def_deploy(<<>>, 0, 0, _, F@_1, F@_2, _) -> #deploy{from = F@_1, service_id = F@_2}.
+
+d_field_deploy_from(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, TrUserData) when N < 57 -> d_field_deploy_from(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, TrUserData);
+d_field_deploy_from(<<0:1, X:7, Rest/binary>>, N, Acc, F, _, F@_2, 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_deploy(RestF, 0, 0, F, NewFValue, F@_2, TrUserData).
+
+d_field_deploy_service_id(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, TrUserData) when N < 57 -> d_field_deploy_service_id(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, TrUserData);
+d_field_deploy_service_id(<<0:1, X:7, Rest/binary>>, N, Acc, F, F@_1, _, 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_deploy(RestF, 0, 0, F, F@_1, NewFValue, TrUserData).
+
+skip_varint_deploy(<<1:1, _:7, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, TrUserData) -> skip_varint_deploy(Rest, Z1, Z2, F, F@_1, F@_2, TrUserData);
+skip_varint_deploy(<<0:1, _:7, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, TrUserData) -> dfp_read_field_def_deploy(Rest, Z1, Z2, F, F@_1, F@_2, TrUserData).
+
+skip_length_delimited_deploy(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, TrUserData) when N < 57 -> skip_length_delimited_deploy(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, TrUserData);
+skip_length_delimited_deploy(<<0:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, TrUserData) ->
+    Length = X bsl N + Acc,
+    <<_:Length/binary, Rest2/binary>> = Rest,
+    dfp_read_field_def_deploy(Rest2, 0, 0, F, F@_1, F@_2, TrUserData).
+
+skip_group_deploy(Bin, _, Z2, FNum, F@_1, F@_2, TrUserData) ->
+    {_, Rest} = read_group(Bin, FNum),
+    dfp_read_field_def_deploy(Rest, 0, Z2, FNum, F@_1, F@_2, TrUserData).
+
+skip_32_deploy(<<_:32, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, TrUserData) -> dfp_read_field_def_deploy(Rest, Z1, Z2, F, F@_1, F@_2, TrUserData).
+
+skip_64_deploy(<<_:64, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, TrUserData) -> dfp_read_field_def_deploy(Rest, Z1, Z2, F, F@_1, F@_2, TrUserData).
+
+decode_msg_deploy_reply(Bin, TrUserData) -> dfp_read_field_def_deploy_reply(Bin, 0, 0, 0, id(false, TrUserData), id(<<>>, TrUserData), TrUserData).
+
+dfp_read_field_def_deploy_reply(<<8, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, TrUserData) -> d_field_deploy_reply_is_ok(Rest, Z1, Z2, F, F@_1, F@_2, TrUserData);
+dfp_read_field_def_deploy_reply(<<18, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, TrUserData) -> d_field_deploy_reply_message(Rest, Z1, Z2, F, F@_1, F@_2, TrUserData);
+dfp_read_field_def_deploy_reply(<<>>, 0, 0, _, F@_1, F@_2, _) -> #deploy_reply{is_ok = F@_1, message = F@_2};
+dfp_read_field_def_deploy_reply(Other, Z1, Z2, F, F@_1, F@_2, TrUserData) -> dg_read_field_def_deploy_reply(Other, Z1, Z2, F, F@_1, F@_2, TrUserData).
+
+dg_read_field_def_deploy_reply(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, TrUserData) when N < 32 - 7 -> dg_read_field_def_deploy_reply(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, TrUserData);
+dg_read_field_def_deploy_reply(<<0:1, X:7, Rest/binary>>, N, Acc, _, F@_1, F@_2, TrUserData) ->
+    Key = X bsl N + Acc,
+    case Key of
+        8 -> d_field_deploy_reply_is_ok(Rest, 0, 0, 0, F@_1, F@_2, TrUserData);
+        18 -> d_field_deploy_reply_message(Rest, 0, 0, 0, F@_1, F@_2, TrUserData);
+        _ ->
+            case Key band 7 of
+                0 -> skip_varint_deploy_reply(Rest, 0, 0, Key bsr 3, F@_1, F@_2, TrUserData);
+                1 -> skip_64_deploy_reply(Rest, 0, 0, Key bsr 3, F@_1, F@_2, TrUserData);
+                2 -> skip_length_delimited_deploy_reply(Rest, 0, 0, Key bsr 3, F@_1, F@_2, TrUserData);
+                3 -> skip_group_deploy_reply(Rest, 0, 0, Key bsr 3, F@_1, F@_2, TrUserData);
+                5 -> skip_32_deploy_reply(Rest, 0, 0, Key bsr 3, F@_1, F@_2, TrUserData)
+            end
+    end;
+dg_read_field_def_deploy_reply(<<>>, 0, 0, _, F@_1, F@_2, _) -> #deploy_reply{is_ok = F@_1, message = F@_2}.
+
+d_field_deploy_reply_is_ok(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, TrUserData) when N < 57 -> d_field_deploy_reply_is_ok(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, TrUserData);
+d_field_deploy_reply_is_ok(<<0:1, X:7, Rest/binary>>, N, Acc, F, _, F@_2, TrUserData) ->
+    {NewFValue, RestF} = {id(X bsl N + Acc =/= 0, TrUserData), Rest},
+    dfp_read_field_def_deploy_reply(RestF, 0, 0, F, NewFValue, F@_2, TrUserData).
+
+d_field_deploy_reply_message(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, TrUserData) when N < 57 -> d_field_deploy_reply_message(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, TrUserData);
+d_field_deploy_reply_message(<<0:1, X:7, Rest/binary>>, N, Acc, F, F@_1, _, 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_deploy_reply(RestF, 0, 0, F, F@_1, NewFValue, TrUserData).
+
+skip_varint_deploy_reply(<<1:1, _:7, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, TrUserData) -> skip_varint_deploy_reply(Rest, Z1, Z2, F, F@_1, F@_2, TrUserData);
+skip_varint_deploy_reply(<<0:1, _:7, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, TrUserData) -> dfp_read_field_def_deploy_reply(Rest, Z1, Z2, F, F@_1, F@_2, TrUserData).
+
+skip_length_delimited_deploy_reply(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, TrUserData) when N < 57 -> skip_length_delimited_deploy_reply(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, TrUserData);
+skip_length_delimited_deploy_reply(<<0:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, TrUserData) ->
+    Length = X bsl N + Acc,
+    <<_:Length/binary, Rest2/binary>> = Rest,
+    dfp_read_field_def_deploy_reply(Rest2, 0, 0, F, F@_1, F@_2, TrUserData).
+
+skip_group_deploy_reply(Bin, _, Z2, FNum, F@_1, F@_2, TrUserData) ->
+    {_, Rest} = read_group(Bin, FNum),
+    dfp_read_field_def_deploy_reply(Rest, 0, Z2, FNum, F@_1, F@_2, TrUserData).
+
+skip_32_deploy_reply(<<_:32, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, TrUserData) -> dfp_read_field_def_deploy_reply(Rest, Z1, Z2, F, F@_1, F@_2, TrUserData).
+
+skip_64_deploy_reply(<<_:64, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, TrUserData) -> dfp_read_field_def_deploy_reply(Rest, Z1, Z2, F, F@_1, F@_2, TrUserData).
+
 decode_msg_topic_message(Bin, TrUserData) -> dfp_read_field_def_topic_message(Bin, 0, 0, 0, id(<<>>, TrUserData), id(<<>>, TrUserData), TrUserData).
 
 dfp_read_field_def_topic_message(<<10, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, TrUserData) -> d_field_topic_message_topic(Rest, Z1, Z2, F, F@_1, F@_2, TrUserData);
@@ -1826,6 +1985,8 @@ merge_msgs(Prev, New, MsgName, Opts) ->
         auth_request -> merge_msg_auth_request(Prev, New, TrUserData);
         auth_reply -> merge_msg_auth_reply(Prev, New, TrUserData);
         activate_push -> merge_msg_activate_push(Prev, New, TrUserData);
+        deploy -> merge_msg_deploy(Prev, New, TrUserData);
+        deploy_reply -> merge_msg_deploy_reply(Prev, New, TrUserData);
         topic_message -> merge_msg_topic_message(Prev, New, TrUserData);
         session_reply -> merge_msg_session_reply(Prev, New, TrUserData);
         data -> merge_msg_data(Prev, New, TrUserData);
@@ -1882,6 +2043,28 @@ merge_msg_activate_push(#activate_push{auth = PFauth}, #activate_push{auth = NFa
                           true -> NFauth
                        end}.
 
+-compile({nowarn_unused_function,merge_msg_deploy/3}).
+merge_msg_deploy(#deploy{from = PFfrom, service_id = PFservice_id}, #deploy{from = NFfrom, service_id = NFservice_id}, _) ->
+    #deploy{from =
+                if NFfrom =:= undefined -> PFfrom;
+                   true -> NFfrom
+                end,
+            service_id =
+                if NFservice_id =:= undefined -> PFservice_id;
+                   true -> NFservice_id
+                end}.
+
+-compile({nowarn_unused_function,merge_msg_deploy_reply/3}).
+merge_msg_deploy_reply(#deploy_reply{is_ok = PFis_ok, message = PFmessage}, #deploy_reply{is_ok = NFis_ok, message = NFmessage}, _) ->
+    #deploy_reply{is_ok =
+                      if NFis_ok =:= undefined -> PFis_ok;
+                         true -> NFis_ok
+                      end,
+                  message =
+                      if NFmessage =:= undefined -> PFmessage;
+                         true -> NFmessage
+                      end}.
+
 -compile({nowarn_unused_function,merge_msg_topic_message/3}).
 merge_msg_topic_message(#topic_message{topic = PFtopic, content = PFcontent}, #topic_message{topic = NFtopic, content = NFcontent}, _) ->
     #topic_message{topic =
@@ -2090,6 +2273,8 @@ verify_msg(Msg, MsgName, Opts) ->
         auth_request -> v_msg_auth_request(Msg, [MsgName], TrUserData);
         auth_reply -> v_msg_auth_reply(Msg, [MsgName], TrUserData);
         activate_push -> v_msg_activate_push(Msg, [MsgName], TrUserData);
+        deploy -> v_msg_deploy(Msg, [MsgName], TrUserData);
+        deploy_reply -> v_msg_deploy_reply(Msg, [MsgName], TrUserData);
         topic_message -> v_msg_topic_message(Msg, [MsgName], TrUserData);
         session_reply -> v_msg_session_reply(Msg, [MsgName], TrUserData);
         data -> v_msg_data(Msg, [MsgName], TrUserData);
@@ -2148,6 +2333,30 @@ v_msg_activate_push(#activate_push{auth = F1}, Path, TrUserData) ->
     ok;
 v_msg_activate_push(X, Path, _TrUserData) -> mk_type_error({expected_msg, activate_push}, X, Path).
 
+-compile({nowarn_unused_function,v_msg_deploy/3}).
+-dialyzer({nowarn_function,v_msg_deploy/3}).
+v_msg_deploy(#deploy{from = F1, service_id = F2}, Path, TrUserData) ->
+    if F1 == undefined -> ok;
+       true -> v_type_string(F1, [from | Path], TrUserData)
+    end,
+    if F2 == undefined -> ok;
+       true -> v_type_string(F2, [service_id | Path], TrUserData)
+    end,
+    ok;
+v_msg_deploy(X, Path, _TrUserData) -> mk_type_error({expected_msg, deploy}, X, Path).
+
+-compile({nowarn_unused_function,v_msg_deploy_reply/3}).
+-dialyzer({nowarn_function,v_msg_deploy_reply/3}).
+v_msg_deploy_reply(#deploy_reply{is_ok = F1, message = F2}, Path, TrUserData) ->
+    if F1 == undefined -> ok;
+       true -> v_type_bool(F1, [is_ok | Path], TrUserData)
+    end,
+    if F2 == undefined -> ok;
+       true -> v_type_string(F2, [message | Path], TrUserData)
+    end,
+    ok;
+v_msg_deploy_reply(X, Path, _TrUserData) -> mk_type_error({expected_msg, deploy_reply}, X, Path).
+
 -compile({nowarn_unused_function,v_msg_topic_message/3}).
 -dialyzer({nowarn_function,v_msg_topic_message/3}).
 v_msg_topic_message(#topic_message{topic = F1, content = F2}, Path, TrUserData) ->
@@ -2468,6 +2677,8 @@ get_msg_defs() ->
        #field{name = message, fnum = 2, rnum = 3, type = string, occurrence = optional, opts = []},
        #field{name = repository_url, fnum = 3, rnum = 4, type = string, occurrence = optional, opts = []}]},
      {{msg, activate_push}, [#field{name = auth, fnum = 1, rnum = 2, type = bool, occurrence = optional, opts = []}]},
+     {{msg, deploy}, [#field{name = from, fnum = 1, rnum = 2, type = string, occurrence = optional, opts = []}, #field{name = service_id, fnum = 2, rnum = 3, type = string, occurrence = optional, opts = []}]},
+     {{msg, deploy_reply}, [#field{name = is_ok, fnum = 1, rnum = 2, type = bool, occurrence = optional, opts = []}, #field{name = message, fnum = 2, rnum = 3, type = string, occurrence = optional, opts = []}]},
      {{msg, topic_message}, [#field{name = topic, fnum = 1, rnum = 2, type = string, occurrence = optional, opts = []}, #field{name = content, fnum = 2, rnum = 3, type = string, occurrence = optional, opts = []}]},
      {{msg, session_reply}, [#field{name = a, fnum = 1, rnum = 2, type = bool, occurrence = optional, opts = []}]},
      {{msg, data},
@@ -2509,13 +2720,13 @@ get_msg_defs() ->
      {{msg, ai_event}, [#field{name = event_type, fnum = 1, rnum = 2, type = uint32, occurrence = optional, opts = []}, #field{name = params, fnum = 2, rnum = 3, type = bytes, occurrence = optional, opts = []}]}].
 
 
-get_msg_names() -> [auth_request, auth_reply, activate_push, topic_message, session_reply, data, ping, service_inform, feedback_step, feedback_result, event, ai_event].
+get_msg_names() -> [auth_request, auth_reply, activate_push, deploy, deploy_reply, topic_message, session_reply, data, ping, service_inform, feedback_step, feedback_result, event, ai_event].
 
 
 get_group_names() -> [].
 
 
-get_msg_or_group_names() -> [auth_request, auth_reply, activate_push, topic_message, session_reply, data, ping, service_inform, feedback_step, feedback_result, event, ai_event].
+get_msg_or_group_names() -> [auth_request, auth_reply, activate_push, deploy, deploy_reply, topic_message, session_reply, data, ping, service_inform, feedback_step, feedback_result, event, ai_event].
 
 
 get_enum_names() -> [].
@@ -2543,6 +2754,8 @@ find_msg_def(auth_reply) ->
      #field{name = message, fnum = 2, rnum = 3, type = string, occurrence = optional, opts = []},
      #field{name = repository_url, fnum = 3, rnum = 4, type = string, occurrence = optional, opts = []}];
 find_msg_def(activate_push) -> [#field{name = auth, fnum = 1, rnum = 2, type = bool, occurrence = optional, opts = []}];
+find_msg_def(deploy) -> [#field{name = from, fnum = 1, rnum = 2, type = string, occurrence = optional, opts = []}, #field{name = service_id, fnum = 2, rnum = 3, type = string, occurrence = optional, opts = []}];
+find_msg_def(deploy_reply) -> [#field{name = is_ok, fnum = 1, rnum = 2, type = bool, occurrence = optional, opts = []}, #field{name = message, fnum = 2, rnum = 3, type = string, occurrence = optional, opts = []}];
 find_msg_def(topic_message) -> [#field{name = topic, fnum = 1, rnum = 2, type = string, occurrence = optional, opts = []}, #field{name = content, fnum = 2, rnum = 3, type = string, occurrence = optional, opts = []}];
 find_msg_def(session_reply) -> [#field{name = a, fnum = 1, rnum = 2, type = bool, occurrence = optional, opts = []}];
 find_msg_def(data) ->
@@ -2643,6 +2856,8 @@ service_and_rpc_name_to_fqbins(S, R) -> error({gpb_error, {badservice_or_rpc, {S
 fqbin_to_msg_name(<<"AuthRequest">>) -> auth_request;
 fqbin_to_msg_name(<<"AuthReply">>) -> auth_reply;
 fqbin_to_msg_name(<<"ActivatePush">>) -> activate_push;
+fqbin_to_msg_name(<<"Deploy">>) -> deploy;
+fqbin_to_msg_name(<<"DeployReply">>) -> deploy_reply;
 fqbin_to_msg_name(<<"TopicMessage">>) -> topic_message;
 fqbin_to_msg_name(<<"SessionReply">>) -> session_reply;
 fqbin_to_msg_name(<<"Data">>) -> data;
@@ -2658,6 +2873,8 @@ fqbin_to_msg_name(E) -> error({gpb_error, {badmsg, E}}).
 msg_name_to_fqbin(auth_request) -> <<"AuthRequest">>;
 msg_name_to_fqbin(auth_reply) -> <<"AuthReply">>;
 msg_name_to_fqbin(activate_push) -> <<"ActivatePush">>;
+msg_name_to_fqbin(deploy) -> <<"Deploy">>;
+msg_name_to_fqbin(deploy_reply) -> <<"DeployReply">>;
 msg_name_to_fqbin(topic_message) -> <<"TopicMessage">>;
 msg_name_to_fqbin(session_reply) -> <<"SessionReply">>;
 msg_name_to_fqbin(data) -> <<"Data">>;
@@ -2705,7 +2922,7 @@ get_all_source_basenames() -> ["message_pb.proto"].
 get_all_proto_names() -> ["message_pb"].
 
 
-get_msg_containment("message_pb") -> [ai_event, activate_push, auth_reply, auth_request, data, event, feedback_result, feedback_step, ping, service_inform, session_reply, topic_message];
+get_msg_containment("message_pb") -> [ai_event, activate_push, auth_reply, auth_request, data, deploy, deploy_reply, event, feedback_result, feedback_step, ping, service_inform, session_reply, topic_message];
 get_msg_containment(P) -> error({gpb_error, {badproto, P}}).
 
 
@@ -2735,6 +2952,8 @@ get_proto_by_msg_name_as_fqbin(<<"TopicMessage">>) -> "message_pb";
 get_proto_by_msg_name_as_fqbin(<<"Ping">>) -> "message_pb";
 get_proto_by_msg_name_as_fqbin(<<"ActivatePush">>) -> "message_pb";
 get_proto_by_msg_name_as_fqbin(<<"SessionReply">>) -> "message_pb";
+get_proto_by_msg_name_as_fqbin(<<"DeployReply">>) -> "message_pb";
+get_proto_by_msg_name_as_fqbin(<<"Deploy">>) -> "message_pb";
 get_proto_by_msg_name_as_fqbin(<<"AuthReply">>) -> "message_pb";
 get_proto_by_msg_name_as_fqbin(<<"ServiceInform">>) -> "message_pb";
 get_proto_by_msg_name_as_fqbin(E) -> error({gpb_error, {badmsg, E}}).

efkafile.json

@@ -6,9 +6,5 @@
   },
   "metrics": {
 
-  },
-  "env": {
-
   }
-
 }

message_pb.proto

@@ -21,6 +21,17 @@ message ActivatePush {
     bool auth = 1;
 }
 
+// 部署逻辑
+message Deploy {
+    string from = 1;
+    string service_id = 2;
+}
+
+message DeployReply {
+    bool is_ok = 1;
+    string message = 2;
+}
+
 // 基于pub/sub的机制实现远程调用
 message TopicMessage {
     string topic = 1;