diff --git a/postgres-shared/src/lib.rs b/postgres-shared/src/lib.rs index cdde9268..406b3917 100644 --- a/postgres-shared/src/lib.rs +++ b/postgres-shared/src/lib.rs @@ -6,8 +6,11 @@ extern crate phf; extern crate postgres_protocol; use fallible_iterator::{FallibleIterator, FromFallibleIterator}; +use std::ascii::AsciiExt; use std::ops::Range; +use types::Type; + pub mod error; pub mod params; pub mod types; @@ -54,3 +57,67 @@ impl RowData { } } } + +pub struct Column { + name: String, + type_: Type, +} + +impl Column { + #[doc(hidden)] + pub fn new(name: String, type_: Type) -> Column { + Column { + name: name, + type_: type_, + } + } + + pub fn name(&self) -> &str { + &self.name + } + + pub fn type_(&self) -> &Type { + &self.type_ + } +} + +/// A trait implemented by types that can index into columns of a row. +pub trait RowIndex { + /// Returns the index of the appropriate column, or `None` if no such + /// column exists. + fn idx(&self, stmt: &[Column]) -> Option; +} + +impl RowIndex for usize { + #[inline] + fn idx(&self, stmt: &[Column]) -> Option { + if *self >= stmt.len() { + None + } else { + Some(*self) + } + } +} + +impl<'a> RowIndex for str { + #[inline] + fn idx(&self, stmt: &[Column]) -> Option { + if let Some(idx) = stmt.iter().position(|d| d.name() == self) { + return Some(idx); + }; + + // FIXME ASCII-only case insensitivity isn't really the right thing to + // do. Postgres itself uses a dubious wrapper around tolower and JDBC + // uses the US locale. + stmt.iter().position(|d| d.name().eq_ignore_ascii_case(self)) + } +} + +impl<'a, T> RowIndex for &'a T + where T: RowIndex +{ + #[inline] + fn idx(&self, columns: &[Column]) -> Option { + T::idx(*self, columns) + } +} diff --git a/postgres-tokio/src/lib.rs b/postgres-tokio/src/lib.rs index b8c5d2a5..e7782ded 100644 --- a/postgres-tokio/src/lib.rs +++ b/postgres-tokio/src/lib.rs @@ -14,19 +14,21 @@ use postgres_protocol::message::{backend, frontend}; use postgres_protocol::message::backend::{ErrorResponseBody, ErrorFields}; use postgres_shared::RowData; use std::collections::HashMap; +use std::error::Error as StdError; use std::fmt; use std::io; use std::mem; +use std::sync::Arc; use std::sync::mpsc::{self, Sender, Receiver}; use tokio_core::reactor::Handle; #[doc(inline)] -pub use postgres_shared::{params, types}; +pub use postgres_shared::{params, types, Column, RowIndex}; use error::{ConnectError, Error, DbError}; use params::{ConnectParams, IntoConnectParams}; use stream::PostgresStream; -use types::{Oid, Type, ToSql, SessionInfo, IsNull}; +use types::{Oid, Type, ToSql, SessionInfo, IsNull, FromSql, WrongType}; pub mod error; mod stream; @@ -293,30 +295,6 @@ impl Connection { .boxed() } - #[allow(dead_code)] - fn read_rows(self, mut rows: Vec) -> BoxFuture<(Vec, Connection), Error> { - self.0.read() - .map_err(Error::Io) - .and_then(|(m, s)| { - match m { - backend::Message::EmptyQueryResponse | - backend::Message::CommandComplete(_) => Connection(s).ready(rows).boxed(), - backend::Message::DataRow(body) => { - match body.values().collect() { - Ok(row) => { - rows.push(row); - Connection(s).read_rows(rows) - } - Err(e) => Err(Error::Io(e)).into_future().boxed(), - } - } - backend::Message::ErrorResponse(body) => Connection(s).ready_err(body), - _ => Err(bad_message()).into_future().boxed(), - } - }) - .boxed() - } - fn ready(self, t: T) -> BoxFuture<(T, Connection), Error> where T: 'static + Send { @@ -452,7 +430,7 @@ impl Connection { s.get_types(r.into_iter(), vec![], |f| f.1, - |f, t| Column { name: f.0, type_: t }) + |f, t| Column::new(f.0, t)) .map(|(r, s)| (p, r, s)) }) .boxed() @@ -594,7 +572,7 @@ impl Connection { close_sender: conn.0.close_sender.clone(), name: name, params: params, - columns: columns, + columns: Arc::new(columns), }; (stmt, conn) }) @@ -615,26 +593,11 @@ impl Connection { } } -pub struct Column { - name: String, - type_: Type, -} - -impl Column { - pub fn name(&self) -> &str { - &self.name - } - - pub fn type_(&self) -> &Type { - &self.type_ - } -} - pub struct Statement { close_sender: Sender<(u8, String)>, name: String, params: Vec, - columns: Vec, + columns: Arc>, } impl Drop for Statement { @@ -664,6 +627,42 @@ impl Statement { } } +pub struct Row { + columns: Arc>, + data: RowData, +} + +impl Row { + pub fn get(&self, idx: I) -> T + where T: FromSql, + I: RowIndex + fmt::Debug + { + match self.try_get(&idx) { + Ok(Some(v)) => v, + Ok(None) => panic!("no such column {:?}", idx), + Err(e) => panic!("error retrieving row {:?}: {}", idx, e), + } + } + + pub fn try_get(&self, idx: I) -> Result, Box> + where T: FromSql, + I: RowIndex + { + let idx = match idx.idx(&self.columns) { + Some(idx) => idx, + None => return Ok(None), + }; + + let ty = self.columns[idx].type_(); + if !T::accepts(ty) { + return Err(Box::new(WrongType::new(ty.clone()))); + } + + // FIXME + T::from_sql_nullable(ty, self.data.get(idx), &SessionInfo::new(&HashMap::new())).map(Some) + } +} + fn connect_err(fields: &mut ErrorFields) -> ConnectError { match DbError::new(fields) { Ok(err) => ConnectError::Db(Box::new(err)),