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

0.3.1 - Improve the ecpool library for rule engine

This commit is contained in:
Feng Lee 2019-03-15 16:59:35 +08:00
parent 80cbf6cbb1
commit 640d6cc225
17 changed files with 422 additions and 442 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.gitignore vendored
View File

@ -11,3 +11,4 @@ rel/example_project
tmp
_build/
rebar.lock
logs/

20
CHANGES
View File

@ -1,20 +0,0 @@
0.2.1-beta (2016/04/13)
-----------------------
Update Copyright
Spec Syntax
Catch exceptions when reconnecting
0.2-beta (2016/01/02)
---------------------
Eunit tests and first release.
0.1-beta (2015/12/30)
---------------------
First commit.

3
Makefile
View File

@ -18,7 +18,7 @@ clean:
@$(REBAR) clean
test:
@$(REBAR) skip_deps=true eunit
@$(REBAR) skip_deps=true ct
edoc:
@$(REBAR) doc
@ -34,3 +34,4 @@ build_plt: compile
dialyzer: compile
dialyzer -Wno_return --plt $(PLT) deps/*/ebin ebin

16
README.md
View File

@ -1,16 +1,15 @@
# Erlang Connection/Client Pool
ecpool is different with worker-pool libraries in that it is designed to pool connection/clients to server or database.
ecpool tries to avoid the erlang application crash when the server or database going down.
`ecpool` is different with worker-pool libraries in that it is designed to pool connection/clients to server or database.
`ecpool` tries to avoid the erlang application crash when the server or database going down.
## Overview
A pool worker to manage/monitor the client to server or database:
A pool worker to manage/monitor the connection to server or database:
```
PoolWorker -> Client -> DB
PoolWorker -> Conn -> DB
```
Use client:
@ -19,7 +18,6 @@ Use client:
ecpool:with_client(Pool, fun(Client) -> call(Client) end).
```
## Usage
### Start the pool
@ -47,7 +45,6 @@ PgOpts = [%% Pool Size
{encoding, utf8}],
ecpool:start_pool(epgsql_pool, epgsql_pool_client, PgOpts)
```
### The Callback Module
@ -77,7 +74,7 @@ squery(Pool, Sql) ->
## Design
The ecpool supervisor tree:
The `ecpool` supervisor tree:
```
pool_sup[one_for_all supervisor]
@ -92,9 +89,8 @@ pool_sup[one_for_all supervisor]
Feng Lee <feng@emqx.io>
## License
The MIT License (MIT)
The Apache License Version 2.0

8
TODO
View File

@ -1,8 +0,0 @@
worker -> ecql
integrage with ecql
management api
unit tests

2
rebar.config
View File

@ -1,7 +1,7 @@
%% -*- mode: erlang;erlang-indent-level: 4;indent-tabs-mode: nil -*-
%% ex: ts=4 sw=4 ft=erlang et
{require_min_otp_vsn, "R17"}.
{require_min_otp_vsn, "R20"}.
%warnings_as_errors, warn_untyped_record,
{erl_opts, [

4
src/ecpool.app.src
View File

@ -1,13 +1,13 @@
{application, ecpool,
[
{description, "Erlang Client/Connection Pool"},
{vsn, "0.2"},
{vsn, "0.3.1"},
{registered, []},
{applications, [
kernel,
stdlib,
gproc
]},
{mod, { ecpool_app, []}},
{mod, {ecpool_app, []}},
{env, []}
]}.

82
src/ecpool.erl
View File

@ -1,47 +1,54 @@
%%% The MIT License (MIT)
%%%
%%% Copyright (c) 2018 EMQ X
%%%
%%% Permission is hereby granted, free of charge, to any person obtaining a copy
%%% of this software and associated documentation files (the "Software"), to deal
%%% in the Software without restriction, including without limitation the rights
%%% to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
%%% copies of the Software, and to permit persons to whom the Software is
%%% furnished to do so, subject to the following conditions:
%%%
%%% The above copyright notice and this permission notice shall be included in all
%%% copies or substantial portions of the Software.
%%%
%%% THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
%%% IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
%%% FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
%%% AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
%%% LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
%%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
%%% SOFTWARE.
%%%
%%%
%%% @author Feng Lee <feng@emqx.io>
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%%
%% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License.
%% You may obtain a copy of the License at
%%
%% http://www.apache.org/licenses/LICENSE-2.0
%%
%% Unless required by applicable law or agreed to in writing, software
%% distributed under the License is distributed on an "AS IS" BASIS,
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and
%% limitations under the License.
-module(ecpool).
-export([pool_spec/4, start_pool/3, start_sup_pool/3, stop_sup_pool/1,
get_client/1, get_client/2, with_client/2, with_client/3,
name/1, workers/1]).
-export([pool_spec/4,
start_pool/3,
start_sup_pool/3,
stop_sup_pool/1,
get_client/1,
get_client/2,
with_client/2,
with_client/3,
workers/1
]).
-type pool_type() :: random | hash | round_robin.
-export([name/1]).
-type option() :: {pool_size, pos_integer()}
-type(pool_name() :: atom()).
-type(pool_type() :: random | hash | round_robin).
-type(option() :: {pool_size, pos_integer()}
| {pool_type, pool_type()}
| {auto_reconnect, false | pos_integer()}
| tuple().
| tuple()).
-export_type([pool_name/0,
pool_type/0,
option/0
]).
pool_spec(ChildId, Pool, Mod, Opts) ->
{ChildId, {?MODULE, start_pool, [Pool, Mod, Opts]},
permanent, 5000, supervisor, [ecpool_pool_sup]}.
#{id => ChildId,
start => {?MODULE, start_pool, [Pool, Mod, Opts]},
restart => permanent,
shutdown => 5000,
type => supervisor,
modules => [ecpool_pool_sup]}.
%% @doc Start the pool
-spec(start_pool(atom(), atom(), [option()]) -> {ok, pid()} | {error, any()}).
%% @doc Start the pool sup.
-spec(start_pool(atom(), atom(), [option()]) -> {ok, pid()} | {error, term()}).
start_pool(Pool, Mod, Opts) when is_atom(Pool) ->
ecpool_pool_sup:start_link(Pool, Mod, Opts).
@ -79,9 +86,10 @@ with_worker(Worker, Fun) ->
{error, Reason} -> {error, Reason}
end.
%% @doc Pool workers
workers(Pool) ->
gproc_pool:active_workers(name(Pool)).
%% @doc ecpool name
name(Pool) -> {?MODULE, Pool}.
%% @doc pool workers
workers(Pool) -> gproc_pool:active_workers(name(Pool)).

37
src/ecpool_app.erl
View File

@ -1,27 +1,16 @@
%%% The MIT License (MIT)
%%%
%%% Copyright (c) 2018 EMQ X
%%%
%%% Permission is hereby granted, free of charge, to any person obtaining a copy
%%% of this software and associated documentation files (the "Software"), to deal
%%% in the Software without restriction, including without limitation the rights
%%% to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
%%% copies of the Software, and to permit persons to whom the Software is
%%% furnished to do so, subject to the following conditions:
%%%
%%% The above copyright notice and this permission notice shall be included in all
%%% copies or substantial portions of the Software.
%%%
%%% THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
%%% IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
%%% FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
%%% AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
%%% LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
%%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
%%% SOFTWARE.
%%%
%%%
%%% @author Feng Lee <feng@emqx.io>
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%%
%% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License.
%% You may obtain a copy of the License at
%%
%% http://www.apache.org/licenses/LICENSE-2.0
%%
%% Unless required by applicable law or agreed to in writing, software
%% distributed under the License is distributed on an "AS IS" BASIS,
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and
%% limitations under the License.
-module(ecpool_app).

102
src/ecpool_pool.erl
View File

@ -1,27 +1,16 @@
%%% The MIT License (MIT)
%%%
%%% Copyright (c) 2018 EMQ X
%%%
%%% Permission is hereby granted, free of charge, to any person obtaining a copy
%%% of this software and associated documentation files (the "Software"), to deal
%%% in the Software without restriction, including without limitation the rights
%%% to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
%%% copies of the Software, and to permit persons to whom the Software is
%%% furnished to do so, subject to the following conditions:
%%%
%%% The above copyright notice and this permission notice shall be included in all
%%% copies or substantial portions of the Software.
%%%
%%% THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
%%% IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
%%% FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
%%% AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
%%% LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
%%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
%%% SOFTWARE.
%%%
%%%
%%% @author Feng Lee <feng@emqx.io>
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%%
%% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License.
%% You may obtain a copy of the License at
%%
%% http://www.apache.org/licenses/LICENSE-2.0
%%
%% Unless required by applicable law or agreed to in writing, software
%% distributed under the License is distributed on an "AS IS" BASIS,
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and
%% limitations under the License.
-module(ecpool_pool).
@ -30,39 +19,47 @@
-import(proplists, [get_value/3]).
%% API Function Exports
-export([start_link/2, info/1]).
-export([start_link/2]).
-export([info/1]).
%% ------------------------------------------------------------------
%% gen_server Function Exports
%% ------------------------------------------------------------------
-export([init/1, handle_call/3, handle_cast/2, handle_info/2,
terminate/2, code_change/3]).
-export([init/1,
handle_call/3,
handle_cast/2,
handle_info/2,
terminate/2,
code_change/3
]).
-record(state, {name, size, type}).
%%%=============================================================================
%%% API
%%%=============================================================================
%%------------------------------------------------------------------------------
%% API
%%------------------------------------------------------------------------------
-spec(start_link(ecpool:pool_name(), list(ecpool:option()))
-> {ok, pid()} | {error, term()}).
start_link(Pool, Opts) ->
gen_server:start_link(?MODULE, [Pool, Opts], []).
-spec(info(pid()) -> list()).
info(Pid) ->
gen_server:call(Pid, info).
%%%=============================================================================
%%% gen_server callbacks
%%%=============================================================================
%%------------------------------------------------------------------------------
%% gen_server callbacks
%%------------------------------------------------------------------------------
init([Pool, Opts]) ->
Schedulers = erlang:system_info(schedulers),
PoolSize = get_value(pool_size, Opts, Schedulers),
PoolType = get_value(pool_type, Opts, random),
ensure_pool(ecpool:name(Pool), PoolType, [{size, PoolSize}]),
lists:foreach(fun(I) ->
ensure_pool_worker(ecpool:name(Pool), {Pool, I}, I)
end, lists:seq(1, PoolSize)),
ok = ensure_pool(ecpool:name(Pool), PoolType, [{size, PoolSize}]),
ok = lists:foreach(
fun(I) ->
ensure_pool_worker(ecpool:name(Pool), {Pool, I}, I)
end, lists:seq(1, PoolSize)),
{ok, #state{name = Pool, size = PoolSize, type = PoolType}}.
ensure_pool(Pool, Type, Opts) ->
@ -78,23 +75,30 @@ ensure_pool_worker(Pool, Name, Slot) ->
end.
handle_call(info, _From, State = #state{name = Pool, size = Size, type = Type}) ->
Info = [{pool_name, Pool}, {pool_size, Size},
{pool_type, Type}, {workers, ecpool:workers(Pool)}],
Workers = ecpool:workers(Pool),
Info = [{pool_name, Pool},
{pool_size, Size},
{pool_type, Type},
{workers, Workers}],
{reply, Info, State};
handle_call(_Request, _From, State) ->
{reply, {error, unexpected_req}, State}.
handle_call(Req, _From, State) ->
logger:error("[Pool] unexpected request: ~p", [Req]),
{reply, ignored, State}.
handle_cast(_Msg, State) ->
handle_cast(Msg, State) ->
logger:error("[Pool] unexpected msg: ~p", [Msg]),
{noreply, State}.
handle_info(_Info, State) ->
handle_info(Info, State) ->
logger:error("[Pool] unexpected info: ~p", [Info]),
{noreply, State}.
terminate(_Reason, #state{name = Pool, size = Size}) ->
lists:foreach(fun(I) ->
gproc_pool:remove_worker(ecpool:name(Pool), {Pool, I})
end, lists:seq(1, Size)),
lists:foreach(
fun(I) ->
gproc_pool:remove_worker(ecpool:name(Pool), {Pool, I})
end, lists:seq(1, Size)),
gproc_pool:delete(ecpool:name(Pool)).
code_change(_OldVsn, State, _Extra) ->

37
src/ecpool_pool_sup.erl
View File

@ -1,27 +1,16 @@
%%% The MIT License (MIT)
%%%
%%% Copyright (c) 2018 EMQ X
%%%
%%% Permission is hereby granted, free of charge, to any person obtaining a copy
%%% of this software and associated documentation files (the "Software"), to deal
%%% in the Software without restriction, including without limitation the rights
%%% to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
%%% copies of the Software, and to permit persons to whom the Software is
%%% furnished to do so, subject to the following conditions:
%%%
%%% The above copyright notice and this permission notice shall be included in all
%%% copies or substantial portions of the Software.
%%%
%%% THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
%%% IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
%%% FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
%%% AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
%%% LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
%%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
%%% SOFTWARE.
%%%
%%%
%%% @author Feng Lee <feng@emqx.io>
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%%
%% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License.
%% You may obtain a copy of the License at
%%
%% http://www.apache.org/licenses/LICENSE-2.0
%%
%% Unless required by applicable law or agreed to in writing, software
%% distributed under the License is distributed on an "AS IS" BASIS,
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and
%% limitations under the License.
-module(ecpool_pool_sup).

90
src/ecpool_sup.erl
View File

@ -1,47 +1,50 @@
%%% The MIT License (MIT)
%%%
%%% Copyright (c) 2018 EMQ X
%%%
%%% Permission is hereby granted, free of charge, to any person obtaining a copy
%%% of this software and associated documentation files (the "Software"), to deal
%%% in the Software without restriction, including without limitation the rights
%%% to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
%%% copies of the Software, and to permit persons to whom the Software is
%%% furnished to do so, subject to the following conditions:
%%%
%%% The above copyright notice and this permission notice shall be included in all
%%% copies or substantial portions of the Software.
%%%
%%% THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
%%% IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
%%% FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
%%% AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
%%% LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
%%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
%%% SOFTWARE.
%%%
%%%
%%% @author Feng Lee <feng@emqx.io>
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%%
%% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License.
%% You may obtain a copy of the License at
%%
%% http://www.apache.org/licenses/LICENSE-2.0
%%
%% Unless required by applicable law or agreed to in writing, software
%% distributed under the License is distributed on an "AS IS" BASIS,
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and
%% limitations under the License.
-module(ecpool_sup).
-behaviour(supervisor).
-export([start_link/0]).
%% API
-export([start_link/0, start_pool/3, stop_pool/1, pools/0, pool/1]).
-export([start_pool/3,
stop_pool/1,
get_pool/1
]).
-export([pools/0]).
%% Supervisor callbacks
-export([init/1]).
%% @doc Start supervisor.
-spec(start_link() -> {ok, pid()} | {error, any()}).
-spec(start_link() -> {ok, pid()} | {error, term()}).
start_link() ->
supervisor:start_link({local, ?MODULE}, ?MODULE, []).
%%------------------------------------------------------------------------------
%% Start/Stop a pool
%%------------------------------------------------------------------------------
%% @doc Start a pool.
-spec(start_pool(atom(), atom(), list(tuple())) -> {ok, pid()} | {error, term()}).
start_pool(Pool, Mod, Opts) when is_atom(Pool) ->
supervisor:start_child(?MODULE, pool_spec(Pool, Mod, Opts)).
-spec(stop_pool(Pool :: atom()) -> ok | {error, any()}).
%% @doc Stop a pool.
-spec(stop_pool(Pool :: atom()) -> ok | {error, term()}).
stop_pool(Pool) when is_atom(Pool) ->
ChildId = child_id(Pool),
case supervisor:terminate_child(?MODULE, ChildId) of
@ -51,32 +54,35 @@ stop_pool(Pool) when is_atom(Pool) ->
{error, Reason}
end.
%% @doc All Pools supervisored by ecpool_sup.
-spec(pools() -> [{atom(), pid()}]).
pools() ->
[{Pool, Pid} || {{pool_sup, Pool}, Pid, supervisor, _}
<- supervisor:which_children(?MODULE)].
%% @doc Find a pool.
-spec(pool(atom()) -> undefined | pid()).
pool(Pool) when is_atom(Pool) ->
%% @doc Get a pool.
-spec(get_pool(atom()) -> undefined | pid()).
get_pool(Pool) when is_atom(Pool) ->
ChildId = child_id(Pool),
case [Pid || {Id, Pid, supervisor, _} <- supervisor:which_children(?MODULE), Id =:= ChildId] of
[] -> undefined;
L -> hd(L)
end.
%%%=============================================================================
%%% Supervisor callbacks
%%%=============================================================================
%% @doc Get All Pools supervisored by the ecpool_sup.
-spec(pools() -> [{atom(), pid()}]).
pools() ->
[{Pool, Pid} || {{pool_sup, Pool}, Pid, supervisor, _}
<- supervisor:which_children(?MODULE)].
%%------------------------------------------------------------------------------
%% Supervisor callbacks
%%------------------------------------------------------------------------------
init([]) ->
{ok, { {one_for_one, 10, 100}, []} }.
pool_spec(Pool, Mod, Opts) ->
{child_id(Pool),
{ecpool_pool_sup, start_link, [Pool, Mod, Opts]},
transient, infinity, supervisor, [ecpool_pool_sup]}.
#{id => child_id(Pool),
start => {ecpool_pool_sup, start_link, [Pool, Mod, Opts]},
restart => transient,
shutdown => infinity,
type => supervisor,
modules => [ecpool_pool_sup]}.
child_id(Pool) -> {pool_sup, Pool}.

139
src/ecpool_worker.erl
View File

@ -1,48 +1,46 @@
%%% The MIT License (MIT)
%%%
%%% Copyright (c) 2018 EMQ X
%%%
%%% Permission is hereby granted, free of charge, to any person obtaining a copy
%%% of this software and associated documentation files (the "Software"), to deal
%%% in the Software without restriction, including without limitation the rights
%%% to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
%%% copies of the Software, and to permit persons to whom the Software is
%%% furnished to do so, subject to the following conditions:
%%%
%%% The above copyright notice and this permission notice shall be included in all
%%% copies or substantial portions of the Software.
%%%
%%% THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
%%% IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
%%% FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
%%% AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
%%% LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
%%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
%%% SOFTWARE.
%%%
%%%
%%% @author Feng Lee <feng@emqx.io>
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%%
%% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License.
%% You may obtain a copy of the License at
%%
%% http://www.apache.org/licenses/LICENSE-2.0
%%
%% Unless required by applicable law or agreed to in writing, software
%% distributed under the License is distributed on an "AS IS" BASIS,
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and
%% limitations under the License.
-module(ecpool_worker).
-behaviour(gen_server).
-export([start_link/4]).
%% API Function Exports
-export([start_link/4, client/1, is_connected/1]).
-export([client/1]).
-export([is_connected/1]).
%% gen_server Function Exports
-export([init/1, handle_call/3, handle_cast/2, handle_info/2,
terminate/2, code_change/3]).
-export([init/1,
handle_call/3,
handle_cast/2,
handle_info/2,
terminate/2,
code_change/3
]).
-record(state, {pool, id, client, mod, opts}).
%%%=============================================================================
%%% Callback
%%%=============================================================================
%%------------------------------------------------------------------------------
%% Callback
%%------------------------------------------------------------------------------
-ifdef(use_specs).
-callback connect(ConnOpts :: list()) -> {ok, pid()} | {error, any()}.
-callback(connect(ConnOpts :: list())
-> {ok, pid()} | {error, Reason :: term()}).
-else.
@ -50,15 +48,14 @@
behaviour_info(callbacks) ->
[{connect, 1}];
behaviour_info(_Other) ->
undefined.
-endif.
%%%=============================================================================
%%% API
%%%=============================================================================
%%------------------------------------------------------------------------------
%% API
%%------------------------------------------------------------------------------
%% @doc Start a pool worker.
-spec(start_link(atom(), pos_integer(), module(), list()) ->
@ -74,33 +71,40 @@ client(Pid) ->
%% @doc Is client connected?
-spec(is_connected(pid()) -> boolean()).
is_connected(Pid) ->
gen_server:call(Pid, is_connected).
gen_server:call(Pid, is_connected, infinity).
%%%=============================================================================
%%% gen_server callbacks
%%%=============================================================================
%%------------------------------------------------------------------------------
%% gen_server callbacks
%%------------------------------------------------------------------------------
init([Pool, Id, Mod, Opts]) ->
process_flag(trap_exit, true),
State = #state{pool = Pool, id = Id, mod = Mod, opts = Opts},
case connect(State) of
{ok, Client} ->
gproc_pool:connect_worker(ecpool:name(Pool), {Pool, Id}),
ok = maybe_apply(proplists:get_value(bind, Opts), self()),
true = gproc_pool:connect_worker(ecpool:name(Pool), {Pool, Id}),
{ok, State#state{client = Client}};
{error, Error} ->
{stop, Error}
end.
handle_call(is_connected, _From, State = #state{client = Client}) ->
{reply, Client =/= undefined andalso is_process_alive(Client), State};
IsAlive = Client =/= undefined andalso is_process_alive(Client),
{reply, IsAlive, State};
handle_call(client, _From, State = #state{client = undefined}) ->
{reply, {error, disconnected}, State};
handle_call(client, _From, State = #state{client = Client}) ->
{reply, {ok, Client}, State}.
{reply, {ok, Client}, State};
handle_cast(_Msg, State) ->
handle_call(Req, _From, State) ->
logger:error("[PoolWorker] unexpected call: ~p", [Req]),
{reply, ignored, State}.
handle_cast(Msg, State) ->
logger:error("[PoolWorker] unexpected cast: ~p", [Msg]),
{noreply, State}.
handle_info({'EXIT', Pid, Reason}, State = #state{client = Pid, opts = Opts}) ->
@ -108,48 +112,61 @@ handle_info({'EXIT', Pid, Reason}, State = #state{client = Pid, opts = Opts}) ->
false ->
{stop, Reason, State};
Secs ->
reconnect(Secs, State)
reconnect_after(Secs, State)
end;
handle_info(reconnect, State = #state{opts = Opts}) ->
handle_info(reconnect, State = #state{pool= Pool, opts = Opts}) ->
try connect(State) of
{ok, Client} ->
{noreply, State#state{client = Client}};
{error, _Reason} ->
reconnect(proplists:get_value(auto_reconnect, Opts), State)
{error, Reason} ->
logger:error("[PoolWorker] ~p reconnect error: ~p", [Pool, Reason]),
reconnect_after(proplists:get_value(auto_reconnect, Opts), State)
catch
_Error:_Reason ->
reconnect(proplists:get_value(auto_reconnect, Opts), State)
_Error:Reason ->
logger:error("[PoolWorker] ~p reconnect error: ~p", [Pool, Reason]),
reconnect_after(proplists:get_value(auto_reconnect, Opts), State)
end;
handle_info(_Info, State) ->
handle_info(Info, State) ->
logger:error("[PoolWorker] unexpected info: ~p", [Info]),
{noreply, State}.
terminate(_Reason, #state{pool = Pool, id = Id}) ->
terminate(_Reason, #state{pool = Pool, id = Id, opts = Opts}) ->
ok = maybe_apply(proplists:get_value(unbind, Opts), self()),
gproc_pool:disconnect_worker(ecpool:name(Pool), {Pool, Id}).
code_change(_OldVsn, State, _Extra) ->
{ok, State}.
%%%=============================================================================
%%% Internal Functions
%%%=============================================================================
%%------------------------------------------------------------------------------
%% Internal Functions
%%------------------------------------------------------------------------------
connect(#state{mod = Mod, opts = Opts}) ->
Mod:connect(connopts(Opts, [])).
connopts([], Acc) ->
Acc;
connopts([{pool_size, _} | Opts], Acc) ->
connopts([{pool_size, _}|Opts], Acc) ->
connopts(Opts, Acc);
connopts([{pool_type, _} | Opts], Acc) ->
connopts([{pool_type, _}|Opts], Acc) ->
connopts(Opts, Acc);
connopts([{auto_reconnect, _} | Opts], Acc) ->
connopts([{auto_reconnect, _}|Opts], Acc) ->
connopts(Opts, Acc);
connopts([Opt | Opts], Acc) ->
connopts(Opts, [Opt | Acc]).
connopts([{bind, _}|Opts], Acc) ->
connopts(Opts, Acc);
connopts([{unbind, _}|Opts], Acc) ->
connopts(Opts, Acc);
connopts([Opt|Opts], Acc) ->
connopts(Opts, [Opt|Acc]).
reconnect(Secs, State) ->
erlang:send_after(timer:seconds(Secs), self(), reconnect),
reconnect_after(Secs, State) ->
_ = erlang:send_after(timer:seconds(Secs), self(), reconnect),
{noreply, State#state{client = undefined}}.
maybe_apply(undefined, _Self) ->
ok;
maybe_apply(Fun, Self) ->
Fun(Self).

51
src/ecpool_worker_sup.erl
View File

@ -1,42 +1,37 @@
%%% The MIT License (MIT)
%%%
%%% Copyright (c) 2018 EMQ X
%%%
%%% Permission is hereby granted, free of charge, to any person obtaining a copy
%%% of this software and associated documentation files (the "Software"), to deal
%%% in the Software without restriction, including without limitation the rights
%%% to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
%%% copies of the Software, and to permit persons to whom the Software is
%%% furnished to do so, subject to the following conditions:
%%%
%%% The above copyright notice and this permission notice shall be included in all
%%% copies or substantial portions of the Software.
%%%
%%% THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
%%% IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
%%% FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
%%% AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
%%% LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
%%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
%%% SOFTWARE.
%%%
%%%
%%% @author Feng Lee <feng@emqx.io>
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%%
%% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License.
%% You may obtain a copy of the License at
%%
%% http://www.apache.org/licenses/LICENSE-2.0
%%
%% Unless required by applicable law or agreed to in writing, software
%% distributed under the License is distributed on an "AS IS" BASIS,
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and
%% limitations under the License.
-module(ecpool_worker_sup).
-behaviour(supervisor).
-export([start_link/3, init/1]).
-export([start_link/3]).
-export([init/1]).
start_link(Pool, Mod, Opts) when is_atom(Pool) ->
supervisor:start_link(?MODULE, [Pool, Mod, Opts]).
init([Pool, Mod, Opts]) ->
WorkerSpec = fun(Id) ->
{{worker, Id}, {ecpool_worker, start_link, [Pool, Id, Mod, Opts]},
transient, 5000, worker, [ecpool_worker, Mod]}
end,
#{id => {worker, Id},
start => {ecpool_worker, start_link, [Pool, Id, Mod, Opts]},
restart => transient,
shutdown => 5000,
type => worker,
modules => [ecpool_worker, Mod]}
end,
Workers = [WorkerSpec(I) || I <- lists:seq(1, pool_size(Opts))],
{ok, { {one_for_one, 10, 60}, Workers} }.

107
test/ecpool_SUITE.erl Normal file
View File

@ -0,0 +1,107 @@
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%%
%% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License.
%% You may obtain a copy of the License at
%%
%% http://www.apache.org/licenses/LICENSE-2.0
%%
%% Unless required by applicable law or agreed to in writing, software
%% distributed under the License is distributed on an "AS IS" BASIS,
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and
%% limitations under the License.
-module(ecpool_SUITE).
-compile(export_all).
-include_lib("eunit/include/eunit.hrl").
-define(POOL, test_pool).
-define(POOL_OPTS, [
%% schedulers number
{pool_size, 10},
%% round-robbin | random | hash
{pool_type, random},
%% false | pos_integer()
{auto_reconnect, false},
%% DB Parameters
{host, "localhost"},
{port, 5432},
{username, "feng"},
{password, ""},
{database, "mqtt"},
{encoding, utf8}]).
all() ->
[{group, all}].
groups() ->
[{all, [sequence],
[t_start_pool,
t_start_sup_pool,
t_restart_client,
t_reconnect_client
]}].
init_per_suite(Config) ->
{ok, _} = application:ensure_all_started(ecpool),
Config.
end_per_suite(_Config) ->
ok = application:stop(ecpool),
ok = application:stop(gproc).
t_start_pool(_Config) ->
ecpool:start_pool(?POOL, test_client, ?POOL_OPTS),
?assertEqual(10, length(ecpool:workers(test_pool))),
?debugFmt("~p~n", [ecpool:workers(test_pool)]),
lists:foreach(fun(I) ->
ecpool:with_client(?POOL, fun(Client) ->
?debugFmt("Call ~p: ~p~n", [I, Client])
end)
end, lists:seq(1, 10)).
t_start_sup_pool(_Config) ->
{ok, Pid1} = ecpool:start_sup_pool(xpool, test_client, ?POOL_OPTS),
{ok, Pid2} = ecpool:start_sup_pool(ypool, test_client, ?POOL_OPTS),
?assertEqual([{xpool, Pid1}, {ypool, Pid2}], lists:sort(ecpool_sup:pools())),
ecpool:stop_sup_pool(ypool),
ecpool:stop_sup_pool(xpool),
?assertEqual([], ecpool_sup:pools()).
t_restart_client(_Config) ->
ecpool:start_pool(?POOL, test_client, [{pool_size, 4}]),
?assertEqual(4, length(ecpool:workers(?POOL))),
ecpool:with_client(?POOL, fun(Client) -> test_client:stop(Client, normal) end),
?debugFmt("~n~p~n", [ecpool:workers(?POOL)]),
?assertEqual(3, length(ecpool:workers(?POOL))),
ecpool:with_client(?POOL, fun(Client) -> test_client:stop(Client, {shutdown, x}) end),
?debugFmt("~n~p~n", [ecpool:workers(?POOL)]),
?assertEqual(2, length(ecpool:workers(?POOL))),
ecpool:with_client(?POOL, fun(Client) ->
test_client:stop(Client, {shutdown, badarg})
end),
timer:sleep(100),
?debugFmt("~n~p~n", [ecpool:workers(?POOL)]),
?assertEqual(2, length(ecpool:workers(?POOL))).
t_reconnect_client(_Config) ->
ecpool:start_pool(?POOL, test_client, [{pool_size, 4}, {auto_reconnect, 1}]),
?assertEqual(4, length(ecpool:workers(?POOL))),
ecpool:with_client(?POOL, fun(Client) ->
test_client:stop(Client, normal)
end),
?assert(lists:member(false, [ecpool_worker:is_connected(Pid) || {_, Pid} <- ecpool:workers(?POOL)])),
timer:sleep(1100),
?assertNot(lists:member(false, [ecpool_worker:is_connected(Pid) || {_, Pid} <- ecpool:workers(?POOL)])),
ecpool:with_client(?POOL, fun(Client) ->
test_client:stop(Client, {shutdown, badarg})
end),
?assert(lists:member(false, [ecpool_worker:is_connected(Pid) || {_, Pid} <- ecpool:workers(?POOL)])),
timer:sleep(1100),
?assertEqual(4, length(ecpool:workers(?POOL))).

110
test/ecpool_test.erl
View File

@ -1,110 +0,0 @@
%%% The MIT License (MIT)
%%%
%%% Copyright (c) 2018 EMQ X
%%%
%%% Permission is hereby granted, free of charge, to any person obtaining a copy
%%% of this software and associated documentation files (the "Software"), to deal
%%% in the Software without restriction, including without limitation the rights
%%% to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
%%% copies of the Software, and to permit persons to whom the Software is
%%% furnished to do so, subject to the following conditions:
%%%
%%% The above copyright notice and this permission notice shall be included in all
%%% copies or substantial portions of the Software.
%%%
%%% THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
%%% IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
%%% FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
%%% AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
%%% LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
%%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
%%% SOFTWARE.
%%%
%%%
%%% @author Feng Lee <feng@emqx.io>
-module(ecpool_test).
-ifdef(TEST).
-include_lib("eunit/include/eunit.hrl").
-define(POOL, test_pool).
-define(POOL_OPTS, [
%% schedulers number
{pool_size, 10},
%% round-robbin | random | hash
{pool_type, random},
%% false | pos_integer()
{auto_reconnect, false},
%% DB Parameters
{host, "localhost"},
{port, 5432},
{username, "feng"},
{password, ""},
{database, "mqtt"},
{encoding, utf8}]).
pool_test_() ->
{foreach,
fun() ->
application:start(gproc),
application:start(ecpool)
end,
fun(_) ->
application:stop(ecpool),
application:stop(gproc)
end,
[?_test(t_start_pool()),
?_test(t_start_sup_pool()),
?_test(t_restart_client()),
?_test(t_reconnect_client())]}.
t_start_pool() ->
ecpool:start_pool(?POOL, test_client, ?POOL_OPTS),
?assertEqual(10, length(ecpool:workers(test_pool))),
?debugFmt("~p~n", [ecpool:workers(test_pool)]),
lists:foreach(fun(I) ->
ecpool:with_client(?POOL, fun(Client) ->
?debugFmt("Call ~p: ~p~n", [I, Client])
end)
end, lists:seq(1, 10)).
t_start_sup_pool() ->
{ok, Pid1} = ecpool:start_sup_pool(xpool, test_client, ?POOL_OPTS),
{ok, Pid2} = ecpool:start_sup_pool(ypool, test_client, ?POOL_OPTS),
?assertEqual([{xpool, Pid1}, {ypool, Pid2}], lists:sort(ecpool_sup:pools())),
ecpool:stop_sup_pool(ypool),
ecpool:stop_sup_pool(xpool),
?assertEqual([], ecpool_sup:pools()).
t_restart_client() ->
ecpool:start_pool(?POOL, test_client, [{pool_size, 4}]),
?assertEqual(4, length(ecpool:workers(?POOL))),
ecpool:with_client(?POOL, fun(Client) -> test_client:stop(Client, normal) end),
?debugFmt("~n~p~n", [ecpool:workers(?POOL)]),
?assertEqual(3, length(ecpool:workers(?POOL))),
ecpool:with_client(?POOL, fun(Client) -> test_client:stop(Client, {shutdown, x}) end),
?debugFmt("~n~p~n", [ecpool:workers(?POOL)]),
?assertEqual(2, length(ecpool:workers(?POOL))),
ecpool:with_client(?POOL, fun(Client) -> test_client:stop(Client, badarg) end),
timer:sleep(100),
?debugFmt("~n~p~n", [ecpool:workers(?POOL)]),
?assertEqual(2, length(ecpool:workers(?POOL))).
t_reconnect_client() ->
ecpool:start_pool(?POOL, test_client, [{pool_size, 4}, {auto_reconnect, 1}]),
?assertEqual(4, length(ecpool:workers(?POOL))),
ecpool:with_client(?POOL, fun(Client) -> test_client:stop(Client, normal) end),
?assert(lists:member(false, [ecpool_worker:is_connected(Pid) || {_, Pid} <- ecpool:workers(?POOL)])),
timer:sleep(1100),
?assertNot(lists:member(false, [ecpool_worker:is_connected(Pid) || {_, Pid} <- ecpool:workers(?POOL)])),
ecpool:with_client(?POOL, fun(Client) -> test_client:stop(Client, badarg) end),
?assert(lists:member(false, [ecpool_worker:is_connected(Pid) || {_, Pid} <- ecpool:workers(?POOL)])),
timer:sleep(1100),
?assertEqual(4, length(ecpool:workers(?POOL))).
-endif.

49
test/test_client.erl
View File

@ -1,27 +1,36 @@
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
%%
%% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License.
%% You may obtain a copy of the License at
%%
%% http://www.apache.org/licenses/LICENSE-2.0
%%
%% Unless required by applicable law or agreed to in writing, software
%% distributed under the License is distributed on an "AS IS" BASIS,
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and
%% limitations under the License.
-module(test_client).
-behaviour(ecpool_worker).
-behaviour(gen_server).
-define(SERVER, ?MODULE).
%% ------------------------------------------------------------------
%% API Function Exports
%% ------------------------------------------------------------------
-export([connect/1,
stop/2
]).
-export([connect/1, stop/2]).
%% ------------------------------------------------------------------
%% gen_server Function Exports
%% ------------------------------------------------------------------
-export([init/1, handle_call/3, handle_cast/2, handle_info/2,
terminate/2, code_change/3]).
%% ------------------------------------------------------------------
%% API Function Definitions
%% ------------------------------------------------------------------
-export([init/1,
handle_call/3,
handle_cast/2,
handle_info/2,
terminate/2,
code_change/3
]).
connect(Opts) ->
gen_server:start_link(?MODULE, [Opts], []).
@ -29,9 +38,9 @@ connect(Opts) ->
stop(Pid, Reason) ->
gen_server:call(Pid, {stop, Reason}).
%% ------------------------------------------------------------------
%%-----------------------------------------------------------------------------
%% gen_server Function Definitions
%% ------------------------------------------------------------------
%%-----------------------------------------------------------------------------
init(Args) ->
{ok, Args}.
@ -39,7 +48,7 @@ init(Args) ->
handle_call({stop, Reason}, _From, State) ->
{stop, Reason, ok, State};
handle_call(_Request, _From, State) ->
handle_call(_Req, _From, State) ->
{reply, ok, State}.
handle_cast(_Msg, State) ->
@ -54,7 +63,3 @@ terminate(_Reason, _State) ->
code_change(_OldVsn, State, _Extra) ->
{ok, State}.
%% ------------------------------------------------------------------
%% Internal Function Definitions
%% ------------------------------------------------------------------