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

Add batch_execute

Browse Source
This commit is contained in:
Steven Fackler 2016-12-20 19:50:44 -08:00
parent 0c56a5ed13
commit 7edf66f7f0
10 changed files with 356 additions and 168 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

106
postgres-shared/src/error/mod.rs
View File

@ -1,5 +1,3 @@
//! Error types.
use fallible_iterator::FallibleIterator;
use postgres_protocol::message::backend::ErrorFields;
use std::error;
@ -141,7 +139,7 @@ pub struct DbError {
impl DbError {
#[doc(hidden)]
pub fn new_raw(fields: &mut ErrorFields) -> io::Result<DbError> {
pub fn new(fields: &mut ErrorFields) -> io::Result<DbError> {
let mut severity = None;
let mut parsed_severity = None;
let mut code = None;
@ -245,22 +243,6 @@ impl DbError {
_p: (),
})
}
#[doc(hidden)]
pub fn new_connect<T>(fields: &mut ErrorFields) -> Result<T, ConnectError> {
match DbError::new_raw(fields) {
Ok(err) => Err(ConnectError::Db(Box::new(err))),
Err(e) => Err(ConnectError::Io(e)),
}
}
#[doc(hidden)]
pub fn new<T>(fields: &mut ErrorFields) -> Result<T, Error> {
match DbError::new_raw(fields) {
Ok(err) => Err(Error::Db(Box::new(err))),
Err(e) => Err(Error::Io(e)),
}
}
}
// manual impl to leave out _p
@ -299,6 +281,20 @@ impl error::Error for DbError {
}
}
/// Represents the position of an error in a query.
#[derive(Clone, PartialEq, Eq, Debug)]
pub enum ErrorPosition {
/// A position in the original query.
Normal(u32),
/// A position in an internally generated query.
Internal {
/// The byte position.
position: u32,
/// A query generated by the Postgres server.
query: String,
},
}
/// Reasons a new Postgres connection could fail.
#[derive(Debug)]
pub enum ConnectError {
@ -355,75 +351,3 @@ impl From<DbError> for ConnectError {
ConnectError::Db(Box::new(err))
}
}
/// Represents the position of an error in a query.
#[derive(Clone, PartialEq, Eq, Debug)]
pub enum ErrorPosition {
/// A position in the original query.
Normal(u32),
/// A position in an internally generated query.
Internal {
/// The byte position.
position: u32,
/// A query generated by the Postgres server.
query: String,
},
}
/// An error encountered when communicating with the Postgres server.
#[derive(Debug)]
pub enum Error {
/// An error reported by the Postgres server.
Db(Box<DbError>),
/// An error communicating with the Postgres server.
Io(io::Error),
/// An error converting between Postgres and Rust types.
Conversion(Box<error::Error + Sync + Send>),
}
impl fmt::Display for Error {
fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
try!(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<DbError> for Error {
fn from(err: DbError) -> Error {
Error::Db(Box::new(err))
}
}
impl From<io::Error> for Error {
fn from(err: io::Error) -> Error {
Error::Io(err)
}
}
impl From<Error> for io::Error {
fn from(err: Error) -> io::Error {
io::Error::new(io::ErrorKind::Other, err)
}
}

46
postgres-shared/src/lib.rs
View File

@ -5,5 +5,51 @@ extern crate fallible_iterator;
extern crate phf;
extern crate postgres_protocol;
use fallible_iterator::{FallibleIterator, FromFallibleIterator};
use std::ops::Range;
pub mod error;
pub mod params;
pub struct RowData {
buf: Vec<u8>,
indices: Vec<Option<Range<usize>>>,
}
impl<'a> FromFallibleIterator<Option<&'a [u8]>> for RowData {
fn from_fallible_iterator<I>(mut it: I) -> Result<RowData, I::Error>
where I: FallibleIterator<Item = Option<&'a [u8]>>
{
let mut row = RowData {
buf: vec![],
indices: Vec::with_capacity(it.size_hint().0),
};
while let Some(cell) = try!(it.next()) {
let index = match cell {
Some(cell) => {
let base = row.buf.len();
row.buf.extend_from_slice(cell);
Some(base..row.buf.len())
}
None => None,
};
row.indices.push(index);
}
Ok(row)
}
}
impl RowData {
pub fn len(&self) -> usize {
self.indices.len()
}
pub fn get(&self, index: usize) -> Option<&[u8]> {
match &self.indices[index] {
&Some(ref range) => Some(&self.buf[range.clone()]),
&None => None,
}
}
}

1
postgres-tokio/Cargo.toml
View File

@ -4,6 +4,7 @@ version = "0.1.0"
authors = ["Steven Fackler <sfackler@gmail.com>"]
[dependencies]
fallible-iterator = "0.1.3"
futures = "0.1.7"
postgres-shared = { path = "../postgres-shared" }
postgres-protocol = "0.2"

50
postgres-tokio/src/error.rs Normal file
View File

@ -0,0 +1,50 @@
use std::error;
use std::io;
use std::fmt;
use Connection;
#[doc(inline)]
pub use postgres_shared::error::*;
#[derive(Debug)]
pub enum Error {
Io(io::Error),
Db(Box<DbError>, Connection),
Conversion(Box<error::Error + Sync + Send>, Connection),
}
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<io::Error> for Error {
fn from(err: io::Error) -> Error {
Error::Io(err)
}
}

134
postgres-tokio/src/lib.rs
View File

@ -1,29 +1,31 @@
extern crate fallible_iterator;
extern crate futures;
extern crate postgres_shared;
extern crate postgres_protocol;
extern crate tokio_core;
extern crate tokio_dns;
extern crate tokio_uds;
#[macro_use]
extern crate futures;
use fallible_iterator::FallibleIterator;
use futures::{Future, IntoFuture, BoxFuture, Stream, Sink, Poll, StartSend};
use futures::future::Either;
use postgres_protocol::authentication;
use postgres_protocol::message::{backend, frontend};
use postgres_protocol::message::backend::ErrorFields;
use postgres_shared::RowData;
use std::collections::HashMap;
use std::fmt;
use std::io;
use tokio_core::reactor::Handle;
#[doc(inline)]
pub use postgres_shared::error;
#[doc(inline)]
pub use postgres_shared::params;
use error::{ConnectError, DbError};
use error::{ConnectError, Error, DbError};
use params::{ConnectParams, IntoConnectParams};
use stream::PostgresStream;
pub mod error;
mod stream;
#[cfg(test)]
@ -99,6 +101,14 @@ impl Sink for InnerConnection {
pub struct Connection(InnerConnection);
// FIXME fill out
impl fmt::Debug for Connection {
fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
fmt.debug_struct("Connection")
.finish()
}
}
impl Connection {
pub fn connect<T>(params: T, handle: &Handle) -> BoxFuture<Connection, ConnectError>
where T: IntoConnectParams
@ -184,9 +194,7 @@ impl Connection {
}
}
}
backend::Message::ErrorResponse(body) => {
DbError::new_connect(&mut body.fields())
}
backend::Message::ErrorResponse(body) => Err(connect_err(&mut body.fields())),
_ => Err(bad_message()),
};
@ -209,9 +217,7 @@ impl Connection {
.and_then(|(m, s)| {
match m {
backend::Message::AuthenticationOk => Ok(Connection(s)),
backend::Message::ErrorResponse(body) => {
DbError::new_connect(&mut body.fields())
}
backend::Message::ErrorResponse(body) => Err(connect_err(&mut body.fields())),
_ => Err(bad_message()),
}
})
@ -230,7 +236,7 @@ impl Connection {
}
backend::Message::ReadyForQuery(_) => Either::B(Ok(Connection(s)).into_future()),
backend::Message::ErrorResponse(body) => {
Either::B(DbError::new_connect(&mut body.fields()).into_future())
Either::B(Err(connect_err(&mut body.fields())).into_future())
}
_ => Either::B(Err(bad_message()).into_future()),
}
@ -238,11 +244,113 @@ impl Connection {
.boxed()
}
fn simple_query(self, query: &str) -> BoxFuture<(Vec<RowData>, Connection), Error> {
let mut buf = vec![];
frontend::query(query, &mut buf)
.map(|()| buf)
.into_future()
.and_then(move |buf| self.0.send(buf))
.and_then(|s| s.flush())
.map_err(Error::Io)
.and_then(|s| Connection(s).simple_read_rows(vec![]))
.boxed()
}
// This has its own read_rows since it will need to handle multiple query completions
fn simple_read_rows(self, mut rows: Vec<RowData>) -> BoxFuture<(Vec<RowData>, Connection), Error> {
self.0.read()
.map_err(|e| Error::Io(e.0))
.and_then(|(m, s)| {
match m {
backend::Message::ReadyForQuery(_) => {
Either::A(Ok((rows, Connection(s))).into_future())
}
backend::Message::DataRow(body) => {
match body.values().collect() {
Ok(row) => {
rows.push(row);
Either::B(Connection(s).simple_read_rows(rows))
}
Err(e) => Either::A(Err(Error::Io(e)).into_future()),
}
}
backend::Message::EmptyQueryResponse |
backend::Message::CommandComplete(_) => {
Either::B(Connection(s).simple_read_rows(rows))
}
backend::Message::ErrorResponse(body) => {
Either::A(Err(err(&mut body.fields(), Connection(s))).into_future())
}
_ => Either::A(Err(bad_message()).into_future()),
}
})
.boxed()
}
fn read_rows(self, mut rows: Vec<RowData>) -> BoxFuture<(Vec<RowData>, Connection), Error> {
self.0.read()
.map_err(|e| Error::Io(e.0))
.and_then(|(m, s)| {
match m {
backend::Message::EmptyQueryResponse |
backend::Message::CommandComplete(_) => {
Either::B(Connection(s).ready(rows))
},
backend::Message::DataRow(body) => {
match body.values().collect() {
Ok(row) => {
rows.push(row);
Either::B(Connection(s).read_rows(rows))
}
Err(e) => Either::A(Err(Error::Io(e)).into_future()),
}
}
backend::Message::ErrorResponse(body) => {
Either::A(Err(err(&mut body.fields(), Connection(s))).into_future())
}
_ => Either::A(Err(bad_message()).into_future()),
}
})
.boxed()
}
fn ready<T>(self, t: T) -> BoxFuture<(T, Connection), Error>
where T: 'static + Send
{
self.0.read()
.map_err(|e| Error::Io(e.0))
.and_then(|(m, s)| {
match m {
backend::Message::ReadyForQuery(_) => Ok((t, Connection(s))),
_ => Err(bad_message())
}
})
.boxed()
}
pub fn batch_execute(self, query: &str) -> BoxFuture<Connection, Error> {
self.simple_query(query).map(|r| r.1).boxed()
}
pub fn cancel_data(&self) -> CancelData {
self.0.cancel_data
}
}
fn connect_err(fields: &mut ErrorFields) -> ConnectError {
match DbError::new(fields) {
Ok(err) => ConnectError::Db(Box::new(err)),
Err(err) => ConnectError::Io(err),
}
}
fn err(fields: &mut ErrorFields, conn: Connection) -> Error {
match DbError::new(fields) {
Ok(err) => Error::Db(Box::new(err), conn),
Err(err) => Error::Io(err),
}
}
fn bad_message<T>() -> T
where T: From<io::Error>
{

25
postgres-tokio/src/test.rs
View File

@ -1,7 +1,8 @@
use futures::Future;
use tokio_core::reactor::Core;
use super::*;
use error::{ConnectError, SqlState};
use error::{Error, ConnectError, SqlState};
#[test]
fn basic() {
@ -75,3 +76,25 @@ fn pass_user_wrong_pass() {
Ok(_) => panic!("unexpected success"),
}
}
#[test]
fn batch_execute_ok() {
let mut l = Core::new().unwrap();
let done = Connection::connect("postgres://postgres@localhost", &l.handle())
.then(|c| c.unwrap().batch_execute("CREATE TEMPORARY TABLE foo (id SERIAL);"));
l.run(done).unwrap();
}
#[test]
fn batch_execute_err() {
let mut l = Core::new().unwrap();
let done = Connection::connect("postgres://postgres@localhost", &l.handle())
.then(|r| r.unwrap().batch_execute("CREATE TEMPORARY TABLE foo (id SERIAL); \
INSERT INTO foo DEFAULT VALUES;"))
.and_then(|c| c.batch_execute("SELECT * FROM bogo"));
match l.run(done) {
Err(Error::Db(ref e, _)) if e.code == SqlState::UndefinedTable => {}
Err(e) => panic!("unexpected error: {}", e),
Ok(_) => panic!("unexpected success"),
}
}

65
postgres/src/error.rs Normal file
View File

@ -0,0 +1,65 @@
//! Error types.
use std::fmt;
use std::io;
use std::error;
#[doc(inline)]
pub use postgres_shared::error::*;
/// An error encountered when communicating with the Postgres server.
#[derive(Debug)]
pub enum Error {
/// An error reported by the Postgres server.
Db(Box<DbError>),
/// An error communicating with the Postgres server.
Io(io::Error),
/// An error converting between Postgres and Rust types.
Conversion(Box<error::Error + Sync + Send>),
}
impl fmt::Display for Error {
fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
try!(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<DbError> for Error {
fn from(err: DbError) -> Error {
Error::Db(Box::new(err))
}
}
impl From<io::Error> for Error {
fn from(err: io::Error) -> Error {
Error::Io(err)
}
}
impl From<Error> for io::Error {
fn from(err: Error) -> io::Error {
io::Error::new(io::ErrorKind::Other, err)
}
}

77
postgres/src/lib.rs
View File

@ -79,19 +79,19 @@ extern crate log;
extern crate postgres_protocol;
extern crate postgres_shared;
use fallible_iterator::{FallibleIterator, FromFallibleIterator};
use fallible_iterator::FallibleIterator;
use std::cell::{Cell, RefCell};
use std::collections::{VecDeque, HashMap};
use std::fmt;
use std::io;
use std::mem;
use std::ops::Range;
use std::result;
use std::sync::Arc;
use std::time::Duration;
use postgres_protocol::authentication;
use postgres_protocol::message::backend;
use postgres_protocol::message::backend::{self, ErrorFields};
use postgres_protocol::message::frontend;
use postgres_shared::RowData;
use error::{Error, ConnectError, SqlState, DbError};
use tls::TlsHandshake;
@ -103,15 +103,13 @@ use stmt::{Statement, Column};
use transaction::{Transaction, IsolationLevel};
use types::{IsNull, Kind, Type, SessionInfo, Oid, Other, WrongType, ToSql, FromSql, Field};
#[doc(inline)]
pub use postgres_shared::error;
#[macro_use]
mod macros;
mod feature_check;
mod priv_io;
mod url;
pub mod error;
pub mod tls;
pub mod notification;
pub mod params;
@ -317,7 +315,7 @@ impl InnerConnection {
}
backend::Message::ReadyForQuery(_) => break,
backend::Message::ErrorResponse(body) => {
return DbError::new_connect(&mut body.fields())
return Err(connect_err(&mut body.fields()));
}
_ => return Err(ConnectError::Io(bad_response())),
}
@ -331,7 +329,7 @@ impl InnerConnection {
loop {
match try_desync!(self, self.stream.read_message()) {
backend::Message::NoticeResponse(body) => {
if let Ok(err) = DbError::new_raw(&mut body.fields()) {
if let Ok(err) = Err(err(&mut body.fields())) {
self.notice_handler.handle_notice(err);
}
}
@ -351,7 +349,7 @@ impl InnerConnection {
loop {
match try_desync!(self, self.stream.read_message_timeout(timeout)) {
Some(backend::Message::NoticeResponse(body)) => {
if let Ok(err) = DbError::new_raw(&mut body.fields()) {
if let Ok(err) = Err(err(&mut body.fields())) {
self.notice_handler.handle_notice(err);
}
}
@ -370,7 +368,7 @@ impl InnerConnection {
loop {
match try_desync!(self, self.stream.read_message_nonblocking()) {
Some(backend::Message::NoticeResponse(body)) => {
if let Ok(err) = DbError::new_raw(&mut body.fields()) {
if let Ok(err) = Err(err(&mut body.fields())) {
self.notice_handler.handle_notice(err);
}
}
@ -425,13 +423,13 @@ impl InnerConnection {
return Err(ConnectError::Io(io::Error::new(io::ErrorKind::Other,
"unsupported authentication")))
}
backend::Message::ErrorResponse(body) => return DbError::new_connect(&mut body.fields()),
backend::Message::ErrorResponse(body) => return Err(connect_err(&mut body.fields())),
_ => return Err(ConnectError::Io(bad_response())),
}
match try!(self.read_message()) {
backend::Message::AuthenticationOk => Ok(()),
backend::Message::ErrorResponse(body) => DbError::new_connect(&mut body.fields()),
backend::Message::ErrorResponse(body) => Err(connect_err(&mut body.fields())),
_ => Err(ConnectError::Io(bad_response())),
}
}
@ -452,7 +450,7 @@ impl InnerConnection {
backend::Message::ParseComplete => {}
backend::Message::ErrorResponse(body) => {
try!(self.wait_for_ready());
return DbError::new(&mut body.fields());
return Err(err(&mut body.fields()));
}
_ => bad_response!(self),
}
@ -509,7 +507,7 @@ impl InnerConnection {
backend::Message::DataRow(body) => consumer(try!(body.values().collect())),
backend::Message::ErrorResponse(body) => {
try!(self.wait_for_ready());
return DbError::new(&mut body.fields());
return Err(err(&mut body.fields()));
}
backend::Message::CopyInResponse(_) => {
try!(self.stream.write_message(|buf| {
@ -586,7 +584,7 @@ impl InnerConnection {
backend::Message::BindComplete => Ok(()),
backend::Message::ErrorResponse(body) => {
try!(self.wait_for_ready());
DbError::new(&mut body.fields())
Err(err(&mut body.fields()))
}
_ => {
self.desynchronized = true;
@ -639,7 +637,7 @@ impl InnerConnection {
try!(self.stream.flush());
let resp = match try!(self.read_message()) {
backend::Message::CloseComplete => Ok(()),
backend::Message::ErrorResponse(body) => DbError::new(&mut body.fields()),
backend::Message::ErrorResponse(body) => Err(err(&mut body.fields())),
_ => bad_response!(self),
};
try!(self.wait_for_ready());
@ -862,7 +860,7 @@ impl InnerConnection {
}
backend::Message::ErrorResponse(body) => {
try!(self.wait_for_ready());
return DbError::new(&mut body.fields());
return Err(err(&mut body.fields()));
}
_ => {}
}
@ -1328,46 +1326,17 @@ impl<'a> GenericConnection for Transaction<'a> {
}
}
struct RowData {
buf: Vec<u8>,
indices: Vec<Option<Range<usize>>>,
}
impl<'a> FromFallibleIterator<Option<&'a [u8]>> for RowData {
fn from_fallible_iterator<I>(mut it: I) -> result::Result<Self, I::Error>
where I: FallibleIterator<Item = Option<&'a [u8]>>
{
let mut row = RowData {
buf: vec![],
indices: Vec::with_capacity(it.size_hint().0),
};
while let Some(cell) = try!(it.next()) {
let index = match cell {
Some(cell) => {
let base = row.buf.len();
row.buf.extend_from_slice(cell);
Some(base..row.buf.len())
}
None => None,
};
row.indices.push(index);
}
Ok(row)
fn connect_err(fields: &mut ErrorFields) -> ConnectError {
match DbError::new(fields) {
Ok(err) => ConnectError::Db(Box::new(err)),
Err(err) => ConnectError::Io(err),
}
}
impl RowData {
fn len(&self) -> usize {
self.indices.len()
}
fn get(&self, index: usize) -> Option<&[u8]> {
match &self.indices[index] {
&Some(ref range) => Some(&self.buf[range.clone()]),
&None => None,
}
fn err(fields: &mut ErrorFields) -> Error {
match DbError::new(fields) {
Ok(err) => Error::Db(Box::new(err)),
Err(err) => Error::Io(err),
}
}

3
postgres/src/rows.rs
View File

@ -2,6 +2,7 @@
use fallible_iterator::FallibleIterator;
use postgres_protocol::message::frontend;
use postgres_shared::RowData;
use std::ascii::AsciiExt;
use std::collections::VecDeque;
use std::fmt;
@ -9,7 +10,7 @@ use std::io;
use std::ops::Deref;
use std::slice;
use {Result, SessionInfoNew, RowsNew, LazyRowsNew, StatementInternals, WrongTypeNew, RowData};
use {Result, SessionInfoNew, RowsNew, LazyRowsNew, StatementInternals, WrongTypeNew};
use transaction::Transaction;
use types::{FromSql, SessionInfo, WrongType};
use stmt::{Statement, Column};

17
postgres/src/stmt.rs
View File

@ -7,12 +7,13 @@ use std::fmt;
use std::io::{self, Read, Write};
use std::sync::Arc;
use postgres_protocol::message::{backend, frontend};
use postgres_shared::RowData;
use error::{Error, DbError};
use error::Error;
use types::{SessionInfo, Type, ToSql};
use rows::{Rows, LazyRows};
use transaction::Transaction;
use {bad_response, Connection, StatementInternals, Result, RowsNew, InnerConnection, RowData,
use {bad_response, err, Connection, StatementInternals, Result, RowsNew, InnerConnection,
SessionInfoNew, LazyRowsNew, ColumnNew, StatementInfo, TransactionInternals};
/// A prepared statement.
@ -138,7 +139,7 @@ impl<'conn> Statement<'conn> {
backend::Message::DataRow(_) => {}
backend::Message::ErrorResponse(body) => {
try!(conn.wait_for_ready());
return DbError::new(&mut body.fields());
return Err(err(&mut body.fields()));
}
backend::Message::CommandComplete(body) => {
num = parse_update_count(try!(body.tag()));
@ -162,7 +163,7 @@ impl<'conn> Statement<'conn> {
backend::Message::CopyDone => break,
backend::Message::ErrorResponse(body) => {
try!(conn.wait_for_ready());
return DbError::new(&mut body.fields());
return Err(err(&mut body.fields()));
}
_ => {}
}
@ -286,7 +287,7 @@ impl<'conn> Statement<'conn> {
}
backend::Message::ErrorResponse(body) => {
try!(conn.wait_for_ready());
return DbError::new(&mut body.fields());
return Err(err(&mut body.fields()));
}
_ => {
loop {
@ -347,7 +348,7 @@ impl<'conn> Statement<'conn> {
backend::Message::CommandComplete(body) => parse_update_count(try!(body.tag())),
backend::Message::ErrorResponse(body) => {
try!(info.conn.wait_for_ready());
return DbError::new(&mut body.fields());
return Err(err(&mut body.fields()));
}
_ => {
info.conn.desynchronized = true;
@ -415,7 +416,7 @@ impl<'conn> Statement<'conn> {
}
backend::Message::ErrorResponse(body) => {
try!(conn.wait_for_ready());
return DbError::new(&mut body.fields());
return Err(err(&mut body.fields()));
}
_ => {
loop {
@ -462,7 +463,7 @@ impl<'conn> Statement<'conn> {
loop {
if let backend::Message::ReadyForQuery(_) =
try!(info.conn.read_message()) {
return DbError::new(&mut body.fields());
return Err(err(&mut body.fields()));
}
}
}