From 16ce48a51016100d98973960b8fc97ddf2b5ff1d Mon Sep 17 00:00:00 2001 From: Steven Fackler Date: Mon, 19 Sep 2016 23:46:25 -0400 Subject: [PATCH] Update for postgres_protocol changes --- src/lib.rs | 56 ++++++++++++++++++++++++------------------------ src/priv_io.rs | 9 ++++++++ src/rows.rs | 4 ++-- src/stmt.rs | 2 +- src/types/mod.rs | 24 ++++++++------------- 5 files changed, 49 insertions(+), 46 deletions(-) diff --git a/src/lib.rs b/src/lib.rs index b14f599f..5c18d3ee 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -165,13 +165,8 @@ pub fn cancel_query(params: T, let params = try!(params.into_connect_params().map_err(ConnectError::ConnectParams)); let mut socket = try!(priv_io::initialize_stream(¶ms, tls)); - let message = frontend::CancelRequest { - process_id: data.process_id, - secret_key: data.secret_key, - }; let mut buf = vec![]; - try!(frontend::Message::write(&message, &mut buf)); - + frontend::cancel_request(data.process_id, data.secret_key, &mut buf); try!(socket.write_all(&buf)); try!(socket.flush()); @@ -516,23 +511,31 @@ impl InnerConnection { debug!("executing statement {} with parameters: {:?}", stmt_name, params); - let mut values = vec![]; - for (param, ty) in params.iter().zip(param_types) { - let mut buf = vec![]; - match try!(param.to_sql_checked(ty, &mut buf, &SessionInfo::new(self)) - .map_err(Error::Conversion)) { - IsNull::Yes => values.push(None), - IsNull::No => values.push(Some(buf)), + + { + let info = SessionInfo::new(&self.parameters); + let r = self.stream.write_message2(|buf| { + frontend::bind(portal_name, + &stmt_name, + Some(1), + params.iter().zip(param_types), + |(param, ty), buf| { + match param.to_sql_checked(ty, buf, &info) { + Ok(IsNull::Yes) => Ok(postgres_protocol::IsNull::Yes), + Ok(IsNull::No) => Ok(postgres_protocol::IsNull::No), + Err(e) => Err(e), + } + }, + Some(1), + buf) + }); + match r { + Ok(()) => {} + Err(frontend::BindError::Conversion(e)) => return Err(Error::Conversion(e)), + Err(frontend::BindError::Serialization(e)) => return Err(Error::Io(e)), } } - try!(self.stream.write_message(&frontend::Bind { - portal: portal_name, - statement: &stmt_name, - formats: &[1], - values: &values, - result_formats: &[1], - })); try!(self.stream.write_message(&frontend::Execute { portal: portal_name, max_rows: row_limit, @@ -592,10 +595,7 @@ impl InnerConnection { } fn close_statement(&mut self, name: &str, type_: u8) -> Result<()> { - try!(self.stream.write_message(&frontend::Close { - variant: type_, - name: name, - })); + try!(self.stream.write_message2(|buf| frontend::close(type_, name, buf))); try!(self.stream.write_message(&frontend::Sync)); try!(self.stream.flush()); let resp = match try!(self.read_message()) { @@ -662,7 +662,7 @@ impl InnerConnection { let row = rows.pop_front().unwrap(); let (name, type_, elem_oid, rngsubtype, basetype, schema, relid) = { - let ctx = SessionInfo::new(self); + let ctx = SessionInfo::new(&self.parameters); let name = try!(String::from_sql(&Type::Name, &mut &**row[0].as_ref().unwrap(), &ctx) .map_err(Error::Conversion)); let type_ = try!(i8::from_sql(&Type::Char, &mut &**row[1].as_ref().unwrap(), &ctx) @@ -740,7 +740,7 @@ impl InnerConnection { let mut rows = VecDeque::new(); try!(self.read_rows(&mut rows)); - let ctx = SessionInfo::new(self); + let ctx = SessionInfo::new(&self.parameters); let mut variants = vec![]; for row in rows { variants.push(try!(String::from_sql(&Type::Name, &mut &**row[0].as_ref().unwrap(), &ctx) @@ -776,7 +776,7 @@ impl InnerConnection { let mut fields = vec![]; for row in rows { let (name, type_) = { - let ctx = SessionInfo::new(self); + let ctx = SessionInfo::new(&self.parameters); let name = try!(String::from_sql(&Type::Name, &mut &**row[0].as_ref().unwrap(), &ctx) .map_err(Error::Conversion)); @@ -1316,7 +1316,7 @@ trait LazyRowsNew<'trans, 'stmt> { } trait SessionInfoNew<'a> { - fn new(conn: &'a InnerConnection) -> SessionInfo<'a>; + fn new(params: &'a HashMap) -> SessionInfo<'a>; } trait StatementInternals<'conn> { diff --git a/src/priv_io.rs b/src/priv_io.rs index b0630ac7..1d5ed27e 100644 --- a/src/priv_io.rs +++ b/src/priv_io.rs @@ -39,6 +39,15 @@ impl MessageStream { self.stream.get_ref() } + pub fn write_message2(&mut self, f: F) -> Result<(), E> + where F: FnOnce(&mut Vec) -> Result<(), E>, + E: From + { + self.buf.clear(); + try!(f(&mut self.buf)); + self.stream.write_all(&self.buf).map_err(From::from) + } + pub fn write_message(&mut self, message: &frontend::Message) -> io::Result<()> { self.buf.clear(); try!(frontend::Message::write(message, &mut self.buf)); diff --git a/src/rows.rs b/src/rows.rs index 11ad9b13..8a094d7d 100644 --- a/src/rows.rs +++ b/src/rows.rs @@ -237,8 +237,8 @@ impl<'a> Row<'a> { } let conn = self.stmt.conn().conn.borrow(); let value = match self.data[idx] { - Some(ref data) => FromSql::from_sql(ty, &mut &**data, &SessionInfo::new(&*conn)), - None => FromSql::from_sql_null(ty, &SessionInfo::new(&*conn)), + Some(ref data) => FromSql::from_sql(ty, &mut &**data, &SessionInfo::new(&conn.parameters)), + None => FromSql::from_sql_null(ty, &SessionInfo::new(&conn.parameters)), }; Some(value.map_err(Error::Conversion)) } diff --git a/src/stmt.rs b/src/stmt.rs index cb3ed996..bc2c412a 100644 --- a/src/stmt.rs +++ b/src/stmt.rs @@ -528,7 +528,7 @@ impl<'a> CopyInfo<'a> { /// Returns session info for the associated connection. pub fn session_info<'b>(&'b self) -> SessionInfo<'b> { - SessionInfo::new(&*self.conn) + SessionInfo::new(&self.conn.parameters) } } diff --git a/src/types/mod.rs b/src/types/mod.rs index 2c78426a..ba1082a6 100644 --- a/src/types/mod.rs +++ b/src/types/mod.rs @@ -1,6 +1,7 @@ //! Traits dealing with Postgres data types use fallible_iterator::FallibleIterator; +use postgres_protocol; use postgres_protocol::types::{self, ArrayDimension}; use std::collections::HashMap; use std::error::Error; @@ -12,7 +13,7 @@ pub use postgres_protocol::Oid; pub use self::type_gen::Type; pub use self::special::{Date, Timestamp}; -use {SessionInfoNew, InnerConnection, OtherNew, WrongTypeNew, FieldNew}; +use {SessionInfoNew, OtherNew, WrongTypeNew, FieldNew}; /// Generates a simple implementation of `ToSql::accepts` which accepts the /// types passed to it. @@ -82,13 +83,14 @@ mod special; mod type_gen; /// A structure providing information for conversion methods. +#[derive(Debug)] pub struct SessionInfo<'a> { - conn: &'a InnerConnection, + parameters: &'a HashMap, } impl<'a> SessionInfoNew<'a> for SessionInfo<'a> { - fn new(conn: &'a InnerConnection) -> SessionInfo<'a> { - SessionInfo { conn: conn } + fn new(parameters: &'a HashMap) -> SessionInfo<'a> { + SessionInfo { parameters: parameters } } } @@ -96,15 +98,7 @@ impl<'a> SessionInfo<'a> { /// Returns the value of the specified Postgres backend parameter, such /// as `timezone` or `server_version`. pub fn parameter(&self, param: &str) -> Option<&'a str> { - self.conn.parameters.get(param).map(|s| &**s) - } -} - -impl<'a> fmt::Debug for SessionInfo<'a> { - fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result { - fmt.debug_struct("SessionInfo") - .field("parameters", &self.conn.parameters) - .finish() + self.parameters.get(param).map(|s| &**s) } } @@ -607,8 +601,8 @@ impl<'a, T: ToSql> ToSql for &'a [T] { self.iter(), |e, w| { match try!(e.to_sql(member_type, w, ctx)) { - IsNull::No => Ok(types::IsNull::No), - IsNull::Yes => Ok(types::IsNull::Yes), + IsNull::No => Ok(postgres_protocol::IsNull::No), + IsNull::Yes => Ok(postgres_protocol::IsNull::Yes), } }, w));