1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69

use crate::query::SqliteQuery;
use crate::{
    Sqlite, SqliteConnection, SqliteQueryResult, SqliteRow, SqliteStatement, SqliteTypeInfo,
};
use either::Either;
use futures_core::future::BoxFuture;
use futures_core::stream::BoxStream;
use futures_util::{TryFutureExt, TryStreamExt};
use rbdc::error::Error;
use std::ops::Deref;
use std::sync::Arc;

impl SqliteConnection {
    pub fn fetch_many(
        &mut self,
        mut query: SqliteQuery,
    ) -> BoxStream<'_, Result<Either<SqliteQueryResult, SqliteRow>, Error>> {
        let sql = query.sql().to_string();
        let persistent = query.persistent() && !query.arguments.is_empty();
        let arguments = query.take_arguments();
        Box::pin(
            self.worker
                .execute(sql, arguments, self.row_channel_size, persistent)
                .map_ok(flume::Receiver::into_stream)
                .try_flatten_stream(),
        )
    }

    pub fn fetch_optional(
        &mut self,
        mut query: SqliteQuery,
    ) -> BoxFuture<'_, Result<Option<SqliteRow>, Error>> {
        let sql = query.sql().to_owned();
        let persistent = query.persistent() && !query.arguments.is_empty();
        let arguments = query.take_arguments();
        Box::pin(async move {
            let stream = self
                .worker
                .execute(sql, arguments, self.row_channel_size, persistent)
                .map_ok(flume::Receiver::into_stream)
                .try_flatten_stream();

            futures_util::pin_mut!(stream);

            while let Some(res) = stream.try_next().await? {
                if let Either::Right(row) = res {
                    return Ok(Some(row));
                }
            }

            Ok(None)
        })
    }

    pub fn prepare_with<'a>(
        &'a mut self,
        sql: &'a str,
        _parameters: &[SqliteTypeInfo],
    ) -> BoxFuture<'_, Result<SqliteStatement, Error>> {
        Box::pin(async move {
            let statement = self.worker.prepare(sql).await?;

            Ok(SqliteStatement {
                sql: sql.into(),
                ..statement
            })
        })
    }
}