Conn.Pool v0.2.1 Conn.Pool

Connection pool helps storing, sharing and using connections. It also make its possible to use the same connection concurrently. For example, if there exists remote API accessible via websocket, pool can provide shared access, making queues of calls. If connection is closed/expires — pool will reinitialize or drop it and awaiting calls will be moved to another connection queue.

Start pool via start_link/1 or start/1. Add connections via init/3 or Conn.init/2 → put!/2. Make calls from pool via call/4.

In the following examples, %Conn.Agent{} represents connection to some Agent that can be created separately or via Conn.init/2. Available methods of interaction with Agent are :get, :get_and_update, :update and :stop (see Conn.methods!/1). This type of connection means to exist only as an example for doctests. More meaningful example would be Conn.Plug wrapper. Also, see Conn docs for detailed example on Conn protocol implementation and use.

iex> {:ok, pool} = Conn.Pool.start_link()
iex> {:ok, agent} = Agent.start_link(fn -> 42 end)
iex> {:ok, id} = Conn.Pool.init(pool, %Conn.Agent{}, res: agent)
iex> Conn.Pool.extra(pool, id, type: :agent) # add `extra` info
{:ok, nil}
#
# Pool wraps conn into `%Conn{}` struct.
iex> {:ok, info} = Conn.Pool.info(pool, id)
iex> info.extra
[type: :agent]
iex> info.methods
[:get, :get_and_update, :update, :stop]
iex> ^agent = Conn.resource(info.conn)
iex> Conn.Pool.call(pool, agent, :get, & &1)
{:ok, 42}
#
# Now we use filter (& &1.extra[:type] != :agent).
iex> Conn.Pool.call(pool, agent, :get, & &1.extra[:type] != :agent, & &1)
{:error, :filter}
#
iex> Conn.Pool.call(pool, agent, :badmethod)
{:error, :method}
iex> Conn.Pool.call(pool, :badres, :badmethod)
{:error, :resource}

In the above example connection was initialized and used directly from pool. %Conn{}.extra information that was given via Conn.Pool.extra/3 was used to filter conn to be selected in Conn.Pool.call/5 call.

In the following example connection will be added via put/2.

iex> {:ok, pool} = Conn.Pool.start_link()
iex> {:ok, agent} = Agent.start_link(fn -> 42 end)
iex> {:ok, conn} = Conn.init(%Conn.Agent{}, res: agent)
iex> Conn.Pool.put!(pool, %Conn{conn: conn, revive: true})
iex> Process.alive?(agent) && not Conn.Pool.empty?(pool, agent)
true
# There are conns for resource `agent` in this pool.
iex> Conn.Pool.call(pool, agent, :stop)
:ok
iex> not Process.alive?(agent)
true
iex> :timer.sleep(10)
iex> Conn.Pool.empty?(pool, agent)
true
# Now conn is `:closed` and will be reinitialized by pool,
# but:
iex> {:error, :dead, :infinity, conn} == Conn.init(conn)
true
# [`Conn.init/2`](Conn.html#init/2) suggests to never reinitialize again
# (`:infinity` timeout) so pool will just drop this conn.
iex> Conn.Pool.resources(pool)
[]

Also, TTL value could be provided. By default expired connection is revived.

iex> {:ok, pool} = Conn.Pool.start_link()
iex> {:ok, agent} = Agent.start_link(fn -> 42 end)
iex> {:ok, id} = Conn.Pool.init(pool, %Conn.Agent{}, res: agent)
iex> Conn.Pool.update(pool, agent, & %{&1 | ttl: 50})
iex> {:ok, info} = Conn.Pool.info(pool, id)
iex> info.ttl
50
iex> Conn.Pool.call(pool, agent, :get, & &1)
{:ok, 42}
iex> :timer.sleep(50)
:ok
#
# Next call will fail because all the conns are expired and conn
# will not be revived (%Conn{}.revive == false).
#
iex> Conn.Pool.call(pool, agent, :get, & &1)
{:error, :resource}
iex> Conn.Pool.resources(pool)
[]
iex> Conn.Pool.init(pool, %Conn.Agent{}, res: agent)
iex> Conn.Pool.call(pool, agent, :get, & &1+1)
{:ok, 43}
# That's because conn that is alive exists.
#
iex> Conn.Pool.update(pool, agent, & %{&1 | ttl: 10, revive: true})
iex> :timer.sleep(10)
iex> Conn.Pool.call(pool, agent, :get, & &1+1)
{:error, :resource}

