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-04-30 16:57:04 +08:00
parent bc8517dc04
commit c21ce9f239
1 changed files with 13 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
apps/efka/src/efka_micro_service.erl
View File

@ -93,8 +93,8 @@ init([Service = #micro_service{service_id = ServiceId, status = Status}]) ->
true -> true ->
case boot_service(Service) of case boot_service(Service) of
{ok, Port} -> {ok, Port} ->
{os_pid, OSPid} = port_info(Port, os_pid), {os_pid, OSPid} = erlang:port_info(Port, os_pid),
lager:debug("[efka_micro_service] service: ~p, boot_service success os_pid: ~p", [ServiceId, OSPid]), lager:debug("[efka_micro_service] service: ~p, port: ~p, boot_service success os_pid: ~p", [ServiceId, Port, OSPid]),
{ok, #state{service = Service, running_status = ?STATUS_RUNNING, port = Port, os_pid = OSPid}}; {ok, #state{service = Service, running_status = ?STATUS_RUNNING, port = Port, os_pid = OSPid}};
{error, Reason} -> {error, Reason} ->
lager:debug("[efka_micro_service] service: ~p, boot_service get error: ~p", [ServiceId, Reason]), lager:debug("[efka_micro_service] service: ~p, boot_service get error: ~p", [ServiceId, Reason]),
@ -136,7 +136,8 @@ handle_call(start_service, _From, State = #state{running_status = ?STATUS_STOPPE
{ok, Port} -> {ok, Port} ->
%% 更新数据库状态 %% 更新数据库状态
micro_service_model:start_service(ServiceId), micro_service_model:start_service(ServiceId),
{os_pid, OSPid} = port_info(Port, os_pid), {os_pid, OSPid} = erlang:port_info(Port, os_pid),
lager:debug("start_service port: ~p, os_pid: ~p", [Port, OSPid]),
{reply, ok, State#state{running_status = ?STATUS_RUNNING, port = Port, os_pid = OSPid, service = Service#micro_service{status = 1}}}; {reply, ok, State#state{running_status = ?STATUS_RUNNING, port = Port, os_pid = OSPid, service = Service#micro_service{status = 1}}};
{error, Reason} -> {error, Reason} ->
%% 启动失败不能更新数据库里面的状态 %% 启动失败不能更新数据库里面的状态
@ -150,6 +151,10 @@ handle_call(stop_service, _From, State = #state{running_status = ?STATUS_STOPPED
handle_call(stop_service, _From, State = #state{running_status = ?STATUS_RUNNING, port = Port, os_pid = OSPid, service = Service = #micro_service{service_id = ServiceId}}) when is_port(Port) -> handle_call(stop_service, _From, State = #state{running_status = ?STATUS_RUNNING, port = Port, os_pid = OSPid, service = Service = #micro_service{service_id = ServiceId}}) when is_port(Port) ->
micro_service_model:stop_service(ServiceId), micro_service_model:stop_service(ServiceId),
kill_os_pid(OSPid), kill_os_pid(OSPid),
erlang:is_port(Port) andalso erlang:port_close(Port),
lager:debug("port: ~p, os_pid: ~p, will closed", [Port, OSPid]),
{reply, ok, State#state{port = undefined, os_pid = undefined, running_status = ?STATUS_STOPPED, service = Service#micro_service{status = 0}}}; {reply, ok, State#state{port = undefined, os_pid = undefined, running_status = ?STATUS_STOPPED, service = Service#micro_service{status = 0}}};
handle_call(_Request, _From, State = #state{}) -> handle_call(_Request, _From, State = #state{}) ->
@ -178,7 +183,7 @@ handle_info({timeout, _, reboot_service}, State = #state{service = Service = #mi
case boot_service(Service) of case boot_service(Service) of
{ok, Port} -> {ok, Port} ->
{os_pid, OSPid} = erlang:port_info(Port, os_pid), {os_pid, OSPid} = erlang:port_info(Port, os_pid),
lager:debug("[efka_micro_service] reboot service success: ~p, os_pid: ~p", [ServiceId, OSPid]), lager:debug("[efka_micro_service] reboot service success: ~p, port: ~p, os_pid: ~p", [ServiceId, Port, OSPid]),
{noreply, State#state{running_status = ?STATUS_RUNNING, port = Port, os_pid = OSPid}}; {noreply, State#state{running_status = ?STATUS_RUNNING, port = Port, os_pid = OSPid}};
{error, Reason} -> {error, Reason} ->
lager:debug("[efka_micro_service] service: ~p, boot_service get error: ~p", [ServiceId, Reason]), lager:debug("[efka_micro_service] service: ~p, boot_service get error: ~p", [ServiceId, Reason]),
@ -190,9 +195,9 @@ handle_info({Port, {data, Data}}, State = #state{port = Port, service = #micro_s
{noreply, State}; {noreply, State};
%% 处理port的消息, Port的被动关闭会触发；因此这个时候的Port和State.port的值是相等的 %% 处理port的消息, Port的被动关闭会触发；因此这个时候的Port和State.port的值是相等的
handle_info({Port, {exit_status, Code}}, State = #state{port = Port, service = #micro_service{service_id = ServiceId}}) -> handle_info({Port, {exit_status, Code}}, State = #state{service = #micro_service{service_id = ServiceId}}) ->
lager:debug("[efka_micro_service] service_id: ~p, port exit with code: ~p", [ServiceId, Code]), lager:debug("[efka_micro_service] service_id: ~p, port: ~p, exit with code: ~p", [ServiceId, Port, Code]),
erlang:start_timer(5000, self(), reboot_service), % erlang:start_timer(5000, self(), reboot_service),
{noreply, State#state{port = undefined, os_pid = undefined, running_status = ?STATUS_STOPPED}}; {noreply, State#state{port = undefined, os_pid = undefined, running_status = ?STATUS_STOPPED}};
%% 处理channel进程的退出 %% 处理channel进程的退出
@ -236,6 +241,7 @@ kill_os_pid(OSPid) when is_integer(OSPid) ->
%% 启动微服务 %% 启动微服务
-spec boot_service(MicroService :: #micro_service{}) -> {ok, Port :: port()} | {error, Reason :: binary()}. -spec boot_service(MicroService :: #micro_service{}) -> {ok, Port :: port()} | {error, Reason :: binary()}.
boot_service(#micro_service{work_dir = WorkDir0, service_id = ServiceId}) -> boot_service(#micro_service{work_dir = WorkDir0, service_id = ServiceId}) ->
lager:debug("who call me here: ~p", ["funck"]),
WorkDir = binary_to_list(WorkDir0), WorkDir = binary_to_list(WorkDir0),
case file:read_file(WorkDir ++ "manifest.json") of case file:read_file(WorkDir ++ "manifest.json") of
{ok, ManifestInfo} -> {ok, ManifestInfo} ->