Merge branch 'release-v0.10.1' into release

Author: sfackler

.travis.yml

@@ -3,11 +3,11 @@ sudo: required
 rust:
 - nightly
 - beta
-- stable
+- 1.3.0
 addons:
   postgresql: 9.4
 before_script:
 - "./.travis/setup.sh"
 script:
 - cargo test
-- cargo test --features "uuid rustc-serialize time unix_socket serde chrono openssl"
+- cargo test --features "uuid rustc-serialize time unix_socket serde_json chrono openssl"

Cargo.toml

@@ -1,11 +1,11 @@
 [package]
 name = "postgres"
-version = "0.10.0"
+version = "0.10.1"
 authors = ["Steven Fackler <sfackler@gmail.com>"]
 license = "MIT"
 description = "A native PostgreSQL driver"
 repository = "https://github.com/sfackler/rust-postgres"
-documentation = "https://sfackler.github.io/rust-postgres/doc/v0.10.0/postgres"
+documentation = "https://sfackler.github.io/rust-postgres/doc/v0.10.1/postgres"
 readme = "README.md"
 keywords = ["database", "sql"]
 build = "build.rs"
@@ -25,15 +25,17 @@ phf_codegen = "0.7"
 
 [dependencies]
 bufstream = "0.1"
-byteorder = "0.3"
+byteorder = ">= 0.3, < 0.5"
 log = "0.3"
 phf = "0.7"
 rustc-serialize = "0.3"
+net2 = { version = "0.2", features = ["nightly"] }
 chrono = { version = "0.2.14", optional = true }
 openssl = { version = "0.6.4", optional = true }
-serde = { version = "0.3", optional = true }
+serde = { version = "0.3", optional = true } # Delete for 0.11
+serde_json = { version = "0.6", optional = true }
 time = { version = "0.1.14", optional = true }
-unix_socket = { version = ">= 0.3, < 0.5", optional = true }
+unix_socket = { version = ">= 0.3, < 0.5", optional = true, features = ["socket_timeout"] }
 uuid = { version = "0.1", optional = true }
 
 [dev-dependencies]

README.md

@@ -1,7 +1,7 @@
 # Rust-Postgres
 A native PostgreSQL driver for Rust.
 
