apps/efka/src/efka_agent.erl

@@ -19,14 +19,10 @@
 %% gen_server callbacks
 -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]).
 
--define(SERVER, ?MODULE).
-
--define(STATE_DENIED, denied).
--define(STATE_ACTIVATED, activated).
+-define(SERVER, ?MODULE). 
 
 -record(state, {
-    transport_pid :: undefined | pid(),
-    status = ?STATE_DENIED
+    transport_pid :: undefined | pid()
 }).
 
 %%%===================================================================
@@ -55,7 +51,8 @@ init([]) ->
 
 %% @private
 %% @doc Handling call messages
--spec(handle_call(Request :: term(), From :: {pid(), Tag :: term()}, State :: #state{}) ->
+-spec(handle_call(Request :: term(), From :: {pid(), Tag :: term()},
+        State :: #state{}) ->
     {reply, Reply :: term(), NewState :: #state{}} |
     {reply, Reply :: term(), NewState :: #state{}, timeout() | hibernate} |
     {noreply, NewState :: #state{}} |
@@ -81,37 +78,12 @@ handle_cast(_Request, State = #state{}) ->
     {noreply, NewState :: #state{}, timeout() | hibernate} |
     {stop, Reason :: term(), NewState :: #state{}}).
 handle_info({timeout, _, create_transport}, State = #state{}) ->
-    case efka_transport:start_link(self()) of
-        {ok, TransportPid} ->
-            Ref = efka_transport:auth_request(TransportPid, 5000),
-            receive
-                 %% 验证通过
-                {auth_reply, Ref, {ok, #auth_reply{code = 1, message = Message, repository_url = RepositoryUrl}}} ->
-                    lager:debug("[efka_agent] auth result: ~p, repository_url: ~p", [Message, RepositoryUrl]),
-                    {noreply, State#state{transport_pid = TransportPid, status = ?STATE_ACTIVATED}};
+    {ok, TransportPid} = efka_transport:start_link(self()),
+    {noreply, State#state{transport_pid = TransportPid}};
 
-                %% 主机denied状态
-                {auth_reply, Ref, {ok, #auth_reply{code = -1, message = Message, repository_url = RepositoryUrl}}} ->
-                    lager:debug("[efka_agent] auth failed, message: ~p, repository_url: ~p", [Message, RepositoryUrl]),
-                    {noreply, State};
-
-                %% 验证不通过
-                {auth_reply, Ref, {ok, #auth_reply{code = -2, message = Message, repository_url = RepositoryUrl}}} ->
-                    lager:debug("[efka_agent] auth failed, message: ~p, repository_url: ~p", [Message, RepositoryUrl]),
-                    {noreply, State};
-
-                {auth_reply, Ref, {error, Reason}} ->
-                    lager:debug("[efka_agent] auth_request failed, error: ~p", [Reason]),
-                    {noreply, State}
-            end;
-        {error, Reason} ->
-            lager:warning("[efka_agent] connect get error: ~p", [Reason]),
-            {noreply, State}
-    end;
-
-handle_info({'EXIT', Pid, Reason}, State = #state{}) ->
-    lager:debug("[efka_agent] transport pid: ~p, exit with reason: ~p", [Pid, Reason]),
-    erlang:start_timer(5000, self(), create_transport),
+handle_info({'EXIT', _Pid, Reason}, State = #state{}) ->
+    lager:debug("[efka_agent] transport exit with reason: ~p", [Reason]),
+    retry_connect(),
     {noreply, State#state{transport_pid = undefined}};
 
 handle_info(Info, State = #state{}) ->
@@ -138,4 +110,7 @@ code_change(_OldVsn, State = #state{}, _Extra) ->
 
 %%%===================================================================
 %%% Internal functions
-%%%===================================================================
+%%%===================================================================
+
+retry_connect() ->
+    erlang:start_timer(5000, self(), create_transport).

apps/efka/src/efka_transport.erl

@@ -15,7 +15,7 @@
 
 %% API
 -export([start_link/1]).
--export([auth_request/2, send/3, response/3, stop/1]).
+-export([auth_request/2, send/3, response/3]).
 
 %% gen_server callbacks
 -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]).
@@ -34,11 +34,8 @@
 %%% API
 %%%===================================================================
 
--spec auth_request(Pid :: pid(), Timeout :: integer()) -> Ref :: reference().
 auth_request(Pid, Timeout) when is_pid(Pid), is_integer(Timeout) ->
-    Ref = make_ref(),
-    gen_server:cast(Pid, {auth_request, self(), Ref, Timeout}),
-    Ref.
+    gen_server:call(Pid, {auth_request, Timeout}).
 
 send(Pid, Method, Packet) when is_pid(Pid), is_integer(Method), is_binary(Packet) ->
     gen_server:cast(Pid, {send, Method, Packet}).
@@ -46,9 +43,6 @@ send(Pid, Method, Packet) when is_pid(Pid), is_integer(Method), is_binary(Packet
 response(Pid, PacketId, Response) when is_pid(Pid), is_integer(PacketId) ->
     gen_server:cast(Pid, {response, PacketId, Response}).
 
-stop(Pid) when is_pid(Pid) ->
-    gen_server:stop(Pid).
-
 %% @doc Spawns the server and registers the local name (unique)
 -spec(start_link(ParentPid :: pid()) ->
     {ok, Pid :: pid()} | ignore | {error, Reason :: term()}).
@@ -72,15 +66,12 @@ init([ParentPid]) ->
     SslOptions = [
         binary,
         {packet, 4},
+        {active, true},
         {verify, verify_none}
     ],
-    case ssl:connect(Host, Port, SslOptions, 5000) of
-        {ok, Socket} ->
-            ssl:controlling_process(Socket, self()),
-            {ok, #state{parent_pid = ParentPid, host = Host, port = Port, socket = Socket}};
-        {error, Reason} ->
-            {stop, Reason}
-    end.
+    {ok, Socket} = ssl:connect(Host, Port, SslOptions, 5000),
+
+    {ok, #state{parent_pid = ParentPid, host = Host, port = Port, socket = Socket}}.
 
 %% @private
 %% @doc Handling call messages
@@ -92,16 +83,7 @@ init([ParentPid]) ->
     {noreply, NewState :: #state{}, timeout() | hibernate} |
     {stop, Reason :: term(), Reply :: term(), NewState :: #state{}} |
     {stop, Reason :: term(), NewState :: #state{}}).
-handle_call(_Req, _From, State = #state{}) ->
-    {reply, ok, State#state{}}.
-
-%% @private
-%% @doc Handling cast messages
--spec(handle_cast(Request :: term(), State :: #state{}) ->
-    {noreply, NewState :: #state{}} |
-    {noreply, NewState :: #state{}, timeout() | hibernate} |
-    {stop, Reason :: term(), NewState :: #state{}}).
-handle_cast({auth_request, ReceiverPid, Ref, Timeout}, State = #state{socket = Socket, packet_id = PacketId}) ->
+handle_call({auth_request, Timeout}, _From, State = #state{socket = Socket, packet_id = PacketId}) ->
     {ok, AuthInfo} = application:get_env(efka, auth),
     UUID = proplists:get_value(uuid, AuthInfo),
     Username = proplists:get_value(username, AuthInfo),
@@ -120,13 +102,19 @@ handle_cast({auth_request, ReceiverPid, Ref, Timeout}, State = #state{socket = S
     %% 需要等待auth返回的结果
     receive
         {ssl, Socket, <<?PACKET_RESPONSE, PacketId:32, ?METHOD_AUTH, ReplyBin/binary>>} ->
-            ReceiverPid ! {auth_reply, Ref, {ok, message_pb:decode_msg(ReplyBin, auth_reply)}},
-            {noreply, State#state{packet_id = PacketId + 1}}
-    after Timeout ->
-        ReceiverPid ! {auth_reply, Ref, {error, timeout}},
-        {stop, normal, State}
-    end;
+            lager:debug("[efka_agent] get a reply bin: ~p", [ReplyBin]),
+            {reply, {ok, message_pb:decode_msg(ReplyBin, auth_reply)}, State#state{packet_id = PacketId + 1}}
+    after
+        Timeout ->
+            {reply, {error, timeout}, State}
+    end.
 
+%% @private
+%% @doc Handling cast messages
+-spec(handle_cast(Request :: term(), State :: #state{}) ->
+    {noreply, NewState :: #state{}} |
+    {noreply, NewState :: #state{}, timeout() | hibernate} |
+    {stop, Reason :: term(), NewState :: #state{}}).
 handle_cast({send, Method, Packet}, State = #state{socket = Socket}) ->
     ok = ssl:send(Socket, <<?PACKET_REQUEST, 0:32, Method, Packet/binary>>),
     {noreply, State};
@@ -153,14 +141,13 @@ handle_info({ssl, Socket, <<?PACKET_PUBLISH, PacketId:32, Msg/binary>>}, State =
     ParentPid ! {efka_message, PacketId, Msg},
     {noreply, State#state{}};
 
-handle_info({ssl_error, Socket, _Reason}, State = #state{socket = Socket}) ->
-    {stop, normal, State};
+handle_info({ssl_error, Socket, Reason}, State = #state{socket = Socket}) ->
+    {stop, Reason, State};
 
 handle_info({ssl_closed, Socket}, State = #state{socket = Socket}) ->
-    {stop, normal, State};
+    {stop, ssl_closed, State};
 
-handle_info(Info, State = #state{}) ->
-    lager:debug("[efka_transport] info: ~p", [Info]),
+handle_info(_Info, State = #state{}) ->
     {noreply, State}.
 
 %% @private