From eb3c9eb19dd2de7fea8de37271fcd273a5981b7d Mon Sep 17 00:00:00 2001 From: anlicheng <244108715@qq.com> Date: Tue, 22 Apr 2025 15:52:00 +0800 Subject: [PATCH] fix --- apps/efka/src/efka_agent.erl | 98 ++++-------------------------------- 1 file changed, 11 insertions(+), 87 deletions(-) diff --git a/apps/efka/src/efka_agent.erl b/apps/efka/src/efka_agent.erl index de66692..b1de784 100644 --- a/apps/efka/src/efka_agent.erl +++ b/apps/efka/src/efka_agent.erl @@ -46,8 +46,8 @@ start_link() -> {stop, Reason :: term()} | ignore). init([]) -> erlang:process_flag(trap_exit, true), - {ok, TransportPid} = efka_transport:start_link(self()), - {ok, #state{transport_pid = TransportPid}}. + erlang:start_timer(0, self(), create_transport), + {ok, #state{}}. %% @private %% @doc Handling call messages @@ -77,57 +77,17 @@ handle_cast(_Request, State = #state{}) -> {noreply, NewState :: #state{}} | {noreply, NewState :: #state{}, timeout() | hibernate} | {stop, Reason :: term(), NewState :: #state{}}). -handle_info({timeout, _, create_connection}, State = #state{host = Host, port = Port, packet_id = PacketId}) -> - case connect(Host, Port) of - {ok, Socket} -> - {ok, AuthInfo} = application:get_env(efka, auth), - UUID = proplists:get_value(uuid, AuthInfo), - Username = proplists:get_value(username, AuthInfo), - Salt = proplists:get_value(salt, AuthInfo), - Token = proplists:get_value(token, AuthInfo), +handle_info({timeout, _, create_transport}, State = #state{}) -> + {ok, TransportPid} = efka_transport:start_link(self()), + {noreply, State#state{transport_pid = TransportPid}}; - RequestBin = message_pb:encode_msg(#auth_request{ - uuid = unicode:characters_to_binary(UUID), - username = unicode:characters_to_binary(Username), - salt = unicode:characters_to_binary(Salt), - token = unicode:characters_to_binary(Token), - timestamp = efka_util:timestamp() - }), - ok = ssl:send(Socket, <>), - - %% 需要等待auth返回的结果 - receive - {ssl, Socket, <>} -> - AuthReply = message_pb:decode_msg(ReplyBin, auth_reply), - lager:debug("[efka_agent] get a reply bin: ~p", [AuthReply]), - - {noreply, State#state{socket = Socket}} - after 5000 -> - lager:debug("call me here timeout"), - ssl:close(Socket), - retry_connect(), - {noreply, State#state{socket = undefined}} - end; - {error, Reason} -> - lager:debug("[efka_agent] create_connection get error: ~p", [Reason]), - retry_connect(), - {noreply, State#state{socket = undefined}} - end; - -handle_info({ssl, Socket, Data}, State = #state{socket = Socket}) -> - lager:debug("[efka_agent] socket get message: ~p", [Data]), - {noreply, State#state{socket = undefined}}; - -handle_info({ssl_error, Socket, Reason}, State = #state{socket = Socket}) -> - lager:warning("[efka_agent] socket close with reason: ~p", [Reason]), +handle_info({'EXIT', _Pid, Reason}, State = #state{}) -> + lager:debug("[efka_agent] transport exit with reason: ~p", [Reason]), retry_connect(), - {noreply, State#state{socket = undefined}}; -handle_info({ssl_closed, Socket}, State = #state{socket = Socket}) -> - lager:warning("[efka_agent] socket closed"), - retry_connect(), - {noreply, State#state{socket = undefined}}; + {noreply, State#state{transport_pid = undefined}}; -handle_info(_Info, State = #state{}) -> +handle_info(Info, State = #state{}) -> + lager:debug("[efka_agent] get unknown info: ~p", [Info]), {noreply, State}. %% @private @@ -153,40 +113,4 @@ code_change(_OldVsn, State = #state{}, _Extra) -> %%%=================================================================== retry_connect() -> - erlang:start_timer(5000, self(), create_connection). - --spec connect(Host :: string(), Port :: integer()) -> {ok, Socket :: ssl:sslsocket()} | {error, Reason :: any()}. -connect(Host, Port) when is_list(Host), is_integer(Port) -> - SslOptions = [ - binary, - {packet, 4}, - {active, true}, - {verify, verify_none} - ], - ssl:connect(Host, Port, SslOptions, 5000). - --spec auth_request(Socket :: ssl:sslsocket(), PacketId :: integer()) -> {ok, AuthReply :: #auth_reply{}} | {error, Reason :: any()}. -auth_request(Socket, PacketId) when is_integer(PacketId) -> - {ok, AuthInfo} = application:get_env(efka, auth), - UUID = proplists:get_value(uuid, AuthInfo), - Username = proplists:get_value(username, AuthInfo), - Salt = proplists:get_value(salt, AuthInfo), - Token = proplists:get_value(token, AuthInfo), - - RequestBin = message_pb:encode_msg(#auth_request{ - uuid = unicode:characters_to_binary(UUID), - username = unicode:characters_to_binary(Username), - salt = unicode:characters_to_binary(Salt), - token = unicode:characters_to_binary(Token), - timestamp = efka_util:timestamp() - }), - ok = ssl:send(Socket, <>), - - %% 需要等待auth返回的结果 - receive - {ssl, Socket, <>} -> - lager:debug("[efka_agent] get a reply bin: ~p", [ReplyBin]), - {ok, message_pb:decode_msg(ReplyBin, auth_reply)} - after 5000 -> - {error, timeout} - end. \ No newline at end of file + erlang:start_timer(5000, self(), create_transport). \ No newline at end of file