tdsql 0.9.0

//! A blocking client, for callers that are not async.
//!
//! Enable the `blocking` feature to get it:
//!
//! ```toml
//! [dependencies]
//! tdsql = { version = "0.8.0", features = ["blocking"] }
//! ```
//!
//! It is the same client with the `async` taken off: each call drives the
//! asynchronous one to completion on a runtime the client owns.
//!
//! ```no_run
//! use tdsql::blocking::Client;
//! use tdsql::Config;
//!
//! # fn main() -> tdsql::Result<()> {
//! let mut client = Client::connect(
//!     &Config::new()
//!         .host("localhost")
//!         .database("master")
//!         .auth("sa", "YourStrong!Passw0rd")
//!         .trust_cert(),
//! )?;
//!
//! let rows = client.query("SELECT id, name FROM users WHERE id > @P1", &[&10i32])?;
//! for row in &rows {
//!     let id: i32 = row.get("id");
//!     println!("{id}");
//! }
//! # Ok(())
//! # }
//! ```
//!
//! # Do not call this from async code
//!
//! Each client owns a runtime, and driving a runtime from a thread that is
//! running another one panics. [`Client::connect`] refuses when a Tokio
//! runtime is current on the calling thread and returns
//! [`Error::BlockingInAsync`]. That includes a `spawn_blocking` closure, where
//! the client would in fact work: connect on a thread of your own and move
//! the client in. The check is made only at connect; a client that is later
//! *used* inside async code panics as Tokio does. Dropping one there is fine.
//! In async code, use [`tdsql::Client`](crate::Client) directly.

use tokio::runtime::{Builder, Runtime};

use crate::command::Command;
use crate::config::Config;
use crate::dataset::DataSet;
use crate::error::{Error, Result};
use crate::row::{Column, Row};
use crate::transaction::IsolationLevel;
use crate::value::{FromSql, ToSql};

fn build_runtime() -> Result<Runtime> {
    // A dedicated worker thread means the connection task keeps making
    // progress between calls, so a queued rollback is sent promptly rather
    // than waiting for the next statement.
    Builder::new_multi_thread()
        .worker_threads(1)
        .enable_all()
        .thread_name("tdsql-connection")
        .build()
        .map_err(Error::Io)
}

/// The client's runtime, which knows how to go away from inside async code.
///
/// Dropping a `Runtime` on a thread that is driving another runtime panics
/// ("Cannot drop a runtime in a context where blocking is not allowed"), and
/// a client moved into async code and dropped there is an easy thing to do.
/// `shutdown_background` is allowed anywhere; it does not wait for the
/// connection task, which the dropped `inner` client has already told to stop.
struct OwnedRuntime(Option<Runtime>);

impl std::ops::Deref for OwnedRuntime {
    type Target = Runtime;

    fn deref(&self) -> &Runtime {
        self.0.as_ref().expect("taken only in Drop")
    }
}

impl Drop for OwnedRuntime {
    fn drop(&mut self) {
        if let Some(runtime) = self.0.take()
            && tokio::runtime::Handle::try_current().is_ok()
        {
            runtime.shutdown_background();
        }
    }
}

/// A blocking connection to SQL Server.
///
/// The blocking counterpart to [`tdsql::Client`](crate::Client); every method
/// mirrors it without `async`/`await`.
pub struct Client {
    runtime: OwnedRuntime,
    inner: crate::Client,
}

impl std::fmt::Debug for Client {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("blocking::Client")
            .field("inner", &self.inner)
            .finish()
    }
}

impl Client {
    /// Connect using this configuration.
    ///
    /// Fails with [`Error::BlockingInAsync`] when a Tokio runtime is current
    /// on this thread, a `spawn_blocking` closure included; see the
    /// [module docs](self#do-not-call-this-from-async-code).
    pub fn connect(config: &Config) -> Result<Self> {
        if tokio::runtime::Handle::try_current().is_ok() {
            return Err(Error::BlockingInAsync);
        }
        let runtime = build_runtime()?;
        let inner = runtime.block_on(crate::Client::connect(config))?;
        Ok(Self {
            runtime: OwnedRuntime(Some(runtime)),
            inner,
        })
    }

    /// Connect using an ADO.NET-style connection string.
    pub fn connect_str(connection_string: &str) -> Result<Self> {
        Self::connect(&Config::from_ado_string(connection_string))
    }

    /// Run a query and collect every row of the first result set.
    pub fn query(&mut self, sql: &str, params: &[&dyn ToSql]) -> Result<Vec<Row>> {
        self.runtime.block_on(self.inner.query(sql, params))
    }