Name registration

An Conn.Pool is bound to the same name registration rules as GenServers. Read more about it in the GenServer docs.

Link to this section Summary

Types

filter()

id()

reason()

t()

Functions

call(pool, resource, method, payload \\ nil)

Select one of the connections to given resource and make Conn.call/3 via given method of interaction

call(pool, resource, method, filter, payload)

child_spec(arg)

Returns a specification to start this module under a supervisor

empty?(pool, resource)

Is pool has conns to the given resource?

extra(pool, id, extra)

:extra field of %Conn{} is intended for filtering conns while making call. Calling extra/3 will change :extra field of connection with given id, while returning the old value in form of {:ok, old extra} or :error if pool don’t known conn with such id

info(pool, id)

Retrives connection wraped in a %Conn{}. Returns {:ok, Conn.info} or :error

init(pool, info_or_conn, init_args \\ nil)

Initialize connection in a separate Task and use put/2 to add connection that is wrapped in %Conn{}

map(pool, resource, fun)

Applies given fun to every resource conn

pop(pool, id)

Deletes connection and returns corresponding %Conn{} struct

pop(pool, resource, filter)

Pops conns to resource that satisfy filter

put!(pool, info)

Adds given %Conn{} to pool, returning id to refer it later. This call always refresh data cached in %Conn{}.methods

resources(pool)

Returns all the known resources

start(opts \\ [])

Starts pool. See start_link/2 for details

start_link(opts \\ [])

Starts pool as a linked process

update(pool, resource, fun)

Update every conn to resource with fun. This function always returns :ok

update(pool, resource, filter, fun)

Updates connections to resource that satisfy given filter. This function always returns :ok

Link to this section Types

filter()

filter() :: (Conn.info() -> as_boolean(term()))

id()

id() :: pos_integer()

reason()

reason() :: any()

t()

t() :: GenServer.t()

Link to this section Functions

call(pool, resource, method, payload \\ nil)

call(Conn.Pool.t(), Conn.resource(), Conn.method(), any()) ::
  :ok
  | {:ok, Conn.reply()}
  | {:error, :resource | :method | :timeout | reason()}

Select one of the connections to given resource and make Conn.call/3 via given method of interaction.

Optional filter param could be provided in form of (%Conn{} -> as_boolean(term())) callback.

Pool respects refresh timeout value returned by Conn.call/3. After each call :timeout field of the corresponding %Conn{} struct is rewrited.

Returns

{:error, :resource} if there is no conns to given resource;
{:error, :method} if there exists conns to given resource but they does not provide given method of interaction;
{:error, :filter} if there is no conns satisfying filter;
{:error, :timeout} if Conn.call/3 returned {:error, :timeout, _} and there is no other connection capable to make this call;
and {:error, reason} in case of Conn.call/3 returned arbitrary error.

In case of Conn.call/3 returns {:error, :timeout | reason, _}, Conn.Pool will use time penalties series, defined per pool (see start_link/2) or per connection (see %Conn{} :penalties field).

{:ok, reply} | :ok in case of success.

call(pool, resource, method, filter, payload)

call(Conn.Pool.t(), Conn.resource(), Conn.method(), filter(), any()) ::
  :ok
  | {:ok, Conn.reply()}
  | {:error, :resource | :method | :filter | :timeout | reason()}

child_spec(arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

empty?(pool, resource)

empty?(Conn.Pool.t(), Conn.resource()) :: boolean()

Is pool has conns to the given resource?

extra(pool, id, extra)

extra(Conn.Pool.t(), id(), any()) :: {:ok, any()} | :error

Example

iex> {:ok, pool} = Conn.Pool.start_link()
iex> {:ok, agent} = Agent.start(fn -> 42 end)
iex> {:ok, id} = Conn.Pool.init(pool, %Conn.Agent{}, res: agent)
iex> Conn.Pool.extra(pool, id, :extra)
{:ok, nil}
iex> Conn.Pool.extra(pool, id, :some)
{:ok, :extra}
iex> badid = -1
iex> Conn.Pool.extra(pool, badid, :some)
:error
#
iex> filter = & &1.extra == :extra
iex> Conn.Pool.call(pool, agent, :get, filter, & &1)
{:error, :filter}
#
# but:
iex> Conn.Pool.call(pool, agent, :get, & &1.extra == :some, & &1)
{:ok, 42}

info(pool, id)

info(Conn.Pool.t(), id()) ::
  {:ok,
   %Conn{
     closed: term(),
     conn: term(),
     extra: term(),
     init_args: term(),
     last_call: term(),
     last_init: term(),
     methods: term(),
     revive: term(),
     stats: term(),
     timeout: term(),
     ttl: term(),
     unsafe: term()
   }}
  | :error

Retrives connection wraped in a %Conn{}. Returns {:ok, Conn.info} or :error.

Example

iex> {:ok, pool} = Conn.Pool.start_link()
iex> {:ok, id} = Conn.Pool.init(pool, %Conn.Agent{}, fn -> 42 end)
iex> {:ok, info} = Conn.Pool.info(pool, id)
iex> info.extra
nil
iex> Conn.Pool.extra(pool, id, :extra)
{:ok, nil}
iex> {:ok, info} = Conn.Pool.info(pool, id)
iex> info.extra
:extra

init(pool, info_or_conn, init_args \\ nil)

init(Conn.Pool.t(), Conn.t() | Conn.info(), any()) ::
  {:ok, id()} | {:error, :timeout | reason()}

Initialize connection in a separate Task and use put/2 to add connection that is wrapped in %Conn{}.

By default, init time is limited by 5000 ms and pool will revive closed connection using init_args provided.

Returns

{:ok, id} in case of Conn.init/2 returns {:ok, conn}, where id is the identifier that can be used to refer conn;
{:error, :timeout} if Conn.init/2 returns {:ok, :timeout};
{:error, reason} in case Conn.init/2 returns an arbitrary error.

map(pool, resource, fun)

map(Conn.Pool.t(), Conn.resource(), (Conn.info() -> a)) :: [a] when a: var

Applies given fun to every resource conn.

pop(pool, id)

pop(Conn.Pool.t(), Conn.id()) :: {:ok, Conn.info()} | :error

Deletes connection and returns corresponding %Conn{} struct.

Returns only after all calls awaiting execution on this conn are made.

Returns

{:ok, Conn struct} in case conn with such id exists;
:error — otherwise.

Examples

iex> {:ok, pool} = Conn.Pool.start_link()
iex> {:ok, agent} = Agent.start_link(fn -> 42 end)
iex> {:ok, id} = Conn.Pool.init(pool, %Conn.Agent{}, res: agent)
iex> Conn.Pool.call(pool, agent, :get, & &1)
{:ok, 42}
iex> {:ok, %Conn{conn: conn}} = Conn.Pool.pop(pool, id)
iex> Conn.Pool.pop(pool, id)
:error
iex> Conn.Pool.call(pool, agent, :get, & &1)
{:error, :resource}
# But, still:
iex> Agent.get(Conn.resource(conn), & &1)
42

Also, you can pop only those connections that satisfy the specified filter.

iex> {:ok, pool} = Conn.Pool.start_link()
iex> {:ok, agent} = Agent.start_link(fn -> 42 end)
iex> {:ok, id1} = Conn.Pool.init(pool, %Conn.Agent{}, res: agent)
iex> {:ok, id2} = Conn.Pool.init(pool, %Conn.Agent{}, res: agent)
iex> {:ok, id3} = Conn.Pool.init(pool, %Conn.Agent{}, res: agent)
iex> Conn.Pool.extra(pool, id1, :takeme)
{:ok, nil}
iex> Conn.Pool.extra(pool, id3, :takeme)
iex> Conn.Pool.pop(pool, agent, & &1.extra == :takeme)
iex> Conn.Pool.empty?(pool, agent)
false
iex> Conn.Pool.pop(pool, id2)
iex> :timer.sleep(10)
iex> Conn.Pool.empty?(pool, agent)
true

Also, see example for put!/2.

pop(pool, resource, filter)

pop(Conn.Pool.t(), Conn.resource(), filter()) :: [Conn.info()]

Pops conns to resource that satisfy filter.

put!(pool, info)

put!(Conn.Pool.t(), Conn.info()) :: id()

Adds given %Conn{} to pool, returning id to refer it later. This call always refresh data cached in %Conn{}.methods.

Example

iex> {:ok, pool} = Conn.Pool.start_link()
iex> {:ok, conn} = Conn.init(%Conn.Agent{}, fn -> 42 end)
iex> ttl = 100 # ms
iex> info = %Conn{
...>   conn: conn,
...>   extra: :extra,
...>   ttl: ttl
...> }
iex> id = Conn.Pool.put!(pool, info)
iex> {:ok, info} = Conn.Pool.info(pool, id)
iex> ^ttl = info.ttl
iex> {:ok, info} = Conn.Pool.info(pool, id)
iex> info.extra
:extra
#
# let's make some tweak
#
iex> {:ok, info} = Conn.Pool.pop(pool, id)
iex> id = Conn.Pool.put!(pool, %{info| ttl: :infinity})
iex> {:ok, info} = Conn.Pool.info(pool, id)
iex> info.ttl
:infinity

resources(pool)

resources(Conn.Pool.t()) :: [Conn.resource()]

Returns all the known resources.

start(opts \\ [])

start(GenServer.options()) :: GenServer.on_start()

Starts pool. See start_link/2 for details.

start_link(opts \\ [])

start_link(GenServer.options()) :: GenServer.on_start()

Starts pool as a linked process.

update(pool, resource, fun)

update(Conn.Pool.t(), Conn.resource(), (Conn.info() -> Conn.info())) :: :ok

Update every conn to resource with fun. This function always returns :ok.

Example

iex> {:ok, pool} = Conn.Pool.start_link()
iex> {:ok, agent} = Agent.start(fn -> 42 end)
iex> Conn.Pool.init(pool, %Conn.Agent{}, res: agent)
iex> Conn.Pool.init(pool, %Conn.Agent{}, res: agent)
iex> Conn.Pool.call(pool, agent, :get, & &1.extra == :extra, & &1)
{:error, :filter}
#
iex> Conn.Pool.update(pool, agent, & %{&1 | extra: :extra})
:ok
iex> Conn.Pool.call(pool, agent, :get, & &1.extra == :extra, & &1)
{:ok, 42}

update(pool, resource, filter, fun)

update(Conn.Pool.t(), Conn.resource(), filter(), (Conn.info() -> Conn.info())) ::
  :ok

Updates connections to resource that satisfy given filter. This function always returns :ok.

Example

iex> {:ok, pool} = Conn.Pool.start_link()
iex> {:ok, agent} = Agent.start(fn -> 42 end)
iex> Conn.Pool.init(pool, %Conn.Agent{}, res: agent)
iex> Conn.Pool.init(pool, %Conn.Agent{}, res: agent)
iex> filter = & &1.extra == :extra
iex> Conn.Pool.call(pool, agent, :get, filter, & &1)
{:error, :filter}
#
iex> Conn.Pool.update(pool, agent, & %{&1 | extra: :extra})
:ok
iex> Conn.Pool.call(pool, agent, :get, filter, & &1)
{:ok, 42}
iex> Conn.Pool.update(pool, agent, filter, &Map.put(&1, :extra, nil))
iex> Conn.Pool.call(pool, agent, :get, filter, & &1)
{:error, :filter}