Struct AsyncPgConnection 

pub struct AsyncPgConnection { /* private fields */ }

Available on crate feature postgres only.

A connection to a PostgreSQL database.

Connection URLs should be in the form postgres://[user[:password]@]host/database_name

Checkout the documentation of the tokio_postgres crate for details about the format

Pipelining

This connection supports pipelined requests. Pipelining can improve performance in use cases in which multiple, independent queries need to be executed. In a traditional workflow, each query is sent to the server after the previous query completes. In contrast, pipelining allows the client to send all of the queries to the server up front, minimizing time spent by one side waiting for the other to finish sending data:

            Sequential                              Pipelined
| Client         | Server          |    | Client         | Server          |
|----------------|-----------------|    |----------------|-----------------|
| send query 1   |                 |    | send query 1   |                 |
|                | process query 1 |    | send query 2   | process query 1 |
| receive rows 1 |                 |    | send query 3   | process query 2 |
| send query 2   |                 |    | receive rows 1 | process query 3 |
|                | process query 2 |    | receive rows 2 |                 |
| receive rows 2 |                 |    | receive rows 3 |                 |
| send query 3   |                 |
|                | process query 3 |
| receive rows 3 |                 |

In both cases, the PostgreSQL server is executing the queries sequentially - pipelining just allows both sides of the connection to work concurrently when possible.

Pipelining happens automatically when futures are polled concurrently (for example, by using the futures join combinator):

use diesel_async::RunQueryDsl;

      let q1 = diesel::select(1_i32.into_sql::<Integer>());
      let q2 = diesel::select(2_i32.into_sql::<Integer>());

      // construct multiple futures for different queries
      let f1 = q1.get_result::<i32>(conn);
      let f2 = q2.get_result::<i32>(conn);

      // wait on both results
      let res = futures_util::try_join!(f1, f2)?;

      assert_eq!(res.0, 1);
      assert_eq!(res.1, 2);

For more complex cases, an immutable reference to the connection need to be used:

use diesel_async::RunQueryDsl;

      async fn fn12(mut conn: &AsyncPgConnection) -> QueryResult<(i32, i32)> {
          let f1 = diesel::select(1_i32.into_sql::<Integer>()).get_result::<i32>(&mut conn);
          let f2 = diesel::select(2_i32.into_sql::<Integer>()).get_result::<i32>(&mut conn);

          futures_util::try_join!(f1, f2)
      }

      async fn fn34(mut conn: &AsyncPgConnection) -> QueryResult<(i32, i32)> {
          let f3 = diesel::select(3_i32.into_sql::<Integer>()).get_result::<i32>(&mut conn);
          let f4 = diesel::select(4_i32.into_sql::<Integer>()).get_result::<i32>(&mut conn);

          futures_util::try_join!(f3, f4)
      }

      let f12 = fn12(&conn);
      let f34 = fn34(&conn);

      let ((r1, r2), (r3, r4)) = futures_util::try_join!(f12, f34).unwrap();

      assert_eq!(r1, 1);
      assert_eq!(r2, 2);
      assert_eq!(r3, 3);
      assert_eq!(r4, 4);

TLS

Connections created by AsyncPgConnection::establish do not support TLS.

TLS support for tokio_postgres connections is implemented by external crates, e.g. tokio_postgres_rustls.

AsyncPgConnection::try_from_client_and_connection can be used to construct a connection from an existing tokio_postgres::Connection with TLS enabled.

Implementations

impl AsyncPgConnection

pub fn build_transaction(&mut self) -> TransactionBuilder<'_, Self>

Build a transaction, specifying additional details such as isolation level

See TransactionBuilder for more examples.

conn.build_transaction()
    .read_only()
    .serializable()
    .deferrable()
    .run(|conn| async move { Ok(()) }.scope_boxed())
    .await

pub async fn try_from(conn: Client) -> ConnectionResult<Self>

Construct a new AsyncPgConnection instance from an existing tokio_postgres::Client

pub async fn try_from_client_and_connection<S>( client: Client, conn: Connection<Socket, S>, ) -> ConnectionResult<Self>

where S: TlsStream + Unpin + Send + 'static,

Constructs a new AsyncPgConnection from an existing tokio_postgres::Client and tokio_postgres::Connection

pub fn cancel_token(&self) -> CancelToken

Constructs a cancellation token that can later be used to request cancellation of a query running on the connection associated with this client.

pub fn notifications_stream( &mut self, ) -> impl Stream<Item = QueryResult<PgNotification>> + '_

See Postgres documentation for SQL commands NOTIFY and LISTEN

The returned stream yields all notifications received by the connection, not only notifications received after calling the function. The returned stream will never close, so no notifications will just result in a pending state.

If there’s no connection available to poll, the stream will yield no notifications and be pending forever. This can happen if you created the AsyncPgConnection by the try_from constructor.

// register the notifications channel we want to receive notifications for
diesel::sql_query("LISTEN example_channel").execute(conn).await?;
// send some notification (usually done from a different connection/thread/application)
diesel::sql_query("NOTIFY example_channel, 'additional data'").execute(conn).await?;