    /// Run a query and read the rows of its first result set as the server
    /// sends them.
    ///
    /// See [`tdsql::Client::query_stream`](crate::Client::query_stream).
    ///
    /// ```no_run
    /// # use tdsql::blocking::Client;
    /// # fn f(client: &mut Client) -> tdsql::Result<()> {
    /// for row in client.query_stream("SELECT id FROM users", &[])? {
    ///     let id: i32 = row?.get("id");
    ///     println!("{id}");
    /// }
    /// # Ok(())
    /// # }
    /// ```
    pub fn query_stream(&mut self, sql: &str, params: &[&dyn ToSql]) -> Result<RowIter<'_>> {
        let Self { runtime, inner } = self;
        let rows = runtime.block_on(inner.query_stream(sql, params))?;
        Ok(RowIter {
            runtime,
            inner: rows,
        })
    }

    /// Run a query expecting exactly one row.
    pub fn query_one(&mut self, sql: &str, params: &[&dyn ToSql]) -> Result<Row> {
        self.runtime.block_on(self.inner.query_one(sql, params))
    }

    /// Run a query expecting at most one row.
    pub fn query_opt(&mut self, sql: &str, params: &[&dyn ToSql]) -> Result<Option<Row>> {
        self.runtime.block_on(self.inner.query_opt(sql, params))
    }

    /// Run a query and read the first column of the first row.
    pub fn query_scalar<T: FromSql>(&mut self, sql: &str, params: &[&dyn ToSql]) -> Result<T> {
        self.runtime.block_on(self.inner.query_scalar(sql, params))
    }

    /// Run a statement and report how many rows it affected.
    pub fn execute(&mut self, sql: &str, params: &[&dyn ToSql]) -> Result<u64> {
        self.runtime.block_on(self.inner.execute(sql, params))
    }

    /// Run a SQL batch verbatim, with no parameters, and collect its rows.
    ///
    /// Use this for DDL; see [`tdsql::Client::batch`](crate::Client::batch).
    pub fn batch(&mut self, sql: &str) -> Result<Vec<Row>> {
        self.runtime.block_on(self.inner.batch(sql))
    }

    /// Run a SQL batch verbatim and collect every result set.
    pub fn batch_dataset(&mut self, sql: &str) -> Result<DataSet> {
        self.runtime.block_on(self.inner.batch_dataset(sql))
    }

    /// Run a [`Command`] and collect every result set it produces.
    pub fn query_dataset(&mut self, command: &Command) -> Result<DataSet> {
        self.runtime.block_on(self.inner.query_dataset(command))
    }

    /// Run a [`Command`] for its row count rather than its rows.
    pub fn execute_command(&mut self, command: &Command) -> Result<u64> {
        self.runtime.block_on(self.inner.execute_command(command))
    }

    /// Begin a transaction at `READ COMMITTED`, SQL Server's default isolation
    /// level.
    ///
    /// ```no_run
    /// # use tdsql::blocking::Client;
    /// # fn f(client: &mut Client) -> tdsql::Result<()> {
    /// let mut tx = client.transaction()?;
    /// tx.execute("INSERT INTO orders (id) VALUES (@P1)", &[&1i32])?;
    /// tx.commit()?;
    /// # Ok(())
    /// # }
    /// ```
    pub fn transaction(&mut self) -> Result<Transaction<'_>> {
        self.transaction_with_isolation(IsolationLevel::ReadCommitted)
    }

    /// Begin a transaction at an explicit isolation level.
    ///
    /// Once it ends, the connection goes back to `READ COMMITTED`.
    pub fn transaction_with_isolation(&mut self, level: IsolationLevel) -> Result<Transaction<'_>> {
        let Self { runtime, inner } = self;
        let tx = runtime.block_on(inner.transaction_with_isolation(level))?;
        Ok(Transaction { runtime, inner: tx })
    }

    /// Close the connection.
    pub fn close(self) -> Result<()> {
        let Self { runtime, inner } = self;
        runtime.block_on(inner.close())
    }
}

/// An in-progress transaction on a blocking [`Client`].
///
/// Behaves exactly like [`tdsql::Transaction`](crate::Transaction): commit to
/// keep the work, and **dropping without committing rolls back**.
pub struct Transaction<'a> {
    runtime: &'a Runtime,
    inner: crate::Transaction<'a>,
}

impl std::fmt::Debug for Transaction<'_> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("blocking::Transaction")
            .field("inner", &self.inner)
            .finish()
    }
}

