From fed246e9fdb9d6579b581b8bac2e25f49e206b1d Mon Sep 17 00:00:00 2001 From: Steven Fackler Date: Wed, 19 Jul 2017 21:22:27 -0700 Subject: [PATCH] Error reform for tokio-postgres --- postgres-shared/src/error/mod.rs | 140 +++++++++--- postgres/src/error.rs | 112 --------- postgres/src/lib.rs | 118 +++++----- postgres/src/priv_io.rs | 10 +- postgres/src/rows.rs | 8 +- tokio-postgres/src/error.rs | 59 ----- tokio-postgres/src/lib.rs | 366 +++++++++++++++--------------- tokio-postgres/src/sink.rs | 164 +++++++++++++ tokio-postgres/src/stream.rs | 20 +- tokio-postgres/src/test.rs | 23 +- tokio-postgres/src/transaction.rs | 22 +- 11 files changed, 542 insertions(+), 500 deletions(-) delete mode 100644 postgres/src/error.rs delete mode 100644 tokio-postgres/src/error.rs create mode 100644 tokio-postgres/src/sink.rs diff --git a/postgres-shared/src/error/mod.rs b/postgres-shared/src/error/mod.rs index 8d07b95f..6fe73597 100644 --- a/postgres-shared/src/error/mod.rs +++ b/postgres-shared/src/error/mod.rs @@ -307,59 +307,131 @@ pub enum ErrorPosition { }, } -/// Reasons a new Postgres connection could fail. -#[derive(Debug)] -pub enum ConnectError { - /// An error relating to connection parameters. - ConnectParams(Box), - /// An error from the Postgres server itself. - Db(Box), - /// An error initializing the TLS session. - Tls(Box), - /// An error communicating with the server. - Io(io::Error), +#[doc(hidden)] +pub fn connect(e: Box) -> Error { + Error(Box::new(ErrorKind::ConnectParams(e))) } -impl fmt::Display for ConnectError { +#[doc(hidden)] +pub fn tls(e: Box) -> Error { + Error(Box::new(ErrorKind::Tls(e))) +} + +#[doc(hidden)] +pub fn db(e: DbError) -> Error { + Error(Box::new(ErrorKind::Db(e))) +} + +#[doc(hidden)] +pub fn io(e: io::Error) -> Error { + Error(Box::new(ErrorKind::Io(e))) +} + +#[doc(hidden)] +pub fn conversion(e: Box) -> Error { + Error(Box::new(ErrorKind::Conversion(e))) +} + +#[derive(Debug)] +enum ErrorKind { + ConnectParams(Box), + Tls(Box), + Db(DbError), + Io(io::Error), + Conversion(Box), +} + +/// An error communicating with the Postgres server. +#[derive(Debug)] +pub struct Error(Box); + +impl fmt::Display for Error { fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result { fmt.write_str(error::Error::description(self))?; - match *self { - ConnectError::ConnectParams(ref msg) => write!(fmt, ": {}", msg), - ConnectError::Db(ref err) => write!(fmt, ": {}", err), - ConnectError::Tls(ref err) => write!(fmt, ": {}", err), - ConnectError::Io(ref err) => write!(fmt, ": {}", err), + match *self.0 { + ErrorKind::ConnectParams(ref err) => write!(fmt, ": {}", err), + ErrorKind::Tls(ref err) => write!(fmt, ": {}", err), + ErrorKind::Db(ref err) => write!(fmt, ": {}", err), + ErrorKind::Io(ref err) => write!(fmt, ": {}", err), + ErrorKind::Conversion(ref err) => write!(fmt, ": {}", err), } } } -impl error::Error for ConnectError { +impl error::Error for Error { fn description(&self) -> &str { - match *self { - ConnectError::ConnectParams(_) => "Invalid connection parameters", - ConnectError::Db(_) => "Error reported by Postgres", - ConnectError::Tls(_) => "Error initiating SSL session", - ConnectError::Io(_) => "Error communicating with the server", + match *self.0 { + ErrorKind::ConnectParams(_) => "invalid connection parameters", + ErrorKind::Tls(_) => "TLS handshake error", + ErrorKind::Db(_) => "database error", + ErrorKind::Io(_) => "IO error", + ErrorKind::Conversion(_) => "type conversion error", } } fn cause(&self) -> Option<&error::Error> { - match *self { - ConnectError::ConnectParams(ref err) | - ConnectError::Tls(ref err) => Some(&**err), - ConnectError::Db(ref err) => Some(&**err), - ConnectError::Io(ref err) => Some(err), + match *self.0 { + ErrorKind::ConnectParams(ref err) => Some(&**err), + ErrorKind::Tls(ref err) => Some(&**err), + ErrorKind::Db(ref err) => Some(err), + ErrorKind::Io(ref err) => Some(err), + ErrorKind::Conversion(ref err) => Some(&**err), } } } -impl From for ConnectError { - fn from(err: io::Error) -> ConnectError { - ConnectError::Io(err) +impl Error { + /// Returns the SQLSTATE error code associated with this error if it is a DB + /// error. + pub fn code(&self) -> Option<&SqlState> { + self.as_db().map(|e| &e.code) + } + + /// Returns the inner error if this is a connection parameter error. + pub fn as_connection(&self) -> Option<&(error::Error + 'static + Sync + Send)> { + match *self.0 { + ErrorKind::ConnectParams(ref err) => Some(&**err), + _ => None, + } + } + + /// Returns the `DbError` associated with this error if it is a DB error. + pub fn as_db(&self) -> Option<&DbError> { + match *self.0 { + ErrorKind::Db(ref err) => Some(err), + _ => None + } + } + + /// Returns the inner error if this is a conversion error. + pub fn as_conversion(&self) -> Option<&(error::Error + 'static + Sync + Send)> { + match *self.0 { + ErrorKind::Conversion(ref err) => Some(&**err), + _ => None, + } + } + + /// Returns the inner `io::Error` associated with this error if it is an IO + /// error. + pub fn as_io(&self) -> Option<&io::Error> { + match *self.0 { + ErrorKind::Io(ref err) => Some(err), + _ => None, + } } } -impl From for ConnectError { - fn from(err: DbError) -> ConnectError { - ConnectError::Db(Box::new(err)) +impl From for Error { + fn from(err: io::Error) -> Error { + Error(Box::new(ErrorKind::Io(err))) + } +} + +impl From for io::Error { + fn from(err: Error) -> io::Error { + match *err.0 { + ErrorKind::Io(e) => e, + _ => io::Error::new(io::ErrorKind::Other, err), + } } } diff --git a/postgres/src/error.rs b/postgres/src/error.rs deleted file mode 100644 index a7cceed8..00000000 --- a/postgres/src/error.rs +++ /dev/null @@ -1,112 +0,0 @@ -//! Error types. -use std::fmt; -use std::io; -use std::error; - -#[doc(inline)] -// FIXME -pub use postgres_shared::error::*; - -#[derive(Debug)] -pub(crate) enum ErrorKind { - ConnectParams(Box), - Tls(Box), - Db(DbError), - Io(io::Error), - Conversion(Box), -} - -/// An error communicating with the Postgres server. -#[derive(Debug)] -pub struct Error(pub(crate) Box); - -impl fmt::Display for Error { - fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result { - fmt.write_str(error::Error::description(self))?; - match *self.0 { - ErrorKind::ConnectParams(ref err) => write!(fmt, ": {}", err), - ErrorKind::Tls(ref err) => write!(fmt, ": {}", err), - ErrorKind::Db(ref err) => write!(fmt, ": {}", err), - ErrorKind::Io(ref err) => write!(fmt, ": {}", err), - ErrorKind::Conversion(ref err) => write!(fmt, ": {}", err), - } - } -} - -impl error::Error for Error { - fn description(&self) -> &str { - match *self.0 { - ErrorKind::ConnectParams(_) => "invalid connection parameters", - ErrorKind::Tls(_) => "TLS handshake error", - ErrorKind::Db(_) => "database error", - ErrorKind::Io(_) => "IO error", - ErrorKind::Conversion(_) => "type conversion error", - } - } - - fn cause(&self) -> Option<&error::Error> { - match *self.0 { - ErrorKind::ConnectParams(ref err) => Some(&**err), - ErrorKind::Tls(ref err) => Some(&**err), - ErrorKind::Db(ref err) => Some(err), - ErrorKind::Io(ref err) => Some(err), - ErrorKind::Conversion(ref err) => Some(&**err), - } - } -} - -impl Error { - /// Returns the SQLSTATE error code associated with this error if it is a DB - /// error. - pub fn code(&self) -> Option<&SqlState> { - self.as_db().map(|e| &e.code) - } - - /// Returns the inner error if this is a connection parameter error. - pub fn as_connection(&self) -> Option<&(error::Error + 'static + Sync + Send)> { - match *self.0 { - ErrorKind::ConnectParams(ref err) => Some(&**err), - _ => None, - } - } - - /// Returns the `DbError` associated with this error if it is a DB error. - pub fn as_db(&self) -> Option<&DbError> { - match *self.0 { - ErrorKind::Db(ref err) => Some(err), - _ => None - } - } - - /// Returns the inner error if this is a conversion error. - pub fn as_conversion(&self) -> Option<&(error::Error + 'static + Sync + Send)> { - match *self.0 { - ErrorKind::Conversion(ref err) => Some(&**err), - _ => None, - } - } - - /// Returns the inner `io::Error` associated with this error if it is an IO - /// error. - pub fn as_io(&self) -> Option<&io::Error> { - match *self.0 { - ErrorKind::Io(ref err) => Some(err), - _ => None, - } - } -} - -impl From for Error { - fn from(err: io::Error) -> Error { - Error(Box::new(ErrorKind::Io(err))) - } -} - -impl From for io::Error { - fn from(err: Error) -> io::Error { - match *err.0 { - ErrorKind::Io(e) => e, - _ => io::Error::new(io::ErrorKind::Other, err), - } - } -} diff --git a/postgres/src/lib.rs b/postgres/src/lib.rs index 3685ab74..af876704 100644 --- a/postgres/src/lib.rs +++ b/postgres/src/lib.rs @@ -93,7 +93,7 @@ use postgres_protocol::message::backend::{self, ErrorFields}; use postgres_protocol::message::frontend; use postgres_shared::rows::RowData; -use error::{ErrorKind, DbError, UNDEFINED_COLUMN, UNDEFINED_TABLE}; +use error::{DbError, UNDEFINED_COLUMN, UNDEFINED_TABLE}; use tls::TlsHandshake; use notification::{Notifications, Notification}; use params::{IntoConnectParams, User}; @@ -106,6 +106,8 @@ use types::{IsNull, Kind, Type, Oid, ToSql, FromSql, Field, OID, NAME, CHAR}; #[doc(inline)] pub use postgres_shared::CancelData; #[doc(inline)] +pub use postgres_shared::error; +#[doc(inline)] pub use error::Error; #[macro_use] @@ -113,7 +115,6 @@ mod macros; mod feature_check; mod priv_io; -pub mod error; pub mod tls; pub mod notification; pub mod params; @@ -184,9 +185,7 @@ pub fn cancel_query(params: T, tls: TlsMode, data: &CancelData) -> Result<()> where T: IntoConnectParams, { - let params = params.into_connect_params().map_err(|e| { - Error(Box::new(ErrorKind::ConnectParams(e))) - })?; + let params = params.into_connect_params().map_err(error::connect)?; let mut socket = priv_io::initialize_stream(¶ms, tls)?; let mut buf = vec![]; @@ -259,17 +258,15 @@ impl InnerConnection { where T: IntoConnectParams, { - let params = params.into_connect_params().map_err(|e| { - Error(Box::new(ErrorKind::ConnectParams(e))) - })?; + let params = params.into_connect_params().map_err(error::connect)?; let stream = priv_io::initialize_stream(¶ms, tls)?; let user = match params.user() { Some(user) => user, None => { - return Err(Error(Box::new(ErrorKind::ConnectParams( - "User missing from connection parameters".into(), - )))); + return Err(error::connect( + "user missing from connection parameters".into(), + )); } }; @@ -414,9 +411,7 @@ impl InnerConnection { backend::Message::AuthenticationOk => return Ok(()), backend::Message::AuthenticationCleartextPassword => { let pass = user.password().ok_or_else(|| { - Error(Box::new(ErrorKind::ConnectParams( - "a password was requested but not provided".into(), - ))) + error::connect("a password was requested but not provided".into()) })?; self.stream.write_message( |buf| frontend::password_message(pass, buf), @@ -425,9 +420,7 @@ impl InnerConnection { } backend::Message::AuthenticationMd5Password(body) => { let pass = user.password().ok_or_else(|| { - Error(Box::new(ErrorKind::ConnectParams( - "a password was requested but not provided".into(), - ))) + error::connect("a password was requested but not provided".into()) })?; let output = authentication::md5_hash(user.name().as_bytes(), pass.as_bytes(), body.salt()); @@ -442,16 +435,14 @@ impl InnerConnection { .filter(|m| *m == sasl::SCRAM_SHA_256) .count()? == 0 { - return Err(io::Error::new( - io::ErrorKind::Other, - "unsupported authentication", - ).into()); + return Err( + io::Error::new(io::ErrorKind::Other, "unsupported authentication") + .into(), + ); } let pass = user.password().ok_or_else(|| { - Error(Box::new(ErrorKind::ConnectParams( - "a password was requested but not provided".into(), - ))) + error::connect("a password was requested but not provided".into()) })?; let mut scram = ScramSha256::new(pass.as_bytes())?; @@ -463,9 +454,7 @@ impl InnerConnection { let body = match self.read_message()? { backend::Message::AuthenticationSaslContinue(body) => body, - backend::Message::ErrorResponse(body) => { - return Err(err(&mut body.fields())) - } + backend::Message::ErrorResponse(body) => return Err(err(&mut body.fields())), _ => return Err(bad_response().into()), }; @@ -478,9 +467,7 @@ impl InnerConnection { let body = match self.read_message()? { backend::Message::AuthenticationSaslFinal(body) => body, - backend::Message::ErrorResponse(body) => { - return Err(err(&mut body.fields())) - } + backend::Message::ErrorResponse(body) => return Err(err(&mut body.fields())), _ => return Err(bad_response().into()), }; @@ -490,10 +477,9 @@ impl InnerConnection { backend::Message::AuthenticationScmCredential | backend::Message::AuthenticationGss | backend::Message::AuthenticationSspi => { - return Err(io::Error::new( - io::ErrorKind::Other, - "unsupported authentication", - ).into()) + return Err( + io::Error::new(io::ErrorKind::Other, "unsupported authentication").into(), + ) } backend::Message::ErrorResponse(body) => return Err(err(&mut body.fields())), _ => return Err(bad_response().into()), @@ -662,7 +648,7 @@ impl InnerConnection { match r { Ok(()) => {} Err(frontend::BindError::Conversion(e)) => { - return Err(Error(Box::new(ErrorKind::Conversion(e)))) + return Err(error::conversion(e)); } Err(frontend::BindError::Serialization(e)) => return Err(e.into()), } @@ -807,29 +793,27 @@ impl InnerConnection { let get_raw = |i: usize| row.as_ref().and_then(|r| r.get(i)); let (name, type_, elem_oid, rngsubtype, basetype, schema, relid) = { - let name = String::from_sql_nullable(&NAME, get_raw(0)).map_err(|e| { - Error(Box::new(ErrorKind::Conversion(e))) - })?; - let type_ = i8::from_sql_nullable(&CHAR, get_raw(1)).map_err(|e| { - Error(Box::new(ErrorKind::Conversion(e))) - })?; - let elem_oid = Oid::from_sql_nullable(&OID, get_raw(2)).map_err(|e| { - Error(Box::new(ErrorKind::Conversion(e))) - })?; - let rngsubtype = Option::::from_sql_nullable(&OID, get_raw(3)).map_err( - |e| { - Error(Box::new(ErrorKind::Conversion(e))) - }, + let name = String::from_sql_nullable(&NAME, get_raw(0)).map_err( + error::conversion, + )?; + let type_ = i8::from_sql_nullable(&CHAR, get_raw(1)).map_err( + error::conversion, + )?; + let elem_oid = Oid::from_sql_nullable(&OID, get_raw(2)).map_err( + error::conversion, + )?; + let rngsubtype = Option::::from_sql_nullable(&OID, get_raw(3)).map_err( + error::conversion, + )?; + let basetype = Oid::from_sql_nullable(&OID, get_raw(4)).map_err( + error::conversion, + )?; + let schema = String::from_sql_nullable(&NAME, get_raw(5)).map_err( + error::conversion, + )?; + let relid = Oid::from_sql_nullable(&OID, get_raw(6)).map_err( + error::conversion, )?; - let basetype = Oid::from_sql_nullable(&OID, get_raw(4)).map_err(|e| { - Error(Box::new(ErrorKind::Conversion(e))) - })?; - let schema = String::from_sql_nullable(&NAME, get_raw(5)).map_err(|e| { - Error(Box::new(ErrorKind::Conversion(e))) - })?; - let relid = Oid::from_sql_nullable(&OID, get_raw(6)).map_err(|e| { - Error(Box::new(ErrorKind::Conversion(e))) - })?; (name, type_, elem_oid, rngsubtype, basetype, schema, relid) }; @@ -897,9 +881,9 @@ impl InnerConnection { let mut variants = vec![]; for row in rows { - variants.push(String::from_sql_nullable(&NAME, row.get(0)).map_err(|e| { - Error(Box::new(ErrorKind::Conversion(e))) - })?); + variants.push(String::from_sql_nullable(&NAME, row.get(0)).map_err( + error::conversion, + )?); } Ok(variants) @@ -939,12 +923,12 @@ impl InnerConnection { let mut fields = vec![]; for row in rows { let (name, type_) = { - let name = String::from_sql_nullable(&NAME, row.get(0)).map_err(|e| { - Error(Box::new(ErrorKind::Conversion(e))) - })?; - let type_ = Oid::from_sql_nullable(&OID, row.get(1)).map_err(|e| { - Error(Box::new(ErrorKind::Conversion(e))) - })?; + let name = String::from_sql_nullable(&NAME, row.get(0)).map_err( + error::conversion, + )?; + let type_ = Oid::from_sql_nullable(&OID, row.get(1)).map_err( + error::conversion, + )?; (name, type_) }; let type_ = self.get_type(type_)?; @@ -1461,7 +1445,7 @@ impl<'a> GenericConnection for Transaction<'a> { fn err(fields: &mut ErrorFields) -> Error { match DbError::new(fields) { - Ok(err) => Error(Box::new(ErrorKind::Db(err))), + Ok(err) => error::db(err), Err(err) => err.into(), } } diff --git a/postgres/src/priv_io.rs b/postgres/src/priv_io.rs index ea6c3ed9..6f07741e 100644 --- a/postgres/src/priv_io.rs +++ b/postgres/src/priv_io.rs @@ -14,7 +14,7 @@ use postgres_protocol::message::frontend; use postgres_protocol::message::backend; use {Error, Result, TlsMode}; -use error::ErrorKind; +use error; use tls::TlsStream; use params::{ConnectParams, Host}; @@ -273,9 +273,7 @@ pub fn initialize_stream(params: &ConnectParams, tls: TlsMode) -> Result Result return Err(::bad_response().into()), }; - handshaker.tls_handshake(host, socket).map_err(|e| { - Error(Box::new(ErrorKind::Tls(e))) - }) + handshaker.tls_handshake(host, socket).map_err(error::tls) } diff --git a/postgres/src/rows.rs b/postgres/src/rows.rs index 8739b57b..64904427 100644 --- a/postgres/src/rows.rs +++ b/postgres/src/rows.rs @@ -12,10 +12,10 @@ use std::slice; use std::sync::Arc; use {Error, Result, StatementInfo}; +use error; use transaction::Transaction; use types::{FromSql, WrongType}; use stmt::{Statement, Column}; -use error::ErrorKind; enum MaybeOwned<'a, T: 'a> { Borrowed(&'a T), @@ -229,12 +229,10 @@ impl<'a> Row<'a> { let ty = self.stmt_info.columns[idx].type_(); if !::accepts(ty) { - return Some(Err(Error(Box::new( - ErrorKind::Conversion(Box::new(WrongType::new(ty.clone()))), - )))); + return Some(Err(error::conversion(Box::new(WrongType::new(ty.clone()))))); } let value = FromSql::from_sql_nullable(ty, self.data.get(idx)); - Some(value.map_err(|e| Error(Box::new(ErrorKind::Conversion(e))))) + Some(value.map_err(error::conversion)) } /// Retrieves the specified field as a raw buffer of Postgres data. diff --git a/tokio-postgres/src/error.rs b/tokio-postgres/src/error.rs deleted file mode 100644 index c0adbcdc..00000000 --- a/tokio-postgres/src/error.rs +++ /dev/null @@ -1,59 +0,0 @@ -//! Error types. - -use std::error; -use std::io; -use std::fmt; - -use Connection; - -#[doc(inline)] -// FIXME -pub use postgres_shared::error::*; - -/// A runtime error. -#[derive(Debug)] -pub enum Error { - /// An error communicating with the database. - /// - /// IO errors are fatal - the connection is not returned. - Io(io::Error), - /// An error reported by the database. - Db(Box, C), - /// An error converting between Rust and Postgres types. - Conversion(Box, C), -} - -impl fmt::Display for Error { - fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result { - fmt.write_str(error::Error::description(self))?; - match *self { - Error::Db(ref err, _) => write!(fmt, ": {}", err), - Error::Io(ref err) => write!(fmt, ": {}", err), - Error::Conversion(ref err, _) => write!(fmt, ": {}", err), - } - } -} - -impl error::Error for Error { - fn description(&self) -> &str { - match *self { - Error::Db(_, _) => "Error reported by Postgres", - Error::Io(_) => "Error communicating with the server", - Error::Conversion(_, _) => "Error converting between Postgres and Rust types", - } - } - - fn cause(&self) -> Option<&error::Error> { - match *self { - Error::Db(ref err, _) => Some(&**err), - Error::Io(ref err) => Some(err), - Error::Conversion(ref err, _) => Some(&**err), - } - } -} - -impl From for Error { - fn from(err: io::Error) -> Error { - Error::Io(err) - } -} diff --git a/tokio-postgres/src/lib.rs b/tokio-postgres/src/lib.rs index d6ac1ad5..fed8f97e 100644 --- a/tokio-postgres/src/lib.rs +++ b/tokio-postgres/src/lib.rs @@ -83,14 +83,16 @@ use std::io; use std::sync::Arc; use std::sync::atomic::{AtomicUsize, ATOMIC_USIZE_INIT, Ordering}; use std::sync::mpsc::{self, Sender, Receiver}; -use tokio_io::IoFuture; use tokio_core::reactor::Handle; #[doc(inline)] -pub use postgres_shared::{params, CancelData, Notification}; +pub use postgres_shared::{error, params, CancelData, Notification}; +#[doc(inline)] +pub use error::Error; -use error::{ConnectError, Error, DbError, UNDEFINED_TABLE, UNDEFINED_COLUMN}; +use error::{DbError, UNDEFINED_TABLE, UNDEFINED_COLUMN}; use params::{ConnectParams, IntoConnectParams}; +use sink::SinkExt; use stmt::{Statement, Column}; use stream::PostgresStream; use tls::Handshake; @@ -98,9 +100,9 @@ use transaction::Transaction; use types::{Oid, Type, ToSql, IsNull, FromSql, Kind, Field, NAME, CHAR, OID}; use rows::Row; -pub mod error; pub mod rows; pub mod stmt; +mod sink; mod stream; pub mod tls; pub mod transaction; @@ -142,7 +144,7 @@ pub fn cancel_query( tls_mode: TlsMode, cancel_data: CancelData, handle: &Handle, -) -> BoxFuture<(), ConnectError> +) -> BoxFuture<(), Error> where T: IntoConnectParams, { @@ -155,14 +157,14 @@ where handle, )) } - Err(e) => Either::B(Err(ConnectError::ConnectParams(e)).into_future()), + Err(e) => Either::B(Err(error::connect(e)).into_future()), }; params .and_then(move |c| { let mut buf = vec![]; frontend::cancel_request(cancel_data.process_id, cancel_data.secret_key, &mut buf); - c.send(buf).map_err(ConnectError::Io) + c.send(buf).map_err(error::io) }) .map(|_| ()) .boxed() @@ -179,22 +181,30 @@ struct InnerConnection { has_typeinfo_query: bool, has_typeinfo_enum_query: bool, has_typeinfo_composite_query: bool, + desynchronized: bool, } impl InnerConnection { - fn read(self) -> IoFuture<(backend::Message, InnerConnection)> { + fn read(self) -> BoxFuture<(backend::Message, InnerConnection), (io::Error, InnerConnection)> { + if self.desynchronized { + let e = io::Error::new( + io::ErrorKind::Other, + "connection desynchronized due to earlier IO error", + ); + return Err((e, self)).into_future().boxed(); + } + self.into_future() - .map_err(|e| e.0) .and_then(|(m, mut s)| match m { Some(backend::Message::NotificationResponse(body)) => { let process_id = body.process_id(); let channel = match body.channel() { Ok(channel) => channel.to_owned(), - Err(e) => return Either::A(Err(e).into_future()), + Err(e) => return Either::A(Err((e, s)).into_future()), }; let message = match body.message() { Ok(channel) => channel.to_owned(), - Err(e) => return Either::A(Err(e).into_future()), + Err(e) => return Either::A(Err((e, s)).into_future()), }; let notification = Notification { process_id: process_id, @@ -207,9 +217,13 @@ impl InnerConnection { Some(m) => Either::A(Ok((m, s)).into_future()), None => { let err = io::Error::new(io::ErrorKind::UnexpectedEof, "unexpected EOF"); - Either::A(Err(err).into_future()) + Either::A(Err((err, s)).into_future()) } }) + .map_err(|(e, mut s)| { + s.desynchronized = true; + (e, s) + }) .boxed() } } @@ -276,11 +290,7 @@ impl Connection { /// path contains non-UTF 8 characters, a `ConnectParams` struct should be /// created manually and passed in. Note that Postgres does not support TLS /// over Unix sockets. - pub fn connect( - params: T, - tls_mode: TlsMode, - handle: &Handle, - ) -> BoxFuture + pub fn connect(params: T, tls_mode: TlsMode, handle: &Handle) -> BoxFuture where T: IntoConnectParams, { @@ -291,7 +301,7 @@ impl Connection { .map(|s| (s, params)), ) } - Err(e) => Either::B(Err(ConnectError::ConnectParams(e)).into_future()), + Err(e) => Either::B(Err(error::connect(e)).into_future()), }; fut.map(|(s, params)| { @@ -311,6 +321,7 @@ impl Connection { has_typeinfo_query: false, has_typeinfo_enum_query: false, has_typeinfo_composite_query: false, + desynchronized: false, }), params, ) @@ -320,10 +331,7 @@ impl Connection { .boxed() } - fn startup( - self, - params: ConnectParams, - ) -> BoxFuture<(Connection, ConnectParams), ConnectError> { + fn startup(self, params: ConnectParams) -> BoxFuture<(Connection, ConnectParams), Error> { let mut buf = vec![]; let result = { let options = [("client_encoding", "UTF8"), ("timezone", "GMT")]; @@ -338,15 +346,15 @@ impl Connection { result .into_future() .and_then(move |()| self.0.send(buf)) - .map_err(ConnectError::Io) + .map_err(error::io) .map(move |s| (Connection(s), params)) .boxed() } - fn handle_auth(self, params: ConnectParams) -> BoxFuture { + fn handle_auth(self, params: ConnectParams) -> BoxFuture { self.0 .read() - .map_err(ConnectError::Io) + .map_err(|(e, _)| error::io(e)) .and_then(move |(m, s)| { let response = match m { backend::Message::AuthenticationOk => Ok(None), @@ -359,10 +367,8 @@ impl Connection { .map_err(Into::into) } None => { - Err(ConnectError::ConnectParams( - "password was required but not \ - provided" - .into(), + Err(error::connect( + "password was required but not provided".into(), )) } } @@ -383,15 +389,13 @@ impl Connection { .map_err(Into::into) } None => { - Err(ConnectError::ConnectParams( - "password was required but not \ - provided" - .into(), + Err(error::connect( + "password was required but not provided".into(), )) } } } - backend::Message::ErrorResponse(body) => Err(connect_err(&mut body.fields())), + backend::Message::ErrorResponse(body) => Err(err(&mut body.fields())), _ => Err(bad_message()), }; @@ -404,23 +408,23 @@ impl Connection { .boxed() } - fn handle_auth_response(self, message: Vec) -> BoxFuture { + fn handle_auth_response(self, message: Vec) -> BoxFuture { self.0 .send(message) - .and_then(|s| s.read()) - .map_err(ConnectError::Io) + .and_then(|s| s.read().map_err(|(e, _)| e)) + .map_err(error::io) .and_then(|(m, s)| match m { backend::Message::AuthenticationOk => Ok(Connection(s)), - backend::Message::ErrorResponse(body) => Err(connect_err(&mut body.fields())), + backend::Message::ErrorResponse(body) => Err(err(&mut body.fields())), _ => Err(bad_message()), }) .boxed() } - fn finish_startup(self) -> BoxFuture { + fn finish_startup(self) -> BoxFuture { self.0 .read() - .map_err(ConnectError::Io) + .map_err(|(e, _)| error::io(e)) .and_then(|(m, mut s)| match m { backend::Message::BackendKeyData(body) => { s.cancel_data.process_id = body.process_id(); @@ -429,19 +433,25 @@ impl Connection { } backend::Message::ReadyForQuery(_) => Either::B(Ok(Connection(s)).into_future()), backend::Message::ErrorResponse(body) => { - Either::B(Err(connect_err(&mut body.fields())).into_future()) + Either::B(Err(err(&mut body.fields())).into_future()) } _ => Either::B(Err(bad_message()).into_future()), }) .boxed() } - fn simple_query(self, query: &str) -> BoxFuture<(Vec, Connection), Error> { + fn simple_query( + self, + query: &str, + ) -> BoxFuture<(Vec, Connection), (Error, Connection)> { let mut buf = vec![]; - frontend::query(query, &mut buf) - .into_future() - .and_then(move |()| self.0.send(buf)) - .map_err(Error::Io) + if let Err(e) = frontend::query(query, &mut buf) { + return Err((error::io(e), self)).into_future().boxed(); + } + + self.0 + .send2(buf) + .map_err(|(e, s)| (error::io(e), Connection(s))) .and_then(|s| Connection(s).simple_read_rows(vec![])) .boxed() } @@ -450,10 +460,10 @@ impl Connection { fn simple_read_rows( self, mut rows: Vec, - ) -> BoxFuture<(Vec, Connection), Error> { + ) -> BoxFuture<(Vec, Connection), (Error, Connection)> { self.0 .read() - .map_err(Error::Io) + .map_err(|(e, s)| (error::io(e), Connection(s))) .and_then(|(m, s)| match m { backend::Message::ReadyForQuery(_) => { Ok((rows, Connection(s))).into_future().boxed() @@ -464,34 +474,34 @@ impl Connection { rows.push(row); Connection(s).simple_read_rows(rows) } - Err(e) => Err(Error::Io(e)).into_future().boxed(), + Err(e) => Err((error::io(e), Connection(s))).into_future().boxed(), } } backend::Message::EmptyQueryResponse | backend::Message::CommandComplete(_) | backend::Message::RowDescription(_) => Connection(s).simple_read_rows(rows), backend::Message::ErrorResponse(body) => Connection(s).ready_err(body), - _ => Err(bad_message()).into_future().boxed(), + _ => Err((bad_message(), Connection(s))).into_future().boxed(), }) .boxed() } - fn ready(self, t: T) -> BoxFuture<(T, Connection), Error> + fn ready(self, t: T) -> BoxFuture<(T, Connection), (Error, Connection)> where T: 'static + Send, { self.0 .read() - .map_err(Error::Io) + .map_err(|(e, s)| (error::io(e), Connection(s))) .and_then(|(m, s)| match m { - backend::Message::ReadyForQuery(_) => Ok((t, Connection(s))), - _ => Err(bad_message()), + backend::Message::ReadyForQuery(_) => Ok(s), + _ => Err((bad_message(), Connection(s))), }) - .and_then(|(t, s)| s.close_gc().map(|s| (t, s))) + .and_then(|s| Connection(s).close_gc().map(|s| (t, s))) .boxed() } - fn close_gc(self) -> BoxFuture { + fn close_gc(self) -> BoxFuture { let mut messages = vec![]; while let Ok((type_, name)) = self.0.close_receiver.try_recv() { let mut buf = vec![]; @@ -506,36 +516,37 @@ impl Connection { frontend::sync(&mut buf); messages.push(buf); self.0 - .send_all(futures::stream::iter( + .send_all2(futures::stream::iter( messages.into_iter().map(Ok::<_, io::Error>), )) - .map_err(Error::Io) + .map_err(|(e, s, _)| (error::io(e), Connection(s))) .and_then(|s| Connection(s.0).finish_close_gc()) .boxed() } - fn finish_close_gc(self) -> BoxFuture { + fn finish_close_gc(self) -> BoxFuture { self.0 .read() - .map_err(Error::Io) + .map_err(|(e, s)| (error::io(e), Connection(s))) .and_then(|(m, s)| match m { backend::Message::ReadyForQuery(_) => Either::A(Ok(Connection(s)).into_future()), backend::Message::CloseComplete => Either::B(Connection(s).finish_close_gc()), backend::Message::ErrorResponse(body) => Either::B(Connection(s).ready_err(body)), - _ => Either::A(Err(bad_message()).into_future()), + _ => Either::A(Err((bad_message(), Connection(s))).into_future()), }) .boxed() } - fn ready_err(self, body: ErrorResponseBody) -> BoxFuture + fn ready_err(self, body: ErrorResponseBody) -> BoxFuture where T: 'static + Send, { - DbError::new(&mut body.fields()) - .map_err(Error::Io) - .into_future() - .and_then(|e| self.ready(e)) - .and_then(|(e, s)| Err(Error::Db(Box::new(e), s))) + let e = match DbError::new(&mut body.fields()) { + Ok(e) => e, + Err(e) => return Err((error::io(e), self)).into_future().boxed(), + }; + self.ready(e) + .and_then(|(e, s)| Err((error::db(e), s))) .boxed() } @@ -552,7 +563,7 @@ impl Connection { /// user-specified data, as it provides functionality to safely embed that /// data in the statement. Do not form statements via string concatenation /// and feed them into this method. - pub fn batch_execute(self, query: &str) -> BoxFuture { + pub fn batch_execute(self, query: &str) -> BoxFuture { self.simple_query(query).map(|r| r.1).boxed() } @@ -560,23 +571,24 @@ impl Connection { self, name: &str, query: &str, - ) -> BoxFuture<(Vec, Vec, Connection), Error> { + ) -> BoxFuture<(Vec, Vec, Connection), (Error, Connection)> { let mut parse = vec![]; let mut describe = vec![]; let mut sync = vec![]; frontend::sync(&mut sync); - frontend::parse(name, query, None, &mut parse) - .and_then(|()| frontend::describe(b'S', name, &mut describe)) - .into_future() - .and_then(move |()| { - let it = Some(parse).into_iter() - .chain(Some(describe)) - .chain(Some(sync)) - .map(Ok::<_, io::Error>); - self.0.send_all(futures::stream::iter(it)) - }) + if let Err(e) = frontend::parse(name, query, None, &mut parse).and_then(|()| frontend::describe(b'S', name, &mut describe)) { + return Err((error::io(e), self)).into_future().boxed(); + } + + let it = Some(parse) + .into_iter() + .chain(Some(describe)) + .chain(Some(sync)) + .map(Ok::<_, io::Error>); + self.0.send_all2(futures::stream::iter(it)) + .map_err(|(e, s, _)| (e, s)) .and_then(|s| s.0.read()) - .map_err(Error::Io) + .map_err(|(e, s)| (error::io(e), Connection(s))) .boxed() // work around nonlinear trans blowup .and_then(|(m, s)| { match m { @@ -584,33 +596,35 @@ impl Connection { backend::Message::ErrorResponse(body) => { Either::B(Connection(s).ready_err(body)) } - _ => Either::A(Err(bad_message()).into_future()), + _ => Either::A(Err((bad_message(), Connection(s))).into_future()), } }) - .and_then(|s| s.read().map_err(Error::Io)) + .and_then(|s| s.read().map_err(|(e, s)| (error::io(e), Connection(s)))) .and_then(|(m, s)| { match m { backend::Message::ParameterDescription(body) => { - body.parameters().collect::>() - .map(|p| (p, s)) - .map_err(Error::Io) + match body.parameters().collect::>() { + Ok(p) => Ok((p, s)), + Err(e) => Err((error::io(e), Connection(s))), + } } - _ => Err(bad_message()), + _ => Err((bad_message(), Connection(s))), } }) - .and_then(|(p, s)| s.read().map(|(m, s)| (p, m, s)).map_err(Error::Io)) + .and_then(|(p, s)| s.read().map(|(m, s)| (p, m, s)).map_err(|(e, s)| (error::io(e), Connection(s)))) .boxed() // work around nonlinear trans blowup .and_then(|(p, m, s)| { match m { backend::Message::RowDescription(body) => { - body.fields() + match body.fields() .map(|f| (f.name().to_owned(), f.type_oid())) - .collect::>() - .map(|d| (p, d, s)) - .map_err(Error::Io) + .collect::>() { + Ok(d) => Ok((p, d, s)), + Err(e) => Err((error::io(e), Connection(s))), + } } backend::Message::NoData => Ok((p, vec![], s)), - _ => Err(bad_message()), + _ => Err((bad_message(), Connection(s))), } }) .and_then(|(p, r, s)| Connection(s).ready((p, r))) @@ -634,7 +648,7 @@ impl Connection { mut out: Vec, mut get_oid: F, mut build: G, - ) -> BoxFuture<(Vec, Connection), Error> + ) -> BoxFuture<(Vec, Connection), (Error, Connection)> where T: 'static + Send, U: 'static + Send, @@ -656,7 +670,7 @@ impl Connection { } } - fn get_type(self, oid: Oid) -> BoxFuture<(Type, Connection), Error> { + fn get_type(self, oid: Oid) -> BoxFuture<(Type, Connection), (Error, Connection)> { if let Some(type_) = Type::from_oid(oid) { return Ok((type_, self)).into_future().boxed(); }; @@ -674,42 +688,40 @@ impl Connection { .boxed() } - fn get_unknown_type(self, oid: Oid) -> BoxFuture<(Type, Connection), Error> { + fn get_unknown_type(self, oid: Oid) -> BoxFuture<(Type, Connection), (Error, Connection)> { self.setup_typeinfo_query() - .and_then(move |c| { - c.raw_execute(TYPEINFO_QUERY, "", &[OID], &[&oid]) - }) + .and_then(move |c| c.raw_execute(TYPEINFO_QUERY, "", &[OID], &[&oid])) .and_then(|c| c.read_rows().collect()) .and_then(move |(r, c)| { let get = |idx| r.get(0).and_then(|r| r.get(idx)); let name = match String::from_sql_nullable(&NAME, get(0)) { Ok(v) => v, - Err(e) => return Either::A(Err(Error::Conversion(e, c)).into_future()), + Err(e) => return Either::A(Err((error::conversion(e), c)).into_future()), }; let type_ = match i8::from_sql_nullable(&CHAR, get(1)) { Ok(v) => v, - Err(e) => return Either::A(Err(Error::Conversion(e, c)).into_future()), + Err(e) => return Either::A(Err((error::conversion(e), c)).into_future()), }; let elem_oid = match Oid::from_sql_nullable(&OID, get(2)) { Ok(v) => v, - Err(e) => return Either::A(Err(Error::Conversion(e, c)).into_future()), + Err(e) => return Either::A(Err((error::conversion(e), c)).into_future()), }; let rngsubtype = match Option::::from_sql_nullable(&OID, get(3)) { Ok(v) => v, - Err(e) => return Either::A(Err(Error::Conversion(e, c)).into_future()), + Err(e) => return Either::A(Err((error::conversion(e), c)).into_future()), }; let basetype = match Oid::from_sql_nullable(&OID, get(4)) { Ok(v) => v, - Err(e) => return Either::A(Err(Error::Conversion(e, c)).into_future()), + Err(e) => return Either::A(Err((error::conversion(e), c)).into_future()), }; let schema = match String::from_sql_nullable(&NAME, get(5)) { Ok(v) => v, - Err(e) => return Either::A(Err(Error::Conversion(e, c)).into_future()), + Err(e) => return Either::A(Err((error::conversion(e), c)).into_future()), }; let relid = match Oid::from_sql_nullable(&OID, get(6)) { Ok(v) => v, - Err(e) => return Either::A(Err(Error::Conversion(e, c)).into_future()), + Err(e) => return Either::A(Err((error::conversion(e), c)).into_future()), }; let kind = if type_ == b'p' as i8 { @@ -755,7 +767,7 @@ impl Connection { .boxed() } - fn setup_typeinfo_query(self) -> BoxFuture { + fn setup_typeinfo_query(self) -> BoxFuture { if self.0.has_typeinfo_query { return Ok(self).into_future().boxed(); } @@ -770,26 +782,21 @@ impl Connection { INNER JOIN pg_catalog.pg_namespace n ON \ t.typnamespace = n.oid \ WHERE t.oid = $1", - ).or_else(|e| { - match e { - // Range types weren't added until Postgres 9.2, so pg_range may not exist - Error::Db(e, c) => { - if e.code != UNDEFINED_TABLE { - return Either::B(Err(Error::Db(e, c)).into_future()); - } - - Either::A(c.raw_prepare( - TYPEINFO_QUERY, - "SELECT t.typname, t.typtype, t.typelem, \ - NULL::OID, t.typbasetype, n.nspname, \ - t.typrelid \ - FROM pg_catalog.pg_type t \ - INNER JOIN pg_catalog.pg_namespace n \ - ON t.typnamespace = n.oid \ - WHERE t.oid = $1", - )) - } - e => Either::B(Err(e).into_future()), + ).or_else(|(e, c)| { + // Range types weren't added until Postgres 9.2, so pg_range may not exist + if e.code() == Some(&UNDEFINED_TABLE) { + Either::A(c.raw_prepare( + TYPEINFO_QUERY, + "SELECT t.typname, t.typtype, t.typelem, \ + NULL::OID, t.typbasetype, n.nspname, \ + t.typrelid \ + FROM pg_catalog.pg_type t \ + INNER JOIN pg_catalog.pg_namespace n \ + ON t.typnamespace = n.oid \ + WHERE t.oid = $1", + )) + } else { + Either::B(Err((e, c)).into_future()) } }) .map(|(_, _, mut c)| { @@ -799,7 +806,10 @@ impl Connection { .boxed() } - fn get_enum_variants(self, oid: Oid) -> BoxFuture<(Vec, Connection), Error> { + fn get_enum_variants( + self, + oid: Oid, + ) -> BoxFuture<(Vec, Connection), (Error, Connection)> { self.setup_typeinfo_enum_query() .and_then(move |c| { c.raw_execute(TYPEINFO_ENUM_QUERY, "", &[OID], &[&oid]) @@ -810,7 +820,7 @@ impl Connection { for row in r { let variant = match String::from_sql_nullable(&NAME, row.get(0)) { Ok(v) => v, - Err(e) => return Err(Error::Conversion(e, c)), + Err(e) => return Err((error::conversion(e), c)), }; variants.push(variant); } @@ -819,7 +829,7 @@ impl Connection { .boxed() } - fn setup_typeinfo_enum_query(self) -> BoxFuture { + fn setup_typeinfo_enum_query(self) -> BoxFuture { if self.0.has_typeinfo_enum_query { return Ok(self).into_future().boxed(); } @@ -830,19 +840,14 @@ impl Connection { FROM pg_catalog.pg_enum \ WHERE enumtypid = $1 \ ORDER BY enumsortorder", - ).or_else(|e| match e { - Error::Db(e, c) => { - if e.code != UNDEFINED_COLUMN { - return Either::B(Err(Error::Db(e, c)).into_future()); - } - - Either::A(c.raw_prepare( - TYPEINFO_ENUM_QUERY, - "SELECT enumlabel FROM pg_catalog.pg_enum WHERE \ + ).or_else(|(e, c)| if e.code() == Some(&UNDEFINED_COLUMN) { + Either::A(c.raw_prepare( + TYPEINFO_ENUM_QUERY, + "SELECT enumlabel FROM pg_catalog.pg_enum WHERE \ enumtypid = $1 ORDER BY oid", - )) - } - e => Either::B(Err(e).into_future()), + )) + } else { + Either::B(Err((e, c)).into_future()) }) .map(|(_, _, mut c)| { c.0.has_typeinfo_enum_query = true; @@ -851,7 +856,7 @@ impl Connection { .boxed() } - fn get_composite_fields(self, oid: Oid) -> BoxFuture<(Vec, Connection), Error> { + fn get_composite_fields(self, oid: Oid) -> BoxFuture<(Vec, Connection), (Error, Connection)> { self.setup_typeinfo_composite_query() .and_then(move |c| { c.raw_execute(TYPEINFO_COMPOSITE_QUERY, "", &[OID], &[&oid]) @@ -863,11 +868,11 @@ impl Connection { |(mut fields, c), row| { let name = match String::from_sql_nullable(&NAME, row.get(0)) { Ok(name) => name, - Err(e) => return Either::A(Err(Error::Conversion(e, c)).into_future()), + Err(e) => return Either::A(Err((error::conversion(e), c)).into_future()), }; let oid = match Oid::from_sql_nullable(&OID, row.get(1)) { Ok(oid) => oid, - Err(e) => return Either::A(Err(Error::Conversion(e, c)).into_future()), + Err(e) => return Either::A(Err((error::conversion(e), c)).into_future()), }; Either::B(c.get_type(oid).map(move |(ty, c)| { fields.push(Field::new(name, ty)); @@ -879,7 +884,7 @@ impl Connection { .boxed() } - fn setup_typeinfo_composite_query(self) -> BoxFuture { + fn setup_typeinfo_composite_query(self) -> BoxFuture { if self.0.has_typeinfo_composite_query { return Ok(self).into_future().boxed(); } @@ -905,7 +910,7 @@ impl Connection { portal: &str, param_types: &[Type], params: &[&ToSql], - ) -> BoxFuture { + ) -> BoxFuture { assert!( param_types.len() == params.len(), "expected {} parameters but got {}", @@ -932,14 +937,15 @@ impl Connection { ); let r = match r { Ok(()) => Ok(self), - Err(frontend::BindError::Conversion(e)) => Err(Error::Conversion(e, self)), - Err(frontend::BindError::Serialization(e)) => Err(Error::Io(e)), + Err(frontend::BindError::Conversion(e)) => Err((error::conversion(e), self)), + Err(frontend::BindError::Serialization(e)) => Err((error::io(e), self)), }; r.and_then(|s| { - frontend::execute(portal, 0, &mut execute) - .map(|()| s) - .map_err(Error::Io) + match frontend::execute(portal, 0, &mut execute) { + Ok(()) => Ok(s), + Err(e) => Err((error::io(e), s)), + } }).into_future() .and_then(|s| { let it = Some(bind) @@ -947,41 +953,42 @@ impl Connection { .chain(Some(execute)) .chain(Some(sync)) .map(Ok::<_, io::Error>); - s.0.send_all(futures::stream::iter(it)).map_err(Error::Io) + s.0.send_all2(futures::stream::iter(it)).map_err(|(e, s, _)| (error::io(e), Connection(s))) }) - .and_then(|s| s.0.read().map_err(Error::Io)) + .and_then(|s| s.0.read().map_err(|(e, s)| (error::io(e), Connection(s)))) .and_then(|(m, s)| match m { backend::Message::BindComplete => Either::A(Ok(Connection(s)).into_future()), backend::Message::ErrorResponse(body) => Either::B(Connection(s).ready_err(body)), - _ => Either::A(Err(bad_message()).into_future()), + _ => Either::A(Err((bad_message(), Connection(s))).into_future()), }) .boxed() } - fn finish_execute(self) -> BoxFuture<(u64, Connection), Error> { + fn finish_execute(self) -> BoxFuture<(u64, Connection), (Error, Connection)> { self.0 .read() - .map_err(Error::Io) + .map_err(|(e, s)| (error::io(e), Connection(s))) .and_then(|(m, s)| match m { backend::Message::DataRow(_) => Connection(s).finish_execute().boxed(), backend::Message::CommandComplete(body) => { - body.tag() + let r = body.tag() .map(|tag| { tag.split_whitespace().last().unwrap().parse().unwrap_or(0) - }) - .map_err(Error::Io) - .into_future() - .and_then(|n| Connection(s).ready(n)) - .boxed() + }); + + match r { + Ok(n) => Connection(s).ready(n).boxed(), + Err(e) => Err((error::io(e), Connection(s))).into_future().boxed(), + } } backend::Message::EmptyQueryResponse => Connection(s).ready(0).boxed(), backend::Message::ErrorResponse(body) => Connection(s).ready_err(body).boxed(), - _ => Err(bad_message()).into_future().boxed(), + _ => Err((bad_message(), Connection(s))).into_future().boxed(), }) .boxed() } - fn read_rows(self) -> BoxStateStream { + fn read_rows(self) -> BoxStateStream { futures_state_stream::unfold(self, |c| { c.read_row().and_then(|(r, c)| match r { Some(data) => { @@ -993,32 +1000,31 @@ impl Connection { }).boxed() } - fn read_row(self) -> BoxFuture<(Option, Connection), Error> { + fn read_row(self) -> BoxFuture<(Option, Connection), (Error, Connection)> { self.0 .read() - .map_err(Error::Io) + .map_err(|(e, s)| (error::io(e), Connection(s))) .and_then(|(m, s)| { let c = Connection(s); match m { backend::Message::DataRow(body) => { - Either::A( - RowData::new(body) - .map(|r| (Some(r), c)) - .map_err(Error::Io) - .into_future(), - ) + let r = match RowData::new(body) { + Ok(r) => Ok((Some(r), c)), + Err(e) => Err((error::io(e), c)) + }; + Either::A(r.into_future()) } backend::Message::EmptyQueryResponse | backend::Message::CommandComplete(_) => Either::A(Ok((None, c)).into_future()), backend::Message::ErrorResponse(body) => Either::B(c.ready_err(body)), - _ => Either::A(Err(bad_message()).into_future()), + _ => Either::A(Err((bad_message(), c)).into_future()), } }) .boxed() } /// Creates a new prepared statement. - pub fn prepare(self, query: &str) -> BoxFuture<(Statement, Connection), Error> { + pub fn prepare(self, query: &str) -> BoxFuture<(Statement, Connection), (Error, Connection)> { let id = NEXT_STMT_ID.fetch_add(1, Ordering::SeqCst); let name = format!("s{}", id); self.raw_prepare(&name, query) @@ -1040,7 +1046,7 @@ impl Connection { self, statement: &Statement, params: &[&ToSql], - ) -> BoxFuture<(u64, Connection), Error> { + ) -> BoxFuture<(u64, Connection), (Error, Connection)> { self.raw_execute(statement.name(), "", statement.parameters(), params) .and_then(|conn| conn.finish_execute()) .boxed() @@ -1056,7 +1062,7 @@ impl Connection { self, statement: &Statement, params: &[&ToSql], - ) -> BoxStateStream { + ) -> BoxStateStream { let columns = statement.columns_arc().clone(); self.raw_execute(statement.name(), "", statement.parameters(), params) .map(|c| c.read_rows().map(move |r| Row::new(columns.clone(), r))) @@ -1065,7 +1071,7 @@ impl Connection { } /// Starts a new transaction. - pub fn transaction(self) -> BoxFuture { + pub fn transaction(self) -> BoxFuture { self.simple_query("BEGIN") .map(|(_, c)| Transaction::new(c)) .boxed() @@ -1126,10 +1132,10 @@ impl Stream for Notifications { } } -fn connect_err(fields: &mut ErrorFields) -> ConnectError { +fn err(fields: &mut ErrorFields) -> Error { match DbError::new(fields) { - Ok(err) => ConnectError::Db(Box::new(err)), - Err(err) => ConnectError::Io(err), + Ok(err) => error::db(err), + Err(err) => error::io(err), } } diff --git a/tokio-postgres/src/sink.rs b/tokio-postgres/src/sink.rs new file mode 100644 index 00000000..518e0d9b --- /dev/null +++ b/tokio-postgres/src/sink.rs @@ -0,0 +1,164 @@ +use futures::{Sink, Future, Poll, AsyncSink, Async, Stream}; +use futures::stream::Fuse; + +pub trait SinkExt: Sink { + fn send2(self, item: Self::SinkItem) -> Send2 + where + Self: Sized; + + // unlike send_all, this doesn't close the stream. + fn send_all2(self, stream: S) -> SendAll2 + where + S: Stream, + Self::SinkError: From, + Self: Sized; +} + +impl SinkExt for T +where + T: Sink, +{ + fn send2(self, item: Self::SinkItem) -> Send2 + where + Self: Sized, + { + Send2 { + sink: Some(self), + item: Some(item), + } + } + + fn send_all2(self, stream: S) -> SendAll2 + where + S: Stream, + Self::SinkError: From, + Self: Sized, + { + SendAll2 { + sink: Some(self), + stream: Some(stream.fuse()), + buffered: None, + } + } +} + +pub struct Send2 +where + T: Sink, +{ + sink: Option, + item: Option, +} + +impl Future for Send2 +where + T: Sink, +{ + type Item = T; + type Error = (T::SinkError, T); + + fn poll(&mut self) -> Poll { + let mut sink = self.sink.take().expect("poll called after completion"); + + if let Some(item) = self.item.take() { + match sink.start_send(item) { + Ok(AsyncSink::NotReady(item)) => { + self.sink = Some(sink); + self.item = Some(item); + return Ok(Async::NotReady); + } + Ok(AsyncSink::Ready) => {} + Err(e) => return Err((e, sink)), + } + } + + match sink.poll_complete() { + Ok(Async::Ready(())) => {} + Ok(Async::NotReady) => { + self.sink = Some(sink); + return Ok(Async::NotReady); + } + Err(e) => return Err((e, sink)), + } + + Ok(Async::Ready(sink)) + } +} + +pub struct SendAll2 +where + U: Stream, +{ + sink: Option, + stream: Option>, + buffered: Option, +} + +impl Future for SendAll2 +where + T: Sink, + U: Stream, + T::SinkError: From, +{ + type Item = (T, U); + type Error = (T::SinkError, T, U); + + fn poll(&mut self) -> Poll<(T, U), (T::SinkError, T, U)> { + let mut stream = self.stream.take().expect("poll called after completion"); + let mut sink = self.sink.take().expect("poll called after completion"); + + if let Some(item) = self.buffered.take() { + match sink.start_send(item) { + Ok(AsyncSink::Ready) => {} + Ok(AsyncSink::NotReady(item)) => { + self.sink = Some(sink); + self.buffered = Some(item); + self.stream = Some(stream); + return Ok(Async::NotReady); + } + Err(e) => return Err((e, sink, stream.into_inner())), + } + } + + loop { + match stream.poll() { + Ok(Async::Ready(Some(item))) => { + match sink.start_send(item) { + Ok(AsyncSink::Ready) => {} + Ok(AsyncSink::NotReady(item)) => { + self.sink = Some(sink); + self.buffered = Some(item); + self.stream = Some(stream); + return Ok(Async::NotReady); + } + Err(e) => return Err((e, sink, stream.into_inner())), + } + } + Ok(Async::Ready(None)) => { + match sink.poll_complete() { + Ok(Async::Ready(())) => return Ok(Async::Ready((sink, stream.into_inner()))), + Ok(Async::NotReady) => { + self.sink = Some(sink); + self.stream = Some(stream); + return Ok(Async::NotReady); + } + Err(e) => return Err((e, sink, stream.into_inner())), + } + } + Ok(Async::NotReady) => { + match sink.poll_complete() { + Ok(Async::Ready(())) => return Ok(Async::Ready((sink, stream.into_inner()))), + Ok(Async::NotReady) => { + self.sink = Some(sink); + self.stream = Some(stream); + return Ok(Async::NotReady); + } + Err(e) => return Err((e, sink, stream.into_inner())), + } + return Ok(Async::NotReady); + } + Err(e) => return Err((e.into(), sink, stream.into_inner())), + } + } + } +} diff --git a/tokio-postgres/src/stream.rs b/tokio-postgres/src/stream.rs index d226201d..9cc1515a 100644 --- a/tokio-postgres/src/stream.rs +++ b/tokio-postgres/src/stream.rs @@ -14,8 +14,8 @@ use tokio_dns; #[cfg(unix)] use tokio_uds::UnixStream; -use TlsMode; -use error::ConnectError; +use {TlsMode, Error}; +use error; use tls::TlsStream; pub type PostgresStream = Framed, PostgresCodec>; @@ -25,13 +25,13 @@ pub fn connect( port: u16, tls_mode: TlsMode, handle: &Handle, -) -> BoxFuture { +) -> BoxFuture { let inner = match host { Host::Tcp(ref host) => { Either::A( tokio_dns::tcp_connect((&**host, port), handle.remote().clone()) .map(|s| Stream(InnerStream::Tcp(s))) - .map_err(ConnectError::Io), + .map_err(error::io), ) } #[cfg(unix)] @@ -40,7 +40,7 @@ pub fn connect( Either::B( UnixStream::connect(addr, handle) .map(|s| Stream(InnerStream::Unix(s))) - .map_err(ConnectError::Io) + .map_err(error::io) .into_future(), ) } @@ -74,15 +74,15 @@ pub fn connect( .and_then(|s| { let mut buf = vec![]; frontend::ssl_request(&mut buf); - s.send(buf).map_err(ConnectError::Io) + s.send(buf).map_err(error::io) }) - .and_then(|s| s.into_future().map_err(|e| ConnectError::Io(e.0))) + .and_then(|s| s.into_future().map_err(|e| error::io(e.0))) .and_then(move |(m, s)| { let s = s.into_inner(); match (m, required) { (Some(b'N'), true) => { Either::A( - Err(ConnectError::Tls("the server does not support TLS".into())) + Err(error::tls("the server does not support TLS".into())) .into_future(), ) } @@ -92,7 +92,7 @@ pub fn connect( } (None, _) => { Either::A( - Err(ConnectError::Io(io::Error::new( + Err(error::io(io::Error::new( io::ErrorKind::UnexpectedEof, "unexpected EOF", ))).into_future(), @@ -103,7 +103,7 @@ pub fn connect( Host::Tcp(ref host) => host, Host::Unix(_) => unreachable!(), }; - Either::B(handshaker.handshake(host, s).map_err(ConnectError::Tls)) + Either::B(handshaker.handshake(host, s).map_err(error::tls)) } } }) diff --git a/tokio-postgres/src/test.rs b/tokio-postgres/src/test.rs index c2763fb4..eceb8b25 100644 --- a/tokio-postgres/src/test.rs +++ b/tokio-postgres/src/test.rs @@ -6,7 +6,7 @@ use std::time::Duration; use tokio_core::reactor::{Core, Interval}; use super::*; -use error::{Error, ConnectError, INVALID_PASSWORD, INVALID_AUTHORIZATION_SPECIFICATION, +use error::{Error, INVALID_PASSWORD, INVALID_AUTHORIZATION_SPECIFICATION, QUERY_CANCELED}; use params::{ConnectParams, Host}; use types::{ToSql, FromSql, Type, IsNull, Kind, BYTEA, TEXT, INT4, NUMERIC}; @@ -33,7 +33,7 @@ fn md5_user_no_pass() { &handle, ); match l.run(done) { - Err(ConnectError::ConnectParams(_)) => {} + Err(ref e) if e.as_connection().is_some() => {} Err(e) => panic!("unexpected error {}", e), Ok(_) => panic!("unexpected success"), } @@ -49,7 +49,7 @@ fn md5_user_wrong_pass() { &handle, ); match l.run(done) { - Err(ConnectError::Db(ref e)) if e.code == INVALID_PASSWORD => {} + Err(ref e) if e.code() == Some(&INVALID_PASSWORD) => {} Err(e) => panic!("unexpected error {}", e), Ok(_) => panic!("unexpected success"), } @@ -77,7 +77,7 @@ fn pass_user_no_pass() { &handle, ); match l.run(done) { - Err(ConnectError::ConnectParams(_)) => {} + Err(ref e) if e.as_connection().is_some() => {} Err(e) => panic!("unexpected error {}", e), Ok(_) => panic!("unexpected success"), } @@ -93,7 +93,7 @@ fn pass_user_wrong_pass() { &handle, ); match l.run(done) { - Err(ConnectError::Db(ref e)) if e.code == INVALID_PASSWORD => {} + Err(ref e) if e.code() == Some(&INVALID_PASSWORD) => {} Err(e) => panic!("unexpected error {}", e), Ok(_) => panic!("unexpected success"), } @@ -129,11 +129,10 @@ fn batch_execute_err() { }) .and_then(|c| c.batch_execute("SELECT * FROM bogo")) .then(|r| match r { - Err(Error::Db(e, s)) => { - assert!(e.code == UNDEFINED_TABLE); + Err((e, s)) => { + assert_eq!(e.code(), Some(&UNDEFINED_TABLE)); s.batch_execute("SELECT * FROM foo") } - Err(e) => panic!("unexpected error: {}", e), Ok(_) => panic!("unexpected success"), }); l.run(done).unwrap(); @@ -268,8 +267,7 @@ fn ssl_user_ssl_required() { ); match l.run(done) { - Err(ConnectError::Db(e)) => assert!(e.code == INVALID_AUTHORIZATION_SPECIFICATION), - Err(e) => panic!("unexpected error {}", e), + Err(ref e) => assert_eq!(e.code(), Some(&INVALID_AUTHORIZATION_SPECIFICATION)), Ok(_) => panic!("unexpected success"), } } @@ -456,8 +454,7 @@ fn cancel() { let (select, cancel) = l.run(done).unwrap(); cancel.unwrap(); match select { - Err(Error::Db(e, _)) => assert_eq!(e.code, QUERY_CANCELED), - Err(e) => panic!("unexpected error {}", e), + Err((e, _)) => assert_eq!(e.code(), Some(&QUERY_CANCELED)), Ok(_) => panic!("unexpected success"), } } @@ -477,7 +474,7 @@ fn notifications() { .map(|_| c1) }) }) - .and_then(|c| c.notifications().into_future().map_err(|(e, _)| e)) + .and_then(|c| c.notifications().into_future().map_err(|(e, n)| (e, n.into_inner()))) .map(|(n, _)| { let n = n.unwrap(); assert_eq!(n.channel, "test_notifications"); diff --git a/tokio-postgres/src/transaction.rs b/tokio-postgres/src/transaction.rs index acaa0c3d..f099917b 100644 --- a/tokio-postgres/src/transaction.rs +++ b/tokio-postgres/src/transaction.rs @@ -19,7 +19,7 @@ impl Transaction { } /// Like `Connection::batch_execute`. - pub fn batch_execute(self, query: &str) -> BoxFuture> { + pub fn batch_execute(self, query: &str) -> BoxFuture { self.0 .batch_execute(query) .map(Transaction) @@ -28,7 +28,7 @@ impl Transaction { } /// Like `Connection::prepare`. - pub fn prepare(self, query: &str) -> BoxFuture<(Statement, Transaction), Error> { + pub fn prepare(self, query: &str) -> BoxFuture<(Statement, Transaction), (Error, Transaction)> { self.0 .prepare(query) .map(|(s, c)| (s, Transaction(c))) @@ -41,7 +41,7 @@ impl Transaction { self, statement: &Statement, params: &[&ToSql], - ) -> BoxFuture<(u64, Transaction), Error> { + ) -> BoxFuture<(u64, Transaction), (Error, Transaction)> { self.0 .execute(statement, params) .map(|(n, c)| (n, Transaction(c))) @@ -54,7 +54,7 @@ impl Transaction { self, statement: &Statement, params: &[&ToSql], - ) -> BoxStateStream> { + ) -> BoxStateStream { self.0 .query(statement, params) .map_state(Transaction) @@ -63,24 +63,20 @@ impl Transaction { } /// Commits the transaction. - pub fn commit(self) -> BoxFuture { + pub fn commit(self) -> BoxFuture { self.finish("COMMIT") } /// Rolls back the transaction. - pub fn rollback(self) -> BoxFuture { + pub fn rollback(self) -> BoxFuture { self.finish("ROLLBACK") } - fn finish(self, query: &str) -> BoxFuture { + fn finish(self, query: &str) -> BoxFuture { self.0.simple_query(query).map(|(_, c)| c).boxed() } } -fn transaction_err(e: Error) -> Error { - match e { - Error::Io(e) => Error::Io(e), - Error::Db(e, c) => Error::Db(e, Transaction(c)), - Error::Conversion(e, c) => Error::Conversion(e, Transaction(c)), - } +fn transaction_err((e, c): (Error, Connection)) -> (Error, Transaction) { + (e, Transaction(c)) }