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

fix transport

Browse Source
This commit is contained in:
anlicheng 2025-04-22 17:35:21 +08:00
parent 5a795fc142
commit 57288fd464
2 changed files with 33 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

46
apps/efka/src/efka_agent.erl
View File

@ -81,32 +81,37 @@ handle_cast(_Request, State = #state{}) ->
{noreply, NewState :: #state{}, timeout() | hibernate} | {noreply, NewState :: #state{}, timeout() | hibernate} |
{stop, Reason :: term(), NewState :: #state{}}). {stop, Reason :: term(), NewState :: #state{}}).
handle_info({timeout, _, create_transport}, State = #state{}) -> handle_info({timeout, _, create_transport}, State = #state{}) ->
{ok, TransportPid} = efka_transport:start_link(self()), case efka_transport:start_link(self()) of
case efka_transport:auth_request(TransportPid, 5000) of {ok, TransportPid} ->
%% 验证通过 case efka_transport:auth_request(TransportPid, 5000) of
{ok, #auth_reply{code = 1, message = Message, repository_url = RepositoryUrl}} -> %% 验证通过
lager:debug("[efka_agent] auth result: ~p, repository_url: ~p", [Message, RepositoryUrl]), {ok, #auth_reply{code = 1, message = Message, repository_url = RepositoryUrl}} ->
{noreply, State#state{transport_pid = TransportPid, status = ?STATE_ACTIVATED}}; lager:debug("[efka_agent] auth result: ~p, repository_url: ~p", [Message, RepositoryUrl]),
{noreply, State#state{transport_pid = TransportPid, status = ?STATE_ACTIVATED}};
%% 主机denied状态 %% 主机denied状态
{ok, #auth_reply{code = -1, message = Message, repository_url = RepositoryUrl}} -> {ok, #auth_reply{code = -1, message = Message, repository_url = RepositoryUrl}} ->
lager:debug("[efka_agent] auth failed, message: ~p, repository_url: ~p", [Message, RepositoryUrl]), lager:debug("[efka_agent] auth failed, message: ~p, repository_url: ~p", [Message, RepositoryUrl]),
{noreply, State#state{transport_pid = TransportPid, status = ?STATE_DENIED}}; {noreply, State};
%% 验证不通过 %% 验证不通过
{ok, #auth_reply{code = -2, message = Message, repository_url = RepositoryUrl}} -> {ok, #auth_reply{code = -2, message = Message, repository_url = RepositoryUrl}} ->
lager:debug("[efka_agent] auth failed, message: ~p, repository_url: ~p", [Message, RepositoryUrl]), lager:debug("[efka_agent] auth failed, message: ~p, repository_url: ~p", [Message, RepositoryUrl]),
{noreply, State#state{transport_pid = undefined, status = ?STATE_DENIED}}; {noreply, State};
{error, Reason} ->
lager:debug("[efka_agent] auth_request failed, error: ~p", [Reason]),
efka_transport:stop(TransportPid),
{noreply, State}
end;
{error, Reason} -> {error, Reason} ->
lager:debug("[efka_agent] auth_request failed, error: ~p", [Reason]), lager:warning("[efka_agent] connect get error: ~p", [Reason]),
efka_transport:stop(TransportPid), {noreply, State}
{noreply, State#state{transport_pid = undefined, status = ?STATE_DENIED}}
end; end;
handle_info({'EXIT', _Pid, Reason}, State = #state{}) -> handle_info({'EXIT', _Pid, Reason}, State = #state{}) ->
lager:debug("[efka_agent] transport exit with reason: ~p", [Reason]), lager:debug("[efka_agent] transport exit with reason: ~p", [Reason]),
retry_connect(), erlang:start_timer(5000, self(), create_transport),
{noreply, State#state{transport_pid = undefined}}; {noreply, State#state{transport_pid = undefined}};
handle_info(Info, State = #state{}) -> handle_info(Info, State = #state{}) ->
@ -133,7 +138,4 @@ code_change(_OldVsn, State = #state{}, _Extra) ->
%%%=================================================================== %%%===================================================================
%%% Internal functions %%% Internal functions
%%%=================================================================== %%%===================================================================
retry_connect() ->
erlang:start_timer(5000, self(), create_transport).

14
apps/efka/src/efka_transport.erl
View File

@ -70,12 +70,15 @@ init([ParentPid]) ->
SslOptions = [ SslOptions = [
binary, binary,
{packet, 4}, {packet, 4},
{active, true},
{verify, verify_none} {verify, verify_none}
], ],
{ok, Socket} = ssl:connect(Host, Port, SslOptions, 5000), case ssl:connect(Host, Port, SslOptions, 5000) of
{ok, Socket} ->
{ok, #state{parent_pid = ParentPid, host = Host, port = Port, socket = Socket}}. ssl:controlling_process(Socket, self()),
{ok, #state{parent_pid = ParentPid, host = Host, port = Port, socket = Socket}};
{error, Reason} ->
{stop, Reason}
end.
%% @private %% @private
%% @doc Handling call messages %% @doc Handling call messages
@ -150,7 +153,8 @@ handle_info({ssl_error, Socket, Reason}, State = #state{socket = Socket}) ->
handle_info({ssl_closed, Socket}, State = #state{socket = Socket}) -> handle_info({ssl_closed, Socket}, State = #state{socket = Socket}) ->
{stop, ssl_closed, State}; {stop, ssl_closed, State};
handle_info(_Info, State = #state{}) -> handle_info(Info, State = #state{}) ->
lager:debug("[efka_transport] info: ~p", [Info]),
{noreply, State}. {noreply, State}.
%% @private %% @private