impl Transaction<'_> {
    /// Commit the work.
    pub fn commit(self) -> Result<()> {
        let Self { runtime, inner } = self;
        runtime.block_on(inner.commit())
    }

    /// Discard the work, reporting any error.
    pub fn rollback(self) -> Result<()> {
        let Self { runtime, inner } = self;
        runtime.block_on(inner.rollback())
    }

    /// Open a nested transaction, backed by an automatically named savepoint.
    pub fn transaction(&mut self) -> Result<Transaction<'_>> {
        let Self { runtime, inner } = self;
        let tx = runtime.block_on(inner.transaction())?;
        Ok(Transaction { runtime, inner: tx })
    }

    /// Open a nested transaction at a named savepoint.
    pub fn savepoint(&mut self, name: impl Into<String>) -> Result<Transaction<'_>> {
        let Self { runtime, inner } = self;
        let tx = runtime.block_on(inner.savepoint(name))?;
        Ok(Transaction { runtime, inner: tx })
    }

    /// Run a query and collect every row of the first result set.
    pub fn query(&mut self, sql: &str, params: &[&dyn ToSql]) -> Result<Vec<Row>> {
        self.runtime.block_on(self.inner.query(sql, params))
    }

    /// Run a query and read the rows of its first result set as the server
    /// sends them.
    pub fn query_stream(&mut self, sql: &str, params: &[&dyn ToSql]) -> Result<RowIter<'_>> {
        let Self { runtime, inner } = self;
        let rows = runtime.block_on(inner.query_stream(sql, params))?;
        Ok(RowIter {
            runtime,
            inner: rows,
        })
    }

    /// Run a query expecting exactly one row.
    pub fn query_one(&mut self, sql: &str, params: &[&dyn ToSql]) -> Result<Row> {
        self.runtime.block_on(self.inner.query_one(sql, params))
    }

    /// Run a query expecting at most one row.
    pub fn query_opt(&mut self, sql: &str, params: &[&dyn ToSql]) -> Result<Option<Row>> {
        self.runtime.block_on(self.inner.query_opt(sql, params))
    }

    /// Run a query and read the first column of the first row.
    pub fn query_scalar<T: FromSql>(&mut self, sql: &str, params: &[&dyn ToSql]) -> Result<T> {
        self.runtime.block_on(self.inner.query_scalar(sql, params))
    }

    /// Run a statement and report how many rows it affected.
    pub fn execute(&mut self, sql: &str, params: &[&dyn ToSql]) -> Result<u64> {
        self.runtime.block_on(self.inner.execute(sql, params))
    }

    /// Run a SQL batch verbatim and collect its rows.
    pub fn batch(&mut self, sql: &str) -> Result<Vec<Row>> {
        self.runtime.block_on(self.inner.batch(sql))
    }

    /// Run a SQL batch verbatim and collect every result set.
    pub fn batch_dataset(&mut self, sql: &str) -> Result<DataSet> {
        self.runtime.block_on(self.inner.batch_dataset(sql))
    }

    /// Run a [`Command`] and collect every result set it produces.
    pub fn query_dataset(&mut self, command: &Command) -> Result<DataSet> {
        self.runtime.block_on(self.inner.query_dataset(command))
    }

    /// Run a [`Command`] for its row count rather than its rows.
    pub fn execute_command(&mut self, command: &Command) -> Result<u64> {
        self.runtime.block_on(self.inner.execute_command(command))
    }
}

/// The rows of a query's first result set, read as the server sends them.
///
/// The blocking counterpart to [`tdsql::RowStream`](crate::RowStream): an
/// [`Iterator`] where that is a `Stream`, and it holds the connection, and
/// cancels its query if dropped early, in exactly the same way.
pub struct RowIter<'a> {
    runtime: &'a Runtime,
    inner: crate::RowStream<'a>,
}

impl std::fmt::Debug for RowIter<'_> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("blocking::RowIter")
            .field("inner", &self.inner)
            .finish()
    }
}

impl RowIter<'_> {
    /// The columns of the result set, known before the first row arrives.
    ///
    /// Empty if the statement produced no result set at all.
    pub fn columns(&self) -> &[Column] {
        self.inner.columns()
    }
}

impl Iterator for RowIter<'_> {
    type Item = Result<Row>;

    fn next(&mut self) -> Option<Self::Item> {
        self.runtime.block_on(self.inner.try_next()).transpose()
    }
}

mod sealed {
    use tokio::runtime::Runtime;

    /// Grants the default method bodies the runtime and the async connection
    /// underneath.
    ///
    /// Deliberately not public: handing out the inner client from a
    /// `Transaction` would let a caller commit or roll back behind the
    /// transaction's back.
    pub trait Sealed {
        fn parts(&mut self) -> (&Runtime, &mut crate::Client);
    }
}

pub(crate) use sealed::Sealed;

impl Sealed for Client {
    fn parts(&mut self) -> (&Runtime, &mut crate::Client) {
        let Self { runtime, inner } = self;
        (runtime, inner)
    }
}

impl Sealed for Transaction<'_> {
    fn parts(&mut self) -> (&Runtime, &mut crate::Client) {
        let Self { runtime, inner } = self;
        (runtime, inner.client_mut())
    }
}