let mut notifications = std::pin::pin!(conn.notifications_stream());
let mut notification = notifications.next().await.unwrap().unwrap();

assert_eq!(notification.channel, "example_channel");
assert_eq!(notification.payload, "additional data");
println!("Notification received from process with id {}", notification.process_id);

Trait Implementations

impl AsyncConnection for AsyncPgConnection

async fn establish(database_url: &str) -> ConnectionResult<Self>

Establishes a new connection to the database

fn set_instrumentation(&mut self, instrumentation: impl Instrumentation)

Set a specific Instrumentation implementation for this connection

fn set_prepared_statement_cache_size(&mut self, size: CacheSize)

Set the prepared statement cache size to CacheSize for this connection

fn transaction<'a, 'conn, R, E, F>( &'conn mut self, callback: F, ) -> BoxFuture<'conn, Result<R, E>>

where F: for<'r> FnOnce(&'r mut Self) -> ScopedBoxFuture<'a, 'r, Result<R, E>> + Send + 'a, E: From<Error> + Send + 'a, R: Send + 'a, 'a: 'conn,

Executes the given function inside of a database transaction

fn begin_test_transaction( &mut self, ) -> impl Future<Output = QueryResult<()>> + Send

Creates a transaction that will never be committed. This is useful for tests. Panics if called while inside of a transaction or if called with a connection containing a broken transaction

fn test_transaction<'conn, 'a, R, E, F>( &'conn mut self, f: F, ) -> impl Future<Output = R> + Send + 'conn

where F: for<'r> FnOnce(&'r mut Self) -> ScopedBoxFuture<'a, 'r, Result<R, E>> + Send + 'a, E: Debug + Send + 'a, R: Send + 'a, 'a: 'conn,

Executes the given function inside a transaction, but does not commit it. Panics if the given function returns an error.

impl AsyncConnectionCore for &AsyncPgConnection

type LoadFuture<'conn, 'query> = <AsyncPgConnection as AsyncConnectionCore>::LoadFuture<'conn, 'query>

The future returned by AsyncConnection::load

type ExecuteFuture<'conn, 'query> = <AsyncPgConnection as AsyncConnectionCore>::ExecuteFuture<'conn, 'query>

The future returned by AsyncConnection::execute

type Stream<'conn, 'query> = <AsyncPgConnection as AsyncConnectionCore>::Stream<'conn, 'query>

The inner stream returned by AsyncConnection::load

type Row<'conn, 'query> = <AsyncPgConnection as AsyncConnectionCore>::Row<'conn, 'query>

The row type used by the stream returned by AsyncConnection::load

type Backend = <AsyncPgConnection as AsyncConnectionCore>::Backend

The backend this type connects to

impl AsyncConnectionCore for AsyncPgConnection

type LoadFuture<'conn, 'query> = Pin<Box<dyn Future<Output = Result<<AsyncPgConnection as AsyncConnectionCore>::Stream<'conn, 'query>, Error>> + Send + 'query>>

The future returned by AsyncConnection::load

type ExecuteFuture<'conn, 'query> = Pin<Box<dyn Future<Output = Result<usize, Error>> + Send + 'query>>

The future returned by AsyncConnection::execute

type Stream<'conn, 'query> = Pin<Box<dyn Stream<Item = Result<PgRow, Error>> + Send>>

The inner stream returned by AsyncConnection::load

type Row<'conn, 'query> = PgRow

The row type used by the stream returned by AsyncConnection::load

type Backend = Pg

The backend this type connects to

impl Drop for AsyncPgConnection

fn drop(&mut self)

Executes the destructor for this type.

impl SimpleAsyncConnection for &AsyncPgConnection

async fn batch_execute(&mut self, query: &str) -> QueryResult<()>

Execute multiple SQL statements within the same string.

impl SimpleAsyncConnection for AsyncPgConnection

async fn batch_execute(&mut self, query: &str) -> QueryResult<()>

Execute multiple SQL statements within the same string.

impl<'b, Changes, Output, Tab, V> UpdateAndFetchResults<Changes, Output> for AsyncPgConnection

where Output: Send + 'static, Changes: Copy + AsChangeset<Target = Tab> + Send + Identifiable<Table = Tab>, Tab: Table + FindDsl<Changes::Id> + 'b, Find<Tab, Changes::Id>: IntoUpdateTarget<Table = Tab, WhereClause = V>, UpdateStatement<Tab, V, Changes::Changeset>: AsQuery, Update<Changes, Changes>: LoadQuery<'b, Self, Output>, V: Send + 'b, Changes::Changeset: Send + 'b, Tab::FromClause: Send,

fn update_and_fetch<'conn, 'changes>( &'conn mut self, changeset: Changes, ) -> BoxFuture<'changes, QueryResult<Output>>

where Changes: 'changes, Changes::Changeset: 'changes, Self: 'changes, 'conn: 'changes,

See the traits documentation.