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-22 15:52:00 +08:00
parent 332e3f8dc0
commit eb3c9eb19d
1 changed files with 11 additions and 87 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -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, <<?PACKET_REQUEST, PacketId:32, ?METHOD_AUTH, RequestBin/binary>>),
%% 需要等待auth返回的结果
receive
{ssl, Socket, <<?PACKET_RESPONSE, PacketId:32, ?METHOD_AUTH, ReplyBin/binary>>} ->
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, <<?PACKET_REQUEST, PacketId:32, ?METHOD_AUTH, RequestBin/binary>>),
%% 需要等待auth返回的结果
receive
{ssl, Socket, <<?PACKET_RESPONSE, PacketId:32, ?METHOD_AUTH, ReplyBin/binary>>} ->
lager:debug("[efka_agent] get a reply bin: ~p", [ReplyBin]),
{ok, message_pb:decode_msg(ReplyBin, auth_reply)}
after 5000 ->
{error, timeout}
end.
erlang:start_timer(5000, self(), create_transport).