impl<T: Sealed + ?Sized> Sealed for &mut T {
    fn parts(&mut self) -> (&Runtime, &mut crate::Client) {
        (**self).parts()
    }
}

/// Anything blocking statements can be run against: a [`Client`] or a
/// [`Transaction`].
///
/// The blocking counterpart to [`tdsql::Executor`](crate::Executor). Take
/// `&mut impl Executor` in a helper and the caller decides whether the work
/// lands on the connection directly or inside a transaction:
///
/// ```no_run
/// use tdsql::blocking::{Client, Executor};
/// use tdsql::{Result, Row};
///
/// fn load_user(db: &mut impl Executor, id: i32) -> Result<Row> {
///     db.query_one("SELECT id, name FROM users WHERE id = @P1", &[&id])
/// }
///
/// # fn f(client: &mut Client) -> Result<()> {
/// // Straight to the connection...
/// let user = load_user(client, 1)?;
///
/// // ...or inside a transaction, unchanged.
/// let mut tx = client.transaction()?;
/// let user = load_user(&mut tx, 1)?;
/// tx.commit()?;
/// # Ok(())
/// # }
/// ```
///
/// The trait is sealed — it describes the two types this crate provides rather
/// than an extension point.
pub trait Executor: Sealed {
    /// Open a transaction scoped to this executor.
    ///
    /// On a [`Client`] this begins a real transaction. On a [`Transaction`] it
    /// opens a savepoint instead, so a helper that wants its own atomic scope
    /// nests correctly rather than trying to begin a second transaction on a
    /// connection that already has one.
    fn transaction(&mut self) -> Result<Transaction<'_>>;

    /// Run a query and collect every row of the first result set.
    fn query(&mut self, sql: &str, params: &[&dyn ToSql]) -> Result<Vec<Row>> {
        let (runtime, client) = self.parts();
        runtime.block_on(client.query(sql, params))
    }

    /// Run a query and read the rows of its first result set as the server
    /// sends them.
    fn query_stream(&mut self, sql: &str, params: &[&dyn ToSql]) -> Result<RowIter<'_>> {
        let (runtime, client) = self.parts();
        let inner = runtime.block_on(client.query_stream(sql, params))?;
        Ok(RowIter { runtime, inner })
    }

    /// Run a query expecting exactly one row.
    fn query_one(&mut self, sql: &str, params: &[&dyn ToSql]) -> Result<Row> {
        let (runtime, client) = self.parts();
        runtime.block_on(client.query_one(sql, params))
    }

    /// Run a query expecting at most one row.
    fn query_opt(&mut self, sql: &str, params: &[&dyn ToSql]) -> Result<Option<Row>> {
        let (runtime, client) = self.parts();
        runtime.block_on(client.query_opt(sql, params))
    }

    /// Run a query and read the first column of the first row.
    fn query_scalar<T: FromSql>(&mut self, sql: &str, params: &[&dyn ToSql]) -> Result<T> {
        let (runtime, client) = self.parts();
        runtime.block_on(client.query_scalar(sql, params))
    }

    /// Run a statement and report how many rows it affected.
    fn execute(&mut self, sql: &str, params: &[&dyn ToSql]) -> Result<u64> {
        let (runtime, client) = self.parts();
        runtime.block_on(client.execute(sql, params))
    }

    /// Run a SQL batch verbatim and collect its rows.
    fn batch(&mut self, sql: &str) -> Result<Vec<Row>> {
        let (runtime, client) = self.parts();
        runtime.block_on(client.batch(sql))
    }

    /// Run a SQL batch verbatim and collect every result set.
    fn batch_dataset(&mut self, sql: &str) -> Result<DataSet> {
        let (runtime, client) = self.parts();
        runtime.block_on(client.batch_dataset(sql))
    }

    /// Run a [`Command`] and collect every result set it produces.
    fn query_dataset(&mut self, command: &Command) -> Result<DataSet> {
        let (runtime, client) = self.parts();
        runtime.block_on(client.query_dataset(command))
    }

    /// Run a [`Command`] for its row count rather than its rows.
    fn execute_command(&mut self, command: &Command) -> Result<u64> {
        let (runtime, client) = self.parts();
        runtime.block_on(client.execute_command(command))
    }
}

impl Executor for Client {
    fn transaction(&mut self) -> Result<Transaction<'_>> {
        Client::transaction(self)
    }
}

impl Executor for Transaction<'_> {
    fn transaction(&mut self) -> Result<Transaction<'_>> {
        Transaction::transaction(self)
    }
}

impl<T: Executor + ?Sized> Executor for &mut T {
    fn transaction(&mut self) -> Result<Transaction<'_>> {
        (**self).transaction()
    }
}