coven/
read.rs

1use std::future::{Future, IntoFuture};
2use std::pin::Pin;
3
4use coven_database::store::StoreReads;
5
6use crate::{CovenError, CovenResult, SqlReadContext};
7
8/// A database read that starts when awaited.
9///
10/// Constructed by [`CovenHandle::read`](crate::CovenHandle::read) or
11/// [`CovenReadHandle::read`](crate::CovenReadHandle::read). Await it to receive
12/// the fetched values, or attach [`process`](Self::process) to compute a result
13/// on separate workers after releasing the database connection.
14///
15/// ```no_run
16/// # async fn example(handle: &coven::CovenHandle) -> coven::CovenResult<()> {
17/// let titles = handle.read(|sql| {
18///     Ok(sql.query("SELECT body FROM notes", [], |row| row.get::<_, String>(0))?)
19/// }).process(|mut titles| {
20///     titles.sort();
21///     Ok(titles)
22/// }).await?;
23/// # Ok(())
24/// # }
25/// ```
26///
27/// Use [`IntoFuture::into_future`] when passing the read to an API that requires
28/// a [`Future`] rather than an awaitable value.
29#[must_use = "reads do not execute until awaited"]
30pub struct Read<'a, F> {
31    database: &'a StoreReads,
32    fetch: F,
33}
34
35impl<'a, F, Raw> Read<'a, F>
36where
37    F: for<'connection> FnOnce(SqlReadContext<'connection>) -> CovenResult<Raw> + Send + 'static,
38    Raw: Send + 'static,
39{
40    pub(crate) fn new(database: &'a StoreReads, fetch: F) -> Self {
41        Self { database, fetch }
42    }
43
44    /// Process the fetched values on bounded workers after the read transaction
45    /// ends. Fetch every database input in the read closure; the processor
46    /// receives owned values without a SQL context. A failed read skips it.
47    pub async fn process<P, R>(self, process: P) -> CovenResult<R>
48    where
49        P: FnOnce(Raw) -> CovenResult<R> + Send + 'static,
50        R: Send + 'static,
51    {
52        let database = self.database;
53        let raw = self.await?;
54        database.process(move || process(raw)).await
55    }
56}
57
58impl<'a, F, R> IntoFuture for Read<'a, F>
59where
60    F: for<'connection> FnOnce(SqlReadContext<'connection>) -> CovenResult<R> + Send + 'static,
61    R: Send + 'static,
62{
63    type Output = CovenResult<R>;
64    type IntoFuture = Pin<Box<dyn Future<Output = Self::Output> + Send + 'a>>;
65
66    fn into_future(self) -> Self::IntoFuture {
67        Box::pin(async move {
68            self.database
69                .read(self.fetch)
70                .await
71                .map_err(CovenError::from)?
72        })
73    }
74}
coven/read.rs

coven/
read.rs