-[Documentation](https://sfackler.github.io/rust-postgres/doc/v0.10.0/postgres)
+[Documentation](https://sfackler.github.io/rust-postgres/doc/v0.10.1/postgres)
 
 [![Build Status](https://travis-ci.org/sfackler/rust-postgres.png?branch=master)](https://travis-ci.org/sfackler/rust-postgres) [![Latest Version](https://img.shields.io/crates/v/postgres.svg)](https://crates.io/crates/postgres)
 
@@ -57,7 +57,7 @@ fn main() {
 ```
 
 ## Requirements
-* **Rust** - Rust-Postgres is developed against the 1.2 release of Rust
+* **Rust** - Rust-Postgres is developed against the 1.3 release of Rust
     available on http://www.rust-lang.org. It should also compile against more
     recent releases.
 
@@ -200,7 +200,7 @@ types. The driver currently supports the following conversions:
             <td>
                 <a href="https://github.com/rust-lang/rustc-serialize">serialize::json::Json</a>
                 and
-                <a href="https://github.com/erickt/serde">serde::json::Value</a>
+                <a href="https://github.com/erickt/serde_json">serde_json::Value</a>
                 (<a href="#optional-features">optional</a>)
             </td>
             <td>JSON, JSONB</td>
@@ -284,7 +284,7 @@ implementations for `uuid`'s `Uuid` type.
 [JSON and JSONB](http://www.postgresql.org/docs/9.4/static/datatype-json.html)
 support is provided optionally by the `rustc-serialize` feature, which adds
 `ToSql` and `FromSql` implementations for `rustc-serialize`'s `Json` type, and
-the `serde` feature, which adds implementations for `serde`'s `json::Value`
+the `serde` feature, which adds implementations for `serde_json`'s `Value`
 type.
 
 ### TIMESTAMP/TIMESTAMPTZ/DATE/TIME types

src/lib.rs

@@ -24,7 +24,7 @@
 //!                   )", &[]).unwrap();
 //!     let me = Person {
 //!         id: 0,
-//!         name: "Steven".to_string(),
+//!         name: "Steven".to_owned(),
 //!         data: None
 //!     };
 //!     conn.execute("INSERT INTO person (name, data) VALUES ($1, $2)",
@@ -41,7 +41,7 @@
 //!     }
 //! }
 //! ```
-#![doc(html_root_url="https://sfackler.github.io/rust-postgres/doc/v0.10.0")]
+#![doc(html_root_url="https://sfackler.github.io/rust-postgres/doc/v0.10.1")]
 #![warn(missing_docs)]
 
 extern crate bufstream;
@@ -52,6 +52,7 @@ extern crate phf;
 extern crate rustc_serialize as serialize;
 #[cfg(feature = "unix_socket")]
 extern crate unix_socket;
+extern crate net2;
 
 use bufstream::BufStream;
 use md5::Md5;
@@ -67,6 +68,7 @@ use std::io::prelude::*;
 use std::marker::Sync as StdSync;
 use std::mem;
 use std::result;
+use std::time::Duration;
 #[cfg(feature = "unix_socket")]
 use std::path::PathBuf;
 
@@ -455,10 +457,12 @@ impl InnerConnection {
 
     fn setup_typeinfo_query(&mut self) -> result::Result<(), ConnectError> {
         match self.raw_prepare(TYPEINFO_QUERY,
-                               "SELECT t.typname, t.typelem, r.rngsubtype \
+                               "SELECT t.typname, t.typelem, r.rngsubtype, n.nspname \
                                 FROM pg_catalog.pg_type t \
                                 LEFT OUTER JOIN pg_catalog.pg_range r \
                                     ON r.rngtypid = t.oid \
+                                INNER JOIN pg_catalog.pg_namespace n \
+                                    ON t.typnamespace = n.oid \
                                 WHERE t.oid = $1") {
             Ok(..) => return Ok(()),
             Err(Error::IoError(e)) => return Err(ConnectError::IoError(e)),
@@ -469,9 +473,11 @@ impl InnerConnection {
         }
 
         match self.raw_prepare(TYPEINFO_QUERY,
-                               "SELECT typname, typelem, NULL::OID \
-                                FROM pg_catalog.pg_type \
-                                WHERE oid = $1") {
+                               "SELECT t.typname, t.typelem, NULL::OID, n.nspname \
+                                FROM pg_catalog.pg_type t \
+                                INNER JOIN pg_catalog.pg_namespace n \
+                                    ON t.typnamespace = n.oid \
+                                WHERE t.oid = $1") {
             Ok(..) => Ok(()),
             Err(Error::IoError(e)) => Err(ConnectError::IoError(e)),
             Err(Error::DbError(e)) => Err(ConnectError::DbError(e)),
@@ -504,6 +510,24 @@ impl InnerConnection {
         }
     }
 
+    fn read_message_with_notification_timeout(&mut self, timeout: Duration)
+                                              -> std::io::Result<Option<BackendMessage>> {
+        debug_assert!(!self.desynchronized);
+        loop {
+            match try_desync!(self, self.stream.read_message_timeout(timeout)) {
+                Some(NoticeResponse { fields }) => {
+                    if let Ok(err) = DbError::new_raw(fields) {
+                        self.notice_handler.handle_notice(err);
+                    }
+                }
+                Some(ParameterStatus { parameter, value }) => {
+                    self.parameters.insert(parameter, value);
+                }
+                val => return Ok(val)
+            }
+        }
+    }
+
     fn read_message(&mut self) -> std_io::Result<BackendMessage> {
         loop {
             match try!(self.read_message_with_notification()) {
@@ -661,7 +685,7 @@ impl InnerConnection {
     }
 
     fn get_type(&mut self, oid: Oid) -> Result<Type> {
-        if let Some(ty) = Type::new(oid) {
+        if let Some(ty) = Type::from_oid(oid) {
             return Ok(ty);
         }
 
@@ -696,7 +720,7 @@ impl InnerConnection {
             }
             _ => bad_response!(self)
         }
-        let (name, elem_oid, rngsubtype): (String, Oid, Option<Oid>) =
+        let (name, elem_oid, rngsubtype, schema): (String, Oid, Option<Oid>, String) =
                 match try!(self.read_message()) {
             DataRow { row } => {
                 let ctx = SessionInfo::new(self);
@@ -708,6 +732,9 @@ impl InnerConnection {
                                                  &ctx)),
                  try!(FromSql::from_sql_nullable(&Type::Oid,
                                                  row[2].as_ref().map(|r| &**r).as_mut(),
+                                                 &ctx)),
+                 try!(FromSql::from_sql_nullable(&Type::Name,
+                                                 row[3].as_ref().map(|r| &**r).as_mut(),
                                                  &ctx)))
             }
             ErrorResponse { fields } => {
@@ -735,7 +762,7 @@ impl InnerConnection {
             }
         };
 
-        let type_ = Type::Other(Box::new(Other::new(name, oid, kind)));
+        let type_ = Type::Other(Box::new(Other::new(name, oid, kind, schema)));
         self.unknown_types.insert(oid, type_.clone());
         Ok(type_)
     }
@@ -860,7 +887,7 @@ impl Connection {
     ///     target: ConnectTarget::Unix(some_crazy_path),
     ///     port: None,
     ///     user: Some(UserInfo {
-    ///         user: "postgres".to_string(),
+    ///         user: "postgres".to_owned(),
     ///         password: None
     ///     }),
     ///     database: None,
@@ -911,7 +938,7 @@ impl Connection {
         self.conn.borrow_mut().prepare(query, self)
     }
 
-    /// Creates cached prepared statement.
+    /// Creates a cached prepared statement.
     ///
     /// Like `prepare`, except that the statement is only prepared once over
     /// the lifetime of the connection and then cached. If the same statement
@@ -1337,7 +1364,7 @@ impl<'a> GenericConnection for Transaction<'a> {
 }
 
 trait OtherNew {
-    fn new(name: String, oid: Oid, kind: Kind) -> Other;
+    fn new(name: String, oid: Oid, kind: Kind, schema: String) -> Other;
 }
 
 trait DbErrorNew {
@@ -1346,10 +1373,6 @@ trait DbErrorNew {
     fn new<T>(fields: Vec<(u8, String)>) -> Result<T>;
 }
 
-trait TypeNew {
-    fn new(oid: Oid) -> Option<Type>;
-}
-
 trait RowsNew<'a> {
     fn new(stmt: &'a Statement<'a>, data: Vec<Vec<Option<Vec<u8>>>>) -> Rows<'a>;
 }

src/message.rs

@@ -1,10 +1,12 @@
 use std::io;
 use std::io::prelude::*;
 use std::mem;
+use std::time::Duration;
 use byteorder::{BigEndian, ReadBytesExt, WriteBytesExt};
 
 use types::Oid;
 use util;
+use priv_io::ReadTimeout;
 
 use self::BackendMessage::*;
 use self::FrontendMessage::*;
@@ -282,12 +284,39 @@ impl<R: BufRead> ReadCStr for R {
 #[doc(hidden)]
 pub trait ReadMessage {
     fn read_message(&mut self) -> io::Result<BackendMessage>;
+
+    fn read_message_timeout(&mut self, timeout: Duration)
+                            -> io::Result<Option<BackendMessage>>;
+
+    fn finish_read_message(&mut self, ident: u8) -> io::Result<BackendMessage>;
 }
 
-impl<R: BufRead> ReadMessage for R {
+impl<R: BufRead + ReadTimeout> ReadMessage for R {
     fn read_message(&mut self) -> io::Result<BackendMessage> {
         let ident = try!(self.read_u8());
+        self.finish_read_message(ident)
+    }
+
+    fn read_message_timeout(&mut self, timeout: Duration)
+                            -> io::Result<Option<BackendMessage>> {
+        try!(self.set_read_timeout(Some(timeout)));
+        let ident = self.read_u8();
+        try!(self.set_read_timeout(None));
+
+        match ident {
+            Ok(ident) => self.finish_read_message(ident).map(Some),
+            Err(e) => {
+                let e: io::Error = e.into();
+                if e.kind() == io::ErrorKind::WouldBlock || e.kind() == io::ErrorKind::TimedOut {
+                    Ok(None)
+                } else {
+                    Err(e)
+                }
+            }
+        }
+    }
 
+    fn finish_read_message(&mut self, ident: u8) -> io::Result<BackendMessage> {
         // subtract size of length value
         let len = try!(self.read_u32::<BigEndian>()) - mem::size_of::<u32>() as u32;
         let mut rdr = self.by_ref().take(len as u64);

src/notification.rs

@@ -1,6 +1,7 @@
 //! Asynchronous notifications.
 
 use std::fmt;
+use std::time::Duration;
 
 use {desynchronized, Result, Connection, NotificationsNew};
 use message::BackendMessage::NotificationResponse;
@@ -41,22 +42,37 @@ impl<'conn> Notifications<'conn> {
     /// # Note
     ///
     /// This iterator may start returning `Some` after previously returning
-    /// `None` if more notifications are received.
+    /// `None` if more notifications are received. However, those notifications
+    /// will not be registered until the connection is used in some way.
     pub fn iter<'a>(&'a self) -> Iter<'a> {
         Iter {
             conn: self.conn,
         }
     }
 
-    /// Returns an iterator over notifications, blocking until one is received
-    /// if none are pending.
+    /// Returns an iterator over notifications that blocks until one is
+    /// received if none are pending.
     ///
     /// The iterator will never return `None`.
     pub fn blocking_iter<'a>(&'a self) -> BlockingIter<'a> {
         BlockingIter {
             conn: self.conn,
         }
     }
+
+    /// Returns an iterator over notifications that blocks for a limited time
+    /// waiting to receive one if none are pending.
+    ///
+    /// # Note
+    ///
+    /// This iterator may start returning `Some` after previously returning
+    /// `None` if more notifications are received.
+    pub fn timeout_iter<'a>(&'a self, timeout: Duration) -> TimeoutIter<'a> {
+        TimeoutIter {
+            conn: self.conn,
+            timeout: timeout,
+        }
+    }
 }
 
 impl<'a, 'conn> IntoIterator for &'a Notifications<'conn> {
@@ -121,3 +137,39 @@ impl<'a> Iterator for BlockingIter<'a> {
         }
     }
 }
+
+/// An iterator over notifications which will block for a period of time if
+/// none are pending.
+pub struct TimeoutIter<'a> {
+    conn: &'a Connection,
+    timeout: Duration,
+}
+
+impl<'a> Iterator for TimeoutIter<'a> {
+    type Item = Result<Notification>;
+
+    fn next(&mut self) -> Option<Result<Notification>> {
+        let mut conn = self.conn.conn.borrow_mut();
+
+        if let Some(notification) = conn.notifications.pop_front() {
+            return Some(Ok(notification));
+        }
+
+        if conn.is_desynchronized() {
+            return Some(Err(Error::IoError(desynchronized())));
+        }
+
+        match conn.read_message_with_notification_timeout(self.timeout) {
+            Ok(Some(NotificationResponse { pid, channel, payload })) => {
+                Some(Ok(Notification {
+                    pid: pid,
+                    channel: channel,
+                    payload: payload
+                }))
+            }
+            Ok(None) => None,
+            Err(err) => Some(Err(Error::IoError(err))),
+            _ => unreachable!()
+        }
+    }
+}