yuanbiao
/
emqx
1
0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

optimize code structure

This commit is contained in:
付强 2019-08-04 19:31:49 +08:00 committed by Shawn
parent 111380bf25
commit b5cc357207
1 changed files with 24 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

43
src/ecpool_worker.erl
View File

@ -90,19 +90,15 @@ init([Pool, Id, Mod, Opts]) ->
State = #state{pool = Pool, id = Id, mod = Mod, opts = Opts, State = #state{pool = Pool, id = Id, mod = Mod, opts = Opts,
on_reconnect = proplists:get_value(on_reconnect, Opts), on_reconnect = proplists:get_value(on_reconnect, Opts),
on_disconnect = proplists:get_value(on_disconnect, Opts)}, on_disconnect = proplists:get_value(on_disconnect, Opts)},
case connect(State) of case connect_internal(State) of
{ok, Client} when is_pid(Client) -> {ok, NewState} ->
erlang:link(Client),
gproc_pool:connect_worker(ecpool:name(Pool), {Pool, Id}), gproc_pool:connect_worker(ecpool:name(Pool), {Pool, Id}),
{ok, State#state{client = Client, supervisees = [Client]}}; {ok, NewState};
{{ok, Client}, #{supervisees := SupPids} = _SupOpts} when is_list(SupPids) ->
[erlang:link(P) || P <- SupPids],
gproc_pool:connect_worker(ecpool:name(Pool), {Pool, Id}),
{ok, State#state{client = Client, supervisees = SupPids}};
{error, Error} -> {error, Error} ->
{stop, Error} {stop, Error}
end. end.
handle_call(is_connected, _From, State = #state{client = Client}) when is_pid(Client) -> handle_call(is_connected, _From, State = #state{client = Client}) when is_pid(Client) ->
{reply, Client =/= undefined andalso is_process_alive(Client), State}; {reply, Client =/= undefined andalso is_process_alive(Client), State};
@ -134,18 +130,13 @@ handle_info({'EXIT', Pid, Reason}, State = #state{opts = Opts, supervisees = Sup
end; end;
handle_info(reconnect, State = #state{opts = Opts, on_reconnect = OnReconnect}) -> handle_info(reconnect, State = #state{opts = Opts, on_reconnect = OnReconnect}) ->
case catch connect(State) of case connect_internal(State) of
{ok, Client} -> {ok, NewState = #state{client = Client}} ->
handle_reconnect(Client, OnReconnect), handle_reconnect(Client, OnReconnect),
erlang:link(Client), {noreply, NewState};
{noreply, State#state{client = Client, supervisees = [Client]}}; {Err, _Reason} when Err =:= error orelse Err =:= 'EXIT' ->
{{ok, Client}, #{supervisees := SupPids} = _SupOpts} -> reconnect(proplists:get_value(auto_reconnect, Opts), State)
handle_reconnect(Client, OnReconnect), end;
[erlang:link(P) || P <- SupPids],
{noreply, State#state{client = Client, supervisees = SupPids}};
{Err, _Reason} when Err =:= error orelse Err =:= 'EXIT' ->
reconnect(proplists:get_value(auto_reconnect, Opts), State)
end;
handle_info(_Info, State) -> handle_info(_Info, State) ->
{noreply, State}. {noreply, State}.
@ -194,3 +185,15 @@ handle_disconnect(_, undefined) ->
ok; ok;
handle_disconnect(Client, Disconnect) -> handle_disconnect(Client, Disconnect) ->
Disconnect(Client). Disconnect(Client).
connect_internal(State) ->
case connect(State) of
{ok, Client} when is_pid(Client) ->
erlang:link(Client),
{ok, State#state{client = Client, supervisees = [Client]}};
{{ok, Client}, #{supervisees := SupPids} = _SupOpts} when is_list(SupPids) ->
[erlang:link(P) || P <- SupPids],
{ok, State#state{client = Client, supervisees = SupPids}};
{error, Error} ->
{error, Error}
end.