anlicheng/ekfa
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix

Browse Source
This commit is contained in:
anlicheng 2025-09-26 15:33:38 +08:00
parent 09ae0bc8c0
commit e334df0939
2 changed files with 25 additions and 25 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

48
apps/efka/src/efka_remote_agent.erl
View File

@ -240,71 +240,71 @@ handle_event(info, flush_cache, _, State) ->
%% 激活消息
%% 微服务部署
handle_event(info, {server_rpc, PacketId, #jsonrpc_request{id = Id, method = <<"deploy">>, params = Params}}, ?STATE_ACTIVATED, State = #state{transport_pid = TransportPid}) ->
handle_event(info, {server_rpc, PacketId, #jsonrpc_request{method = <<"deploy">>, params = Params}}, ?STATE_ACTIVATED, State = #state{transport_pid = TransportPid}) ->
#{<<"task_id">> := TaskId, <<"config">> := Config} = Params,
%% 短暂的等待，efka_inetd收到消息后就立即返回了
case docker_manager:deploy(TaskId, Config) of
ok ->
efka_transport:rpc_reply(TransportPid, PacketId, reply_success(Id, <<"ok">>));
efka_transport:rpc_reply(TransportPid, PacketId, reply_success(<<"ok">>));
{error, Reason} when is_binary(Reason) ->
efka_transport:rpc_reply(TransportPid, PacketId, reply_error(Id, -1, Reason))
efka_transport:rpc_reply(TransportPid, PacketId, reply_error(-1, Reason))
end,
{keep_state, State};
%% 启动微服务
handle_event(info, {server_rpc, PacketId, #jsonrpc_request{id = Id, method = <<"start_container">>, params = Params}}, ?STATE_ACTIVATED, State = #state{transport_pid = TransportPid}) ->
handle_event(info, {server_rpc, PacketId, #jsonrpc_request{method = <<"start_container">>, params = Params}}, ?STATE_ACTIVATED, State = #state{transport_pid = TransportPid}) ->
#{<<"container_name">> := ContainerName} = Params,
%% 短暂的等待，efka_inetd收到消息后就立即返回了
case docker_manager:start_container(ContainerName) of
ok ->
efka_transport:rpc_reply(TransportPid, PacketId, reply_success(Id, <<"ok">>));
efka_transport:rpc_reply(TransportPid, PacketId, reply_success(<<"ok">>));
{error, Reason} when is_binary(Reason) ->
efka_transport:rpc_reply(TransportPid, PacketId, reply_error(Id, -1, Reason))
efka_transport:rpc_reply(TransportPid, PacketId, reply_error(-1, Reason))
end,
{keep_state, State};
%% 停止微服务
handle_event(info, {server_rpc, PacketId, #jsonrpc_request{id = Id, method = <<"stop_container">>, params = Params}}, ?STATE_ACTIVATED, State = #state{transport_pid = TransportPid}) ->
handle_event(info, {server_rpc, PacketId, #jsonrpc_request{method = <<"stop_container">>, params = Params}}, ?STATE_ACTIVATED, State = #state{transport_pid = TransportPid}) ->
#{<<"container_name">> := ContainerName} = Params,
%% 短暂的等待，efka_inetd收到消息后就立即返回了
case docker_manager:stop_container(ContainerName) of
ok ->
efka_transport:rpc_reply(TransportPid, PacketId, reply_success(Id, <<"ok">>));
efka_transport:rpc_reply(TransportPid, PacketId, reply_success(<<"ok">>));
{error, Reason} when is_binary(Reason) ->
efka_transport:rpc_reply(TransportPid, PacketId, reply_error(Id, -1, Reason))
efka_transport:rpc_reply(TransportPid, PacketId, reply_error(-1, Reason))
end,
{keep_state, State};
handle_event(info, {server_rpc, PacketId, #jsonrpc_request{id = Id, method = <<"kill_container">>, params = Params}}, ?STATE_ACTIVATED, State = #state{transport_pid = TransportPid}) ->
handle_event(info, {server_rpc, PacketId, #jsonrpc_request{method = <<"kill_container">>, params = Params}}, ?STATE_ACTIVATED, State = #state{transport_pid = TransportPid}) ->
#{<<"container_name">> := ContainerName} = Params,
%% 短暂的等待，efka_inetd收到消息后就立即返回了
case docker_manager:kill_container(ContainerName) of
ok ->
efka_transport:rpc_reply(TransportPid, PacketId, reply_success(Id, <<"ok">>));
efka_transport:rpc_reply(TransportPid, PacketId, reply_success(<<"ok">>));
{error, Reason} when is_binary(Reason) ->
efka_transport:rpc_reply(TransportPid, PacketId, reply_error(Id, -1, Reason))
efka_transport:rpc_reply(TransportPid, PacketId, reply_error(-1, Reason))
end,
{keep_state, State};
handle_event(info, {server_rpc, PacketId, #jsonrpc_request{id = Id, method = <<"remove_container">>, params = Params}}, ?STATE_ACTIVATED, State = #state{transport_pid = TransportPid}) ->
handle_event(info, {server_rpc, PacketId, #jsonrpc_request{method = <<"remove_container">>, params = Params}}, ?STATE_ACTIVATED, State = #state{transport_pid = TransportPid}) ->
#{<<"container_name">> := ContainerName} = Params,
%% 短暂的等待，efka_inetd收到消息后就立即返回了
case docker_manager:remove_container(ContainerName) of
ok ->
efka_transport:rpc_reply(TransportPid, PacketId, reply_success(Id, <<"ok">>));
efka_transport:rpc_reply(TransportPid, PacketId, reply_success(<<"ok">>));
{error, Reason} when is_binary(Reason) ->
efka_transport:rpc_reply(TransportPid, PacketId, reply_error(Id, -1, Reason))
efka_transport:rpc_reply(TransportPid, PacketId, reply_error(-1, Reason))
end,
{keep_state, State};
%% config.json配置信息
handle_event(info, {server_rpc, PacketId, #jsonrpc_request{id = Id, method = <<"config_container">>, params = Params}}, ?STATE_ACTIVATED, State = #state{transport_pid = TransportPid}) ->
handle_event(info, {server_rpc, PacketId, #jsonrpc_request{method = <<"config_container">>, params = Params}}, ?STATE_ACTIVATED, State = #state{transport_pid = TransportPid}) ->
#{<<"container_name">> := ContainerName, <<"config">> := Config} = Params,
case docker_manager:config_container(ContainerName, Config) of
ok ->
efka_transport:rpc_reply(TransportPid, PacketId, reply_success(Id, <<"ok">>));
efka_transport:rpc_reply(TransportPid, PacketId, reply_success(<<"ok">>));
{error, Reason} ->
efka_transport:rpc_reply(TransportPid, PacketId, reply_error(Id, -1, Reason))
efka_transport:rpc_reply(TransportPid, PacketId, reply_error(-1, Reason))
end,
{keep_state, State};
@ -392,14 +392,14 @@ auth_request() ->
timestamp = efka_util:timestamp()
}).
-spec reply_success(Id :: integer(), Result :: any()) -> binary().
reply_success(Id, Result) when is_binary(Result) ->
message_codec:encode(?MESSAGE_JSONRPC_REPLY, #jsonrpc_reply{id = Id, result = Result}).
-spec reply_success(Result :: any()) -> binary().
reply_success(Result) when is_binary(Result) ->
message_codec:encode(?MESSAGE_JSONRPC_REPLY, #jsonrpc_reply{result = Result}).
-spec reply_error(Id :: integer(), Code :: integer(), Message :: binary()) -> binary().
reply_error(Id, Code, Message) when is_integer(Id), is_integer(Code), is_binary(Message) ->
-spec reply_error(Code :: integer(), Message :: binary()) -> binary().
reply_error(Code, Message) when is_integer(Code), is_binary(Message) ->
Error = #{
<<"code">> => Code,
<<"message">> => Message
},
message_codec:encode(?MESSAGE_JSONRPC_REPLY, #jsonrpc_reply{id = Id, error = Error}).
message_codec:encode(?MESSAGE_JSONRPC_REPLY, #jsonrpc_reply{error = Error}).

2
apps/efka/src/message/message_codec.erl
View File

@ -87,7 +87,7 @@ decode0(?MESSAGE_JSONRPC_REPLY, [ReplyBin]) ->
case jiffy:decode(ReplyBin, [return_maps]) of
#{<<"result">> := Result} ->
{ok, #jsonrpc_reply{result = Result}};
#{<<"id">> := Id, <<"error">> := Error} ->
#{<<"error">> := Error} ->
{ok, #jsonrpc_reply{error = Error}};
_ ->
error