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
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90

use cdbc::describe::Describe;
use cdbc::error::Error;
use cdbc::executor::{Execute, Executor};
use crate::connection::describe::describe;
use crate::statement::{StatementHandle, VirtualStatement};
use crate::{
    Sqlite, SqliteArguments, SqliteConnection, SqliteQueryResult, SqliteRow, SqliteStatement,
    SqliteTypeInfo,
};
use either::Either;
use libsqlite3_sys::sqlite3_last_insert_rowid;
use std::borrow::Cow;
use std::sync::Arc;
use std::sync::mpsc::RecvError;
use cogo::std::io::TryStream;
use cogo::std::sync::mpmc;
use cogo::std::sync::mpmc::Receiver;
use cdbc::database::{Database, HasStatement};
use cdbc::io::chan_stream::ChanStream;
use cdbc::utils::statement_cache::StatementCache;
use crate::connection::executor_mut;


impl Executor for &mut SqliteConnection {
    type Database = Sqlite;

    fn fetch_many<'q, E: 'q>(&mut self,
                             mut query: E,
    ) -> ChanStream<Either<SqliteQueryResult, SqliteRow>>
        where
            E: Execute<'q, Self::Database>,
    {
        let sql = query.sql();
        let arguments = query.take_arguments();
        let persistent = query.persistent() && arguments.is_some();
        let s = self.worker
            .execute(sql, arguments, self.row_channel_size, persistent);
        if s.is_err() {
            let c = ChanStream::new(|sender|
                Err(s.err().unwrap())
            );
            return c;
        }
        let s = s.unwrap();
        executor_mut::sender_to_stream(s)
    }

    fn fetch_optional<'q, E: 'q>(
        &mut self,
        mut query: E,
    ) -> Result<Option<SqliteRow>, Error>
        where
            E: Execute<'q, Self::Database>,
    {
        let sql = query.sql();
        let arguments = query.take_arguments();
        let persistent = query.persistent() && arguments.is_some();
        let mut stream = self
            .worker
            .execute(sql, arguments, self.row_channel_size, persistent)?;
        let mut stream = executor_mut::sender_to_stream(stream);
        use crate::cdbc::io::chan_stream::TryStream;
        while let Some(res) = stream.try_next()? {
            if let Either::Right(row) = res {
                return Ok(Some(row));
            }
        }
        Ok(None)
    }

    fn prepare_with<'q>(
        &mut self,
        sql: &'q str,
        _parameters: &[SqliteTypeInfo],
    ) -> Result<SqliteStatement<'q>, Error>
        where
    {
        let statement = self.worker.prepare(sql)?;
        Ok(SqliteStatement {
            sql: sql.into(),
            ..statement
        })
    }

    #[doc(hidden)]
    fn describe<'q>(&mut self, sql: &'q str) -> Result<Describe<Sqlite>, Error>
    {
        self.worker.describe(sql)
    }
}