dnim/rust-postgres
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Update postgres

Browse Source
This commit is contained in:
Steven Fackler
2019-08-03 18:09:27 -07:00
parent 59923bd425
commit f5a8b1de68
16 changed files with 281 additions and 289 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
postgres/Cargo.toml
View File

@@ -30,9 +30,14 @@ runtime = ["tokio-postgres/runtime", "tokio", "lazy_static", "log"]
[dependencies]
bytes = "0.4"
fallible-iterator = "0.2"
futures = "0.1"
futures-preview = "0.3.0-alpha.17"
pin-utils = "0.1.0-alpha.4"
tokio-postgres = { version = "0.4.0-rc.2", path = "../tokio-postgres", default-features = false }
tokio-executor = { git = "https://github.com/tokio-rs/tokio" }
tokio = { version = "0.1", optional = true }
tokio = { git = "https://github.com/tokio-rs/tokio", optional = true }
lazy_static = { version = "1.0", optional = true }
log = { version = "0.4", optional = true }
[dev-dependencies]
tokio = { git = "https://github.com/tokio-rs/tokio" }

64
postgres/src/client.rs
View File

@@ -1,15 +1,18 @@
use fallible_iterator::FallibleIterator;
use futures::{Async, Future, Poll, Stream};
use std::io::{self, Read};
use futures::executor;
use std::io::{BufRead, Read};
use tokio_postgres::tls::{MakeTlsConnect, TlsConnect};
use tokio_postgres::types::{ToSql, Type};
#[cfg(feature = "runtime")]
use tokio_postgres::Socket;
use tokio_postgres::{Error, Row, SimpleQueryMessage};
use crate::copy_in_stream::CopyInStream;
use crate::copy_out_reader::CopyOutReader;
use crate::iter::Iter;
#[cfg(feature = "runtime")]
use crate::Config;
use crate::{CopyOutReader, QueryIter, SimpleQueryIter, Statement, ToStatement, Transaction};
use crate::{Statement, ToStatement, Transaction};
/// A synchronous PostgreSQL client.
///
@@ -82,7 +85,7 @@ impl Client {
T: ?Sized + ToStatement,
{
let statement = query.__statement(self)?;
self.0.execute(&statement, params).wait()
executor::block_on(self.0.execute(&statement, params))
}
/// Executes a statement, returning the resulting rows.
@@ -149,16 +152,16 @@ impl Client {
/// # Ok(())
/// # }
/// ```
pub fn query_iter<T>(
&mut self,
pub fn query_iter<'a, T>(
&'a mut self,
query: &T,
params: &[&dyn ToSql],
) -> Result<QueryIter<'_>, Error>
) -> Result<impl FallibleIterator<Item = Row, Error = Error> + 'a, Error>
where
T: ?Sized + ToStatement,
{
let statement = query.__statement(self)?;
Ok(QueryIter::new(self.0.query(&statement, params)))
Ok(Iter::new(self.0.query(&statement, params)))
}
/// Creates a new prepared statement.
@@ -185,7 +188,7 @@ impl Client {
/// # }
/// ```
pub fn prepare(&mut self, query: &str) -> Result<Statement, Error> {
self.0.prepare(query).wait()
executor::block_on(self.0.prepare(query))
}
/// Like `prepare`, but allows the types of query parameters to be explicitly specified.
@@ -216,7 +219,7 @@ impl Client {
/// # }
/// ```
pub fn prepare_typed(&mut self, query: &str, types: &[Type]) -> Result<Statement, Error> {
self.0.prepare_typed(query, types).wait()
executor::block_on(self.0.prepare_typed(query, types))
}
/// Executes a `COPY FROM STDIN` statement, returning the number of rows created.
@@ -244,12 +247,10 @@ impl Client {
) -> Result<u64, Error>
where
T: ?Sized + ToStatement,
R: Read,
R: Read + Unpin,
{
let statement = query.__statement(self)?;
self.0
.copy_in(&statement, params, CopyInStream(reader))
.wait()
executor::block_on(self.0.copy_in(&statement, params, CopyInStream(reader)))
}
/// Executes a `COPY TO STDOUT` statement, returning a reader of the resulting data.
@@ -271,11 +272,11 @@ impl Client {
/// # Ok(())
/// # }
/// ```
pub fn copy_out<T>(
&mut self,
pub fn copy_out<'a, T>(
&'a mut self,
query: &T,
params: &[&dyn ToSql],
) -> Result<CopyOutReader<'_>, Error>
) -> Result<impl BufRead + 'a, Error>
where
T: ?Sized + ToStatement,
{
@@ -311,8 +312,11 @@ impl Client {
/// Prepared statements should be use for any query which contains user-specified data, as they provided the
/// functionality to safely imbed that data in the request. Do not form statements via string concatenation and pass
/// them to this method!
pub fn simple_query_iter(&mut self, query: &str) -> Result<SimpleQueryIter<'_>, Error> {
Ok(SimpleQueryIter::new(self.0.simple_query(query)))
pub fn simple_query_iter<'a>(
&'a mut self,
query: &str,
) -> Result<impl FallibleIterator<Item = SimpleQueryMessage, Error = Error> + 'a, Error> {
Ok(Iter::new(self.0.simple_query(query)))
}
/// Begins a new database transaction.
@@ -336,8 +340,8 @@ impl Client {
/// # }
/// ```
pub fn transaction(&mut self) -> Result<Transaction<'_>, Error> {
self.simple_query("BEGIN")?;
Ok(Transaction::new(self))
let transaction = executor::block_on(self.0.transaction())?;
Ok(Transaction::new(transaction))
}
/// Determines if the client's connection has already closed.
@@ -368,21 +372,3 @@ impl From<tokio_postgres::Client> for Client {
Client(c)
}
}
struct CopyInStream<R>(R);
impl<R> Stream for CopyInStream<R>
where
R: Read,
{
type Item = Vec<u8>;
type Error = io::Error;
fn poll(&mut self) -> Poll<Option<Vec<u8>>, io::Error> {
let mut buf = vec![];
match self.0.by_ref().take(4096).read_to_end(&mut buf)? {
0 => Ok(Async::Ready(None)),
_ => Ok(Async::Ready(Some(buf))),
}
}
}

69
postgres/src/config.rs
View File

@@ -2,25 +2,22 @@
//!
//! Requires the `runtime` Cargo feature (enabled by default).
use futures::future::Executor;
use futures::sync::oneshot;
use futures::Future;
use futures::FutureExt;
use log::error;
use std::fmt;
use std::path::Path;
use std::str::FromStr;
use std::sync::Arc;
use std::sync::{mpsc, Arc, Mutex};
use std::time::Duration;
use tokio_executor::Executor;
use tokio_postgres::tls::{MakeTlsConnect, TlsConnect};
use tokio_postgres::{Error, Socket};
#[doc(inline)]
use tokio_postgres::config::{SslMode, TargetSessionAttrs};
pub use tokio_postgres::config::{SslMode, TargetSessionAttrs};
use crate::{Client, RUNTIME};
type DynExecutor = dyn Executor<Box<dyn Future<Item = (), Error = ()> + Send>> + Sync + Send;
/// Connection configuration.
///
/// Configuration can be parsed from libpq-style connection strings. These strings come in two formats:
@@ -98,7 +95,7 @@ type DynExecutor = dyn Executor<Box<dyn Future<Item = (), Error = ()> + Send>> +
pub struct Config {
config: tokio_postgres::Config,
// this is an option since we don't want to boot up our default runtime unless we're actually going to use it.
executor: Option<Arc<DynExecutor>>,
executor: Option<Arc<Mutex<dyn Executor + Sync + Send>>>,
}
impl fmt::Debug for Config {
@@ -242,45 +239,53 @@ impl Config {
/// Defaults to a postgres-specific tokio `Runtime`.
pub fn executor<E>(&mut self, executor: E) -> &mut Config
where
E: Executor<Box<dyn Future<Item = (), Error = ()> + Send>> + 'static + Sync + Send,
E: Executor + 'static + Sync + Send,
{
self.executor = Some(Arc::new(executor));
self.executor = Some(Arc::new(Mutex::new(executor)));
self
}
/// Opens a connection to a PostgreSQL database.
pub fn connect<T>(&self, tls_mode: T) -> Result<Client, Error>
pub fn connect<T>(&self, tls: T) -> Result<Client, Error>
where
T: MakeTlsConnect<Socket> + 'static + Send,
T::TlsConnect: Send,
T::Stream: Send,
<T::TlsConnect as TlsConnect<Socket>>::Future: Send,
{
let (tx, rx) = oneshot::channel();
let connect = self
.config
.connect(tls_mode)
.then(|r| tx.send(r).map_err(|_| ()));
self.with_executor(|e| e.execute(Box::new(connect)))
.unwrap();
let (client, connection) = rx.wait().unwrap()?;
let (client, connection) = match &self.executor {
Some(executor) => {
let (tx, rx) = mpsc::channel();
let config = self.config.clone();
let connect = async move {
let r = config.connect(tls).await;
let _ = tx.send(r);
};
executor.lock().unwrap().spawn(Box::pin(connect)).unwrap();
rx.recv().unwrap()?
}
None => {
let connect = self.config.connect(tls);
RUNTIME.block_on(connect)?
}
};
let connection = connection.map_err(|e| error!("postgres connection error: {}", e));
self.with_executor(|e| e.execute(Box::new(connection)))
.unwrap();
let connection = connection.map(|r| {
if let Err(e) = r {
error!("postgres connection error: {}", e)
}
});
match &self.executor {
Some(executor) => {
executor.lock().unwrap().spawn(Box::pin(connection)).unwrap();
},
None => {
RUNTIME.spawn(connection);
}
}
Ok(Client::from(client))
}
fn with_executor<F, T>(&self, f: F) -> T
where
F: FnOnce(&dyn Executor<Box<dyn Future<Item = (), Error = ()> + Send>>) -> T,
{
match &self.executor {
Some(e) => f(&**e),
None => f(&RUNTIME.executor()),
}
}
}
impl FromStr for Config {

25
postgres/src/copy_in_stream.rs Normal file
View File

@@ -0,0 +1,25 @@
use futures::Stream;
use std::io;
use std::io::Read;
use std::pin::Pin;
use std::task::{Context, Poll};
pub struct CopyInStream<R>(pub R);
impl<R> Stream for CopyInStream<R>
where
R: Read + Unpin,
{
type Item = io::Result<Vec<u8>>;
fn poll_next(
mut self: Pin<&mut Self>,
_: &mut Context<'_>,
) -> Poll<Option<io::Result<Vec<u8>>>> {
let mut buf = vec![];
match self.0.by_ref().take(4096).read_to_end(&mut buf)? {
0 => Poll::Ready(None),
_ => Poll::Ready(Some(Ok(buf))),
}
}
}

36
postgres/src/copy_out_reader.rs
View File

@@ -1,26 +1,34 @@
use bytes::{Buf, Bytes};
use futures::stream::{self, Stream};
use futures::{executor, Stream};
use std::io::{self, BufRead, Cursor, Read};
use std::marker::PhantomData;
use tokio_postgres::impls;
use std::pin::Pin;
use tokio_postgres::Error;
/// The reader returned by the `copy_out` method.
pub struct CopyOutReader<'a> {
it: stream::Wait<impls::CopyOut>,
pub struct CopyOutReader<'a, S>
where
S: Stream,
{
it: executor::BlockingStream<Pin<Box<S>>>,
cur: Cursor<Bytes>,
_p: PhantomData<&'a mut ()>,
}
// no-op impl to extend borrow until drop
impl<'a> Drop for CopyOutReader<'a> {
impl<'a, S> Drop for CopyOutReader<'a, S>
where
S: Stream,
{
fn drop(&mut self) {}
}
impl<'a> CopyOutReader<'a> {
#[allow(clippy::new_ret_no_self)]
pub(crate) fn new(stream: impls::CopyOut) -> Result<CopyOutReader<'a>, Error> {
let mut it = stream.wait();
impl<'a, S> CopyOutReader<'a, S>
where
S: Stream<Item = Result<Bytes, Error>>,
{
pub(crate) fn new(stream: S) -> Result<CopyOutReader<'a, S>, Error> {
let mut it = executor::block_on_stream(Box::pin(stream));
let cur = match it.next() {
Some(Ok(cur)) => cur,
Some(Err(e)) => return Err(e),
@@ -35,7 +43,10 @@ impl<'a> CopyOutReader<'a> {
}
}
impl<'a> Read for CopyOutReader<'a> {
impl<'a, S> Read for CopyOutReader<'a, S>
where
S: Stream<Item = Result<Bytes, Error>>,
{
fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
let b = self.fill_buf()?;
let len = usize::min(buf.len(), b.len());
@@ -45,7 +56,10 @@ impl<'a> Read for CopyOutReader<'a> {
}
}
impl<'a> BufRead for CopyOutReader<'a> {
impl<'a, S> BufRead for CopyOutReader<'a, S>
where
S: Stream<Item = Result<Bytes, Error>>,
{
fn fill_buf(&mut self) -> io::Result<&[u8]> {
if self.cur.remaining() == 0 {
match self.it.next() {

45
postgres/src/iter.rs Normal file
View File

@@ -0,0 +1,45 @@
use fallible_iterator::FallibleIterator;
use futures::executor::{self, BlockingStream};
use futures::Stream;
use std::marker::PhantomData;
use std::pin::Pin;
pub struct Iter<'a, S>
where
S: Stream,
{
it: BlockingStream<Pin<Box<S>>>,
_p: PhantomData<&'a mut ()>,
}
// no-op impl to extend the borrow until drop
impl<'a, S> Drop for Iter<'a, S>
where
S: Stream,
{
fn drop(&mut self) {}
}
impl<'a, S> Iter<'a, S>
where
S: Stream,
{
pub fn new(stream: S) -> Iter<'a, S> {
Iter {
it: executor::block_on_stream(Box::pin(stream)),
_p: PhantomData,
}
}
}
impl<'a, S, T, E> FallibleIterator for Iter<'a, S>
where
S: Stream<Item = Result<T, E>>,
{
type Item = T;
type Error = E;
fn next(&mut self) -> Result<Option<T>, E> {
self.it.next().transpose()
}
}

10
postgres/src/lib.rs
View File

@@ -54,6 +54,7 @@
//! crates, respectively.
#![doc(html_root_url = "https://docs.rs/postgres/0.16.0-rc.2")]
#![warn(clippy::all, rust_2018_idioms, missing_docs)]
#![feature(async_await)]
#[cfg(feature = "runtime")]
use lazy_static::lazy_static;
@@ -69,14 +70,10 @@ pub use tokio_postgres::{
pub use crate::client::*;
#[cfg(feature = "runtime")]
pub use crate::config::Config;
pub use crate::copy_out_reader::*;
#[doc(no_inline)]
pub use crate::error::Error;
pub use crate::query_iter::*;
pub use crate::query_portal_iter::*;
#[doc(no_inline)]
pub use crate::row::{Row, SimpleQueryRow};
pub use crate::simple_query_iter::*;
#[doc(no_inline)]
pub use crate::tls::NoTls;
pub use crate::to_statement::*;
@@ -85,10 +82,9 @@ pub use crate::transaction::*;
mod client;
#[cfg(feature = "runtime")]
pub mod config;
mod copy_in_stream;
mod copy_out_reader;
mod query_iter;
mod query_portal_iter;
mod simple_query_iter;
mod iter;
mod to_statement;
mod transaction;

38
postgres/src/query_iter.rs
View File

@@ -1,38 +0,0 @@
use fallible_iterator::FallibleIterator;
use futures::stream::{self, Stream};
use std::marker::PhantomData;
use tokio_postgres::impls;
use tokio_postgres::{Error, Row};
/// The iterator returned by the `query_iter` method.
pub struct QueryIter<'a> {
it: stream::Wait<impls::Query>,
_p: PhantomData<&'a mut ()>,
}
// no-op impl to extend the borrow until drop
impl<'a> Drop for QueryIter<'a> {
fn drop(&mut self) {}
}
impl<'a> QueryIter<'a> {
pub(crate) fn new(stream: impls::Query) -> QueryIter<'a> {
QueryIter {
it: stream.wait(),
_p: PhantomData,
}
}
}
impl<'a> FallibleIterator for QueryIter<'a> {
type Item = Row;
type Error = Error;
fn next(&mut self) -> Result<Option<Row>, Error> {
match self.it.next() {
Some(Ok(row)) => Ok(Some(row)),
Some(Err(e)) => Err(e),
None => Ok(None),
}
}
}

38
postgres/src/query_portal_iter.rs
View File

@@ -1,38 +0,0 @@
use fallible_iterator::FallibleIterator;
use futures::stream::{self, Stream};
use std::marker::PhantomData;
use tokio_postgres::impls;
use tokio_postgres::{Error, Row};
/// The iterator returned by the `query_portal_iter` method.
pub struct QueryPortalIter<'a> {
it: stream::Wait<impls::QueryPortal>,
_p: PhantomData<&'a mut ()>,
}
// no-op impl to extend the borrow until drop
impl<'a> Drop for QueryPortalIter<'a> {
fn drop(&mut self) {}
}
impl<'a> QueryPortalIter<'a> {
pub(crate) fn new(stream: impls::QueryPortal) -> QueryPortalIter<'a> {
QueryPortalIter {
it: stream.wait(),
_p: PhantomData,
}
}
}
impl<'a> FallibleIterator for QueryPortalIter<'a> {
type Item = Row;
type Error = Error;
fn next(&mut self) -> Result<Option<Row>, Error> {
match self.it.next() {
Some(Ok(row)) => Ok(Some(row)),
Some(Err(e)) => Err(e),
None => Ok(None),
}
}
}

38
postgres/src/simple_query_iter.rs
View File

@@ -1,38 +0,0 @@
use fallible_iterator::FallibleIterator;
use futures::stream::{self, Stream};
use std::marker::PhantomData;
use tokio_postgres::impls;
use tokio_postgres::{Error, SimpleQueryMessage};
/// The iterator returned by the `simple_query_iter` method.
pub struct SimpleQueryIter<'a> {
it: stream::Wait<impls::SimpleQuery>,
_p: PhantomData<&'a mut ()>,
}
// no-op impl to extend borrow until drop
impl<'a> Drop for SimpleQueryIter<'a> {
fn drop(&mut self) {}
}
impl<'a> SimpleQueryIter<'a> {
pub(crate) fn new(stream: impls::SimpleQuery) -> SimpleQueryIter<'a> {
SimpleQueryIter {
it: stream.wait(),
_p: PhantomData,
}
}
}
impl<'a> FallibleIterator for SimpleQueryIter<'a> {
type Item = SimpleQueryMessage;
type Error = Error;
fn next(&mut self) -> Result<Option<SimpleQueryMessage>, Error> {
match self.it.next() {
Some(Ok(row)) => Ok(Some(row)),
Some(Err(e)) => Err(e),
None => Ok(None),
}
}
}

21
postgres/src/test.rs
View File

@@ -1,4 +1,5 @@
use std::io::Read;
use tokio::runtime::Runtime;
use tokio_postgres::types::Type;
use tokio_postgres::NoTls;
@@ -95,6 +96,7 @@ fn transaction_drop() {
assert_eq!(rows.len(), 0);
}
/*
#[test]
fn nested_transactions() {
let mut client = Client::connect("host=localhost port=5433 user=postgres", NoTls).unwrap();
@@ -145,6 +147,7 @@ fn nested_transactions() {
assert_eq!(rows[1].get::<_, i32>(0), 3);
assert_eq!(rows[2].get::<_, i32>(0), 4);
}
*/
#[test]
fn copy_in() {
@@ -222,3 +225,21 @@ fn portal() {
assert_eq!(rows.len(), 1);
assert_eq!(rows[0].get::<_, i32>(0), 3);
}
#[test]
fn custom_executor() {
let runtime = Runtime::new().unwrap();
let mut config = "host=localhost port=5433 user=postgres"
.parse::<crate::Config>()
.unwrap();
config.executor(runtime.executor());
let mut client = config.connect(NoTls).unwrap();
let rows = client.query("SELECT $1::TEXT", &[&"hello"]).unwrap();
assert_eq!(rows.len(), 1);
assert_eq!(rows[0].get::<_, &str>(0), "hello");
drop(runtime);
assert!(client.is_closed());
}

33
postgres/src/to_statement.rs
View File

@@ -1,11 +1,28 @@
use tokio_postgres::Error;
use crate::{Client, Statement};
use crate::{Client, Statement, Transaction};
mod sealed {
pub trait Sealed {}
}
#[doc(hidden)]
pub trait Prepare {
fn prepare(&mut self, query: &str) -> Result<Statement, Error>;
}
impl Prepare for Client {
fn prepare(&mut self, query: &str) -> Result<Statement, Error> {
self.prepare(query)
}
}
impl<'a> Prepare for Transaction<'a> {
fn prepare(&mut self, query: &str) -> Result<Statement, Error> {
self.prepare(query)
}
}
/// A trait abstracting over prepared and unprepared statements.
///
/// Many methods are generic over this bound, so that they support both a raw query string as well as a statement which
@@ -14,13 +31,18 @@ mod sealed {
/// This trait is "sealed" and cannot be implemented by anything outside this crate.
pub trait ToStatement: sealed::Sealed {
#[doc(hidden)]
fn __statement(&self, client: &mut Client) -> Result<Statement, Error>;
fn __statement<T>(&self, client: &mut T) -> Result<Statement, Error>
where
T: Prepare;
}
impl sealed::Sealed for str {}
impl ToStatement for str {
fn __statement(&self, client: &mut Client) -> Result<Statement, Error> {
fn __statement<T>(&self, client: &mut T) -> Result<Statement, Error>
where
T: Prepare,
{
client.prepare(self)
}
}
@@ -28,7 +50,10 @@ impl ToStatement for str {
impl sealed::Sealed for Statement {}
impl ToStatement for Statement {
fn __statement(&self, _: &mut Client) -> Result<Statement, Error> {
fn __statement<T>(&self, _: &mut T) -> Result<Statement, Error>
where
T: Prepare,
{
Ok(self.clone())
}
}

130
postgres/src/transaction.rs
View File

@@ -1,79 +1,45 @@
use fallible_iterator::FallibleIterator;
use futures::Future;
use std::io::Read;
use futures::executor;
use std::io::{BufRead, Read};
use tokio_postgres::types::{ToSql, Type};
use tokio_postgres::{Error, Row, SimpleQueryMessage};
use crate::{
Client, CopyOutReader, Portal, QueryIter, QueryPortalIter, SimpleQueryIter, Statement,
ToStatement,
};
use crate::copy_in_stream::CopyInStream;
use crate::copy_out_reader::CopyOutReader;
use crate::iter::Iter;
use crate::{Portal, Statement, ToStatement};
/// A representation of a PostgreSQL database transaction.
///
/// Transactions will implicitly roll back by default when dropped. Use the `commit` method to commit the changes made
/// in the transaction. Transactions can be nested, with inner transactions implemented via safepoints.
pub struct Transaction<'a> {
client: &'a mut Client,
depth: u32,
done: bool,
}
impl<'a> Drop for Transaction<'a> {
fn drop(&mut self) {
if !self.done {
let _ = self.rollback_inner();
}
}
}
pub struct Transaction<'a>(tokio_postgres::Transaction<'a>);
impl<'a> Transaction<'a> {
pub(crate) fn new(client: &'a mut Client) -> Transaction<'a> {
Transaction {
client,
depth: 0,
done: false,
}
pub(crate) fn new(transaction: tokio_postgres::Transaction<'a>) -> Transaction<'a> {
Transaction(transaction)
}
/// Consumes the transaction, committing all changes made within it.
pub fn commit(mut self) -> Result<(), Error> {
self.done = true;
if self.depth == 0 {
self.client.simple_query("COMMIT")?;
} else {
self.client
.simple_query(&format!("RELEASE sp{}", self.depth))?;
}
Ok(())
pub fn commit(self) -> Result<(), Error> {
executor::block_on(self.0.commit())
}
/// Rolls the transaction back, discarding all changes made within it.
///
/// This is equivalent to `Transaction`'s `Drop` implementation, but provides any error encountered to the caller.
pub fn rollback(mut self) -> Result<(), Error> {
self.done = true;
self.rollback_inner()
}
fn rollback_inner(&mut self) -> Result<(), Error> {
if self.depth == 0 {
self.client.simple_query("ROLLBACK")?;
} else {
self.client
.simple_query(&format!("ROLLBACK TO sp{}", self.depth))?;
}
Ok(())
pub fn rollback(self) -> Result<(), Error> {
executor::block_on(self.0.rollback())
}
/// Like `Client::prepare`.
pub fn prepare(&mut self, query: &str) -> Result<Statement, Error> {
self.client.prepare(query)
executor::block_on(self.0.prepare(query))
}
/// Like `Client::prepare_typed`.
pub fn prepare_typed(&mut self, query: &str, types: &[Type]) -> Result<Statement, Error> {
self.client.prepare_typed(query, types)
executor::block_on(self.0.prepare_typed(query, types))
}
/// Like `Client::execute`.
@@ -81,7 +47,8 @@ impl<'a> Transaction<'a> {
where
T: ?Sized + ToStatement,
{
self.client.execute(query, params)
let statement = query.__statement(self)?;
executor::block_on(self.0.execute(&statement, params))
}
/// Like `Client::query`.
@@ -89,7 +56,7 @@ impl<'a> Transaction<'a> {
where
T: ?Sized + ToStatement,
{
self.client.query(query, params)
self.query_iter(query, params)?.collect()
}
/// Like `Client::query_iter`.
@@ -97,11 +64,12 @@ impl<'a> Transaction<'a> {
&mut self,
query: &T,
params: &[&dyn ToSql],
) -> Result<QueryIter<'_>, Error>
) -> Result<impl FallibleIterator<Item = Row, Error = Error>, Error>
where
T: ?Sized + ToStatement,
{
self.client.query_iter(query, params)
let statement = query.__statement(self)?;
Ok(Iter::new(self.0.query(&statement, params)))
}
/// Binds parameters to a statement, creating a "portal".
@@ -118,8 +86,8 @@ impl<'a> Transaction<'a> {
where
T: ?Sized + ToStatement,
{
let statement = query.__statement(&mut self.client)?;
self.client.get_mut().bind(&statement, params).wait()
let statement = query.__statement(self)?;
executor::block_on(self.0.bind(&statement, params))
}
/// Continues execution of a portal, returning the next set of rows.
@@ -136,10 +104,8 @@ impl<'a> Transaction<'a> {
&mut self,
portal: &Portal,
max_rows: i32,
) -> Result<QueryPortalIter<'_>, Error> {
Ok(QueryPortalIter::new(
self.client.get_mut().query_portal(&portal, max_rows),
))
) -> Result<impl FallibleIterator<Item = Row, Error = Error>, Error> {
Ok(Iter::new(self.0.query_portal(&portal, max_rows)))
}
/// Like `Client::copy_in`.
@@ -151,42 +117,48 @@ impl<'a> Transaction<'a> {
) -> Result<u64, Error>
where
T: ?Sized + ToStatement,
R: Read,
R: Read + Unpin,
{
self.client.copy_in(query, params, reader)
let statement = query.__statement(self)?;
executor::block_on(self.0.copy_in(&statement, params, CopyInStream(reader)))
}
/// Like `Client::copy_out`.
pub fn copy_out<T>(
&mut self,
pub fn copy_out<'b, T>(
&'a mut self,
query: &T,
params: &[&dyn ToSql],
) -> Result<CopyOutReader<'_>, Error>
) -> Result<impl BufRead + 'b, Error>
where
T: ?Sized + ToStatement,
{
self.client.copy_out(query, params)
let statement = query.__statement(self)?;
let stream = self.0.copy_out(&statement, params);
CopyOutReader::new(stream)
}
/// Like `Client::simple_query`.
pub fn simple_query(&mut self, query: &str) -> Result<Vec<SimpleQueryMessage>, Error> {
self.client.simple_query(query)
self.simple_query_iter(query)?.collect()
}
/// Like `Client::simple_query_iter`.
pub fn simple_query_iter(&mut self, query: &str) -> Result<SimpleQueryIter<'_>, Error> {
self.client.simple_query_iter(query)
pub fn simple_query_iter<'b>(
&'b mut self,
query: &str,
) -> Result<impl FallibleIterator<Item = SimpleQueryMessage, Error = Error> + 'b, Error> {
Ok(Iter::new(self.0.simple_query(query)))
}
/// Like `Client::transaction`.
pub fn transaction(&mut self) -> Result<Transaction<'_>, Error> {
let depth = self.depth + 1;
self.client
.simple_query(&format!("SAVEPOINT sp{}", depth))?;
Ok(Transaction {
client: self.client,
depth,
done: false,
})
}
// /// Like `Client::transaction`.
// pub fn transaction(&mut self) -> Result<Transaction<'_>, Error> {
// let depth = self.depth + 1;
// self.client
// .simple_query(&format!("SAVEPOINT sp{}", depth))?;
// Ok(Transaction {
// client: self.client,
// depth,
// done: false,
// })
// }
}