Struct ockam::SqlxDatabase
source · pub struct SqlxDatabase {
pub pool: Pool<Sqlite>,
}
Expand description
We use sqlx as our primary interface for interacting with the database The database driver is currently Sqlite
Fields§
§pool: Pool<Sqlite>
Pool of connections to the database
Implementations§
source§impl SqlxDatabase
impl SqlxDatabase
sourcepub async fn create(path: impl AsRef<Path>) -> Result<SqlxDatabase, Error>
pub async fn create(path: impl AsRef<Path>) -> Result<SqlxDatabase, Error>
Constructor for a database persisted on disk
sourcepub async fn in_memory(usage: &str) -> Result<Arc<SqlxDatabase>, Error>
pub async fn in_memory(usage: &str) -> Result<Arc<SqlxDatabase>, Error>
Constructor for an in-memory database The implementation blocks during the creation of the database so that we don’t have to propagate async in all the code base when using an in-memory database, especially when writing examples
sourcepub fn map_sql_err(err: Error) -> Error
pub fn map_sql_err(err: Error) -> Error
Map a sqlx error into an ockam error
sourcepub fn map_migrate_err(err: MigrateError) -> Error
pub fn map_migrate_err(err: MigrateError) -> Error
Map a sqlx migration error into an ockam error
sourcepub fn map_decode_err(err: Error) -> Error
pub fn map_decode_err(err: Error) -> Error
Map a minicbor decode error into an ockam error
Methods from Deref<Target = Pool<Sqlite>>§
pub fn acquire(
&self
) -> impl Future<Output = Result<PoolConnection<DB>, Error>> + 'static
pub fn acquire( &self ) -> impl Future<Output = Result<PoolConnection<DB>, Error>> + 'static
Retrieves a connection from the pool.
The total time this method is allowed to execute is capped by
[PoolOptions::acquire_timeout
].
If that timeout elapses, this will return [Error::PoolClosed
].
Note: Cancellation/Timeout May Drop Connections
If acquire
is cancelled or times out after it acquires a connection from the idle queue or
opens a new one, it will drop that connection because we don’t want to assume it
is safe to return to the pool, and testing it to see if it’s safe to release could introduce
subtle bugs if not implemented correctly. To avoid that entirely, we’ve decided to not
gracefully handle cancellation here.
However, if your workload is sensitive to dropped connections such as using an in-memory
SQLite database with a pool size of 1, you can pretty easily ensure that a cancelled
acquire()
call will never drop connections by tweaking your [PoolOptions
]:
- Set [
test_before_acquire(false)
][PoolOptions::test_before_acquire] - Never set [
before_acquire
][PoolOptions::before_acquire] or [after_connect
][PoolOptions::after_connect].
This should eliminate any potential .await
points between acquiring a connection and
returning it.
pub fn try_acquire(&self) -> Option<PoolConnection<DB>>
pub fn try_acquire(&self) -> Option<PoolConnection<DB>>
Attempts to retrieve a connection from the pool if there is one available.
Returns None
immediately if there are no idle connections available in the pool
or there are tasks waiting for a connection which have yet to wake.
pub async fn begin(&self) -> Result<Transaction<'static, DB>, Error>
pub async fn begin(&self) -> Result<Transaction<'static, DB>, Error>
Retrieves a connection and immediately begins a new transaction.
pub async fn try_begin(&self) -> Result<Option<Transaction<'static, DB>>, Error>
pub async fn try_begin(&self) -> Result<Option<Transaction<'static, DB>>, Error>
Attempts to retrieve a connection and immediately begins a new transaction if successful.
pub fn close(&self) -> impl Future<Output = ()>
pub fn close(&self) -> impl Future<Output = ()>
Shut down the connection pool, immediately waking all tasks waiting for a connection.
Upon calling this method, any currently waiting or subsequent calls to [Pool::acquire
] and
the like will immediately return [Error::PoolClosed
] and no new connections will be opened.
Checked-out connections are unaffected, but will be gracefully closed on-drop
rather than being returned to the pool.
Returns a Future
which can be .await
ed to ensure all connections are
gracefully closed. It will first close any idle connections currently waiting in the pool,
then wait for all checked-out connections to be returned or closed.
Waiting for connections to be gracefully closed is optional, but will allow the database server to clean up the resources sooner rather than later. This is especially important for tests that create a new pool every time, otherwise you may see errors about connection limits being exhausted even when running tests in a single thread.
If the returned Future
is not run to completion, any remaining connections will be dropped
when the last handle for the given pool instance is dropped, which could happen in a task
spawned by Pool
internally and so may be unpredictable otherwise.
.close()
may be safely called and .await
ed on multiple handles concurrently.
pub fn is_closed(&self) -> bool
pub fn is_closed(&self) -> bool
Returns true
if [.close()
][Pool::close] has been called on the pool, false
otherwise.
pub fn close_event(&self) -> CloseEvent ⓘ
pub fn close_event(&self) -> CloseEvent ⓘ
Get a future that resolves when [Pool::close()
] is called.
If the pool is already closed, the future resolves immediately.
This can be used to cancel long-running operations that hold onto a [PoolConnection
]
so they don’t prevent the pool from closing (which would otherwise wait until all
connections are returned).
Examples
These examples use Postgres and Tokio, but should suffice to demonstrate the concept.
Do something when the pool is closed:
use sqlx::PgPool;
let pool = PgPool::connect("postgresql://...").await?;
let pool2 = pool.clone();
tokio::spawn(async move {
// Demonstrates that `CloseEvent` is itself a `Future` you can wait on.
// This lets you implement any kind of on-close event that you like.
pool2.close_event().await;
println!("Pool is closing!");
// Imagine maybe recording application statistics or logging a report, etc.
});
// The rest of the application executes normally...
// Close the pool before the application exits...
pool.close().await;
Cancel a long-running operation:
use sqlx::{Executor, PgPool};
let pool = PgPool::connect("postgresql://...").await?;
let pool2 = pool.clone();
tokio::spawn(async move {
pool2.close_event().do_until(async {
// This statement normally won't return for 30 days!
// (Assuming the connection doesn't time out first, of course.)
pool2.execute("SELECT pg_sleep('30 days')").await;
// If the pool is closed before the statement completes, this won't be printed.
// This is because `.do_until()` cancels the future it's given if the
// pool is closed first.
println!("Waited!");
}).await;
});
// This normally wouldn't return until the above statement completed and the connection
// was returned to the pool. However, thanks to `.do_until()`, the operation was
// cancelled as soon as we called `.close().await`.
pool.close().await;
pub fn size(&self) -> u32
pub fn size(&self) -> u32
Returns the number of connections currently active. This includes idle connections.
pub fn num_idle(&self) -> usize
pub fn num_idle(&self) -> usize
Returns the number of connections active and idle (not in use).
As of 0.6.0, this has been fixed to use a separate atomic counter and so should be fine to call even at high load.
This previously called [crossbeam::queue::ArrayQueue::len()
] which waits for the head and
tail pointers to be in a consistent state, which may never happen at high levels of churn.
pub fn connect_options(
&self
) -> Arc<<<DB as Database>::Connection as Connection>::Options>
pub fn connect_options( &self ) -> Arc<<<DB as Database>::Connection as Connection>::Options>
Gets a clone of the connection options for this pool
pub fn set_connect_options(
&self,
connect_options: <<DB as Database>::Connection as Connection>::Options
)
pub fn set_connect_options( &self, connect_options: <<DB as Database>::Connection as Connection>::Options )
Updates the connection options this pool will use when opening any future connections. Any existing open connection in the pool will be left as-is.
pub fn options(&self) -> &PoolOptions<DB>
pub fn options(&self) -> &PoolOptions<DB>
Get the options for this pool