View Source DBConnection behaviour (db_connection v2.5.0)

A behaviour module for implementing efficient database connection client processes, pools and transactions.

DBConnection handles callbacks differently to most behaviours. Some callbacks will be called in the calling process, with the state copied to and from the calling process. This is useful when the data for a request is large and means that a calling process can interact with a socket directly.

A side effect of this is that query handling can be written in a simple blocking fashion, while the connection process itself will remain responsive to OTP messages and can enqueue and cancel queued requests.

If a request or series of requests takes too long to handle in the client process a timeout will trigger and the socket can be cleanly disconnected by the connection process.

If a calling process waits too long to start its request it will timeout and its request will be cancelled. This prevents requests building up when the database can not keep up.

If no requests are received for an idle interval, the pool will ping all stale connections which can then ping the database to keep the connection alive.

Should the connection be lost, attempts will be made to reconnect with (configurable) exponential random backoff to reconnect. All state is lost when a connection disconnects but the process is reused.

The DBConnection.Query protocol provide utility functions so that queries can be encoded and decoded without blocking the connection or pool.

connection-pools
Connection pools

DBConnection connections support using different pools via the :pool option passed to start_link/1. The default pool is DBConnection.ConnectionPool. Another supported pool that is commonly used for tests is DBConnection.Ownership.

For now, using custom pools is not supported since the API for pools is not public.

errors
Errors

Most functions in this module raise a DBConnection.ConnectionError exception when failing to check out a connection from the pool.

Link to this section Summary

Types

conn()

cursor()

option()

params()

query()

result()

start_option()

status()

t()

Run or transaction connection reference.

Callbacks

checkout(state)

Checkouts the state from the connection process. Return {:ok, state} to allow the checkout or {:disconnect, exception, state} to disconnect.

connect(opts)

Connect to the database. Return {:ok, state} on success or {:error, exception} on failure.

disconnect(err, state)

Disconnect from the database. Return :ok.

handle_begin(opts, state)

Handle the beginning of a transaction.

handle_close(query, opts, state)

Close a query prepared by handle_prepare/3 with the database. Return {:ok, result, state} on success and to continue, {:error, exception, state} to return an error and continue, or {:disconnect, exception, state} to return an error and disconnect.

handle_commit(opts, state)

Handle committing a transaction. Return {:ok, result, state} on successfully committing transaction, {status, state} to notify caller that the transaction can not commit due to the transaction status status, or {:disconnect, exception, state} to error and disconnect.

handle_deallocate(query, cursor, opts, state)

Deallocate a cursor declared by handle_declare/4 with the database. Return {:ok, result, state} on success and to continue, {:error, exception, state} to return an error and continue, or {:disconnect, exception, state} to return an error and disconnect.

handle_declare(query, params, opts, state)

Declare a cursor using a query prepared by handle_prepare/3. Return {:ok, query, cursor, state} to return altered query query and cursor cursor for a stream and continue, {:error, exception, state} to return an error and continue or {:disconnect, exception, state} to return an error and disconnect.

handle_execute(query, params, opts, state)

Execute a query prepared by handle_prepare/3. Return {:ok, query, result, state} to return altered query query and result result and continue, {:error, exception, state} to return an error and continue or {:disconnect, exception, state} to return an error and disconnect.

handle_fetch(query, cursor, opts, state)

Fetch the next result from a cursor declared by handle_declare/4. Return {:cont, result, state} to return the result result and continue using cursor, {:halt, result, state} to return the result result and close the cursor, {:error, exception, state} to return an error and close the cursor, {:disconnect, exception, state} to return an error and disconnect.

handle_prepare(query, opts, state)

Prepare a query with the database. Return {:ok, query, state} where query is a query to pass to execute/4 or close/3, {:error, exception, state} to return an error and continue or {:disconnect, exception, state} to return an error and disconnect.

handle_rollback(opts, state)

Handle rolling back a transaction. Return {:ok, result, state} on successfully rolling back transaction, {status, state} to notify caller that the transaction can not rollback due to the transaction status status or {:disconnect, exception, state} to error and disconnect.

handle_status(opts, state)

Handle getting the transaction status. Return {:idle, state} if outside a transaction, {:transaction, state} if inside a transaction, {:error, state} if inside an aborted transaction, or {:disconnect, exception, state} to error and disconnect.

ping(state)

Called when the connection has been idle for a period of time. Return {:ok, state} to continue or {:disconnect, exception, state} to disconnect.

Functions

__using__(_)

Use DBConnection to set the behaviour.

child_spec(conn_mod, opts)

Creates a supervisor child specification for a pool of connections.

close(conn, query, opts \\ [])

Close a prepared query on a database connection and return {:ok, result} on success or {:error, exception} on error.

close!(conn, query, opts \\ [])

Close a prepared query on a database connection and return the result. Raises an exception on error.

connection_module(conn)

Returns connection module used by the given connection pool.

disconnect_all(conn, interval, opts \\ [])

Forces all connections in the pool to disconnect within the given interval in milliseconds.

execute(conn, query, params, opts \\ [])

Execute a prepared query with a database connection and return {:ok, query, result} on success or {:error, exception} if there was an error.

execute!(conn, query, params, opts \\ [])

Execute a prepared query with a database connection and return the result. Raises an exception on error.

prepare(conn, query, opts \\ [])

Prepare a query with a database connection for later execution.

prepare!(conn, query, opts \\ [])

Prepare a query with a database connection and return the prepared query. An exception is raised on error.

prepare_execute(conn, query, params, opts \\ [])

Prepare a query and execute it with a database connection and return both the prepared query and the result, {:ok, query, result} on success or {:error, exception} if there was an error.

prepare_execute!(conn, query, params, opts \\ [])

Prepare a query and execute it with a database connection and return both the prepared query and result. An exception is raised on error.

prepare_stream(conn, query, params, opts \\ [])

Create a stream that will prepare a query, execute it and stream results using a cursor.

reduce(stream, acc, fun)

Reduces a previously built stream or prepared stream.

rollback(conn, reason)

Rollback a database transaction and release lock on connection.

run(conn, fun, opts \\ [])

Acquire a lock on a connection and run a series of requests on it.

start_link(conn_mod, opts)

Starts and links to a database connection process.

status(conn, opts \\ [])

Return the transaction status of a connection.

stream(conn, query, params, opts \\ [])

Create a stream that will execute a prepared query and stream results using a cursor.

transaction(conn, fun, opts \\ [])

Acquire a lock on a connection and run a series of requests inside a transaction. The result of the transaction fun is return inside an :ok tuple: {:ok, result}.

Link to this section Types

conn()

@type conn() :: GenServer.server() | t()

cursor()

@type cursor() :: any()

option()

@type option() ::
  {:log,
   (DBConnection.LogEntry.t() -> any()) | {module(), atom(), [any()]} | nil}
  | {:queue, boolean()}
  | {:timeout, timeout()}
  | {:deadline, integer() | nil}

params()

@type params() :: any()

query()

@type query() :: DBConnection.Query.t()

result()

@type result() :: any()

start_option()

@type start_option() ::
  {:after_connect, (t() -> any()) | {module(), atom(), [any()]} | nil}
  | {:after_connect_timeout, timeout()}
  | {:connection_listeners, [Process.dest()] | nil}
  | {:backoff_max, non_neg_integer()}
  | {:backoff_min, non_neg_integer()}
  | {:backoff_type, :stop | :exp | :rand | :rand_exp}
  | {:configure, (keyword() -> keyword()) | {module(), atom(), [any()]} | nil}
  | {:idle_interval, non_neg_integer()}
  | {:idle_limit, non_neg_integer()}
  | {:max_restarts, non_neg_integer()}
  | {:max_seconds, pos_integer()}
  | {:name, GenServer.name()}
  | {:pool, module()}
  | {:pool_size, pos_integer()}
  | {:queue_interval, non_neg_integer()}
  | {:queue_target, non_neg_integer()}
  | {:show_sensitive_data_on_connection_error, boolean()}

status()

@type status() :: :idle | :transaction | :error

t()

@type t() :: %DBConnection{conn_mode: term(), conn_ref: reference(), pool_ref: any()}

Run or transaction connection reference.

Link to this section Callbacks

checkout(state)

@callback checkout(state :: any()) ::
  {:ok, new_state :: any()} | {:disconnect, Exception.t(), new_state :: any()}

Checkouts the state from the connection process. Return {:ok, state} to allow the checkout or {:disconnect, exception, state} to disconnect.

This callback is called immediately after the connection is established and the state is never effetively checked in again. That's because DBConnection keeps the connection state in an ETS table that is moved between the different clients checking out connections. There is no checkin callback. The state is only handed back to the connection process during pings and (re)connects.

This callback is called in the connection process.

connect(opts)

@callback connect(opts :: Keyword.t()) :: {:ok, state :: any()} | {:error, Exception.t()}

Connect to the database. Return {:ok, state} on success or {:error, exception} on failure.

If an error is returned it will be logged and another connection attempt will be made after a backoff interval.

This callback is called in the connection process.

disconnect(err, state)

@callback disconnect(err :: Exception.t(), state :: any()) :: :ok

Disconnect from the database. Return :ok.

The exception as first argument is the exception from a :disconnect 3-tuple returned by a previous callback.

If the state is controlled by a client and it exits or takes too long to process a request the state will be last known state. In these cases the exception will be a DBConnection.ConnectionError.

This callback is called in the connection process.

handle_begin(opts, state)

@callback handle_begin(opts :: Keyword.t(), state :: any()) ::
  {:ok, result(), new_state :: any()}
  | {status(), new_state :: any()}
  | {:disconnect, Exception.t(), new_state :: any()}

Handle the beginning of a transaction.

Return {:ok, result, state} to continue, {status, state} to notify caller that the transaction can not begin due to the transaction status status, or {:disconnect, exception, state} to error and disconnect.

A callback implementation should only return status if it can determine the database's transaction status without side effect.