Make notification iterators fallible

Author: sfackler

src/notification.rs

@@ -1,5 +1,6 @@
 //! Asynchronous notifications.
 
+use fallible_iterator::{FallibleIterator, IntoFallibleIterator};
 use std::fmt;
 use std::time::Duration;
 
@@ -42,7 +43,7 @@ impl<'conn> Notifications<'conn> {
         self.len() == 0
     }
 
-    /// Returns an iterator over pending notifications.
+    /// Returns a fallible iterator over pending notifications.
     ///
     /// # Note
     ///
@@ -52,16 +53,16 @@ impl<'conn> Notifications<'conn> {
         Iter { conn: self.conn }
     }
 
-    /// Returns an iterator over notifications that blocks until one is
+    /// Returns a fallible 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.
+    /// Returns a fallible iterator over notifications that blocks for a limited
+    /// time waiting to receive one if none are pending.
     ///
     /// # Note
     ///
@@ -75,11 +76,12 @@ impl<'conn> Notifications<'conn> {
     }
 }
 
-impl<'a, 'conn> IntoIterator for &'a Notifications<'conn> {
-    type Item = Result<Notification>;
+impl<'a, 'conn> IntoFallibleIterator for &'a Notifications<'conn> {
+    type Item = Notification;
+    type Error = Error;
     type IntoIter = Iter<'a>;
 
-    fn into_iter(self) -> Iter<'a> {
+    fn into_fallible_iterator(self) -> Iter<'a> {
         self.iter()
     }
 }
@@ -90,35 +92,36 @@ impl<'conn> NotificationsNew<'conn> for Notifications<'conn> {
     }
 }
 
-/// An iterator over pending notifications.
+/// A fallible iterator over pending notifications.
 pub struct Iter<'a> {
     conn: &'a Connection,
 }
 
-impl<'a> Iterator for Iter<'a> {
-    type Item = Result<Notification>;
+impl<'a> FallibleIterator for Iter<'a> {
+    type Item = Notification;
+    type Error = Error;
 
-    fn next(&mut self) -> Option<Result<Notification>> {
+    fn next(&mut self) -> Result<Option<Notification>> {
         let mut conn = self.conn.conn.borrow_mut();
 
         if let Some(notification) = conn.notifications.pop_front() {
-            return Some(Ok(notification));
+            return Ok(Some(notification));
         }
 
         if conn.is_desynchronized() {
-            return Some(Err(Error::Io(desynchronized())));
+            return Err(Error::Io(desynchronized()));
         }
 
         match conn.read_message_with_notification_nonblocking() {
             Ok(Some(Backend::NotificationResponse { process_id, channel, payload })) => {
-                Some(Ok(Notification {
+                Ok(Some(Notification {
                     process_id: process_id,
                     channel: channel,
                     payload: payload,
                 }))
             }
-            Ok(None) => None,
-            Err(err) => Some(Err(Error::Io(err))),
+            Ok(None) => Ok(None),
+            Err(err) => Err(Error::Io(err)),
             _ => unreachable!(),
         }
     }
@@ -133,36 +136,33 @@ pub struct BlockingIter<'a> {
     conn: &'a Connection,
 }
 
-impl<'a> Iterator for BlockingIter<'a> {
-    type Item = Result<Notification>;
+impl<'a> FallibleIterator for BlockingIter<'a> {
+    type Item = Notification;
+    type Error = Error;
 
-    fn next(&mut self) -> Option<Result<Notification>> {
+    fn next(&mut self) -> Result<Option<Notification>> {
         let mut conn = self.conn.conn.borrow_mut();
 
         if let Some(notification) = conn.notifications.pop_front() {
-            return Some(Ok(notification));
+            return Ok(Some(notification));
         }
 
         if conn.is_desynchronized() {
-            return Some(Err(Error::Io(desynchronized())));
+            return Err(Error::Io(desynchronized()));
         }
 
         match conn.read_message_with_notification() {
             Ok(Backend::NotificationResponse { process_id, channel, payload }) => {
-                Some(Ok(Notification {
+                Ok(Some(Notification {
                     process_id: process_id,
                     channel: channel,
                     payload: payload,
                 }))
             }
-            Err(err) => Some(Err(Error::Io(err))),
+            Err(err) => Err(Error::Io(err)),
             _ => unreachable!(),
         }
     }
-
-    fn size_hint(&self) -> (usize, Option<usize>) {
-        (usize::max_value(), None)
-    }
 }
 
 /// An iterator over notifications which will block for a period of time if
@@ -172,30 +172,31 @@ pub struct TimeoutIter<'a> {
     timeout: Duration,
 }
 
-impl<'a> Iterator for TimeoutIter<'a> {
-    type Item = Result<Notification>;
+impl<'a> FallibleIterator for TimeoutIter<'a> {
+    type Item = Notification;
+    type Error = Error;
 
-    fn next(&mut self) -> Option<Result<Notification>> {
+    fn next(&mut self) -> Result<Option<Notification>> {
         let mut conn = self.conn.conn.borrow_mut();
 
         if let Some(notification) = conn.notifications.pop_front() {
-            return Some(Ok(notification));
+            return Ok(Some(notification));
         }
 
         if conn.is_desynchronized() {
-            return Some(Err(Error::Io(desynchronized())));
+            return Err(Error::Io(desynchronized()));
         }
 
         match conn.read_message_with_notification_timeout(self.timeout) {
             Ok(Some(Backend::NotificationResponse { process_id, channel, payload })) => {
-                Some(Ok(Notification {
+                Ok(Some(Notification {
                     process_id: process_id,
                     channel: channel,
                     payload: payload,
                 }))
             }
-            Ok(None) => None,
-            Err(err) => Some(Err(Error::Io(err))),
+            Ok(None) => Ok(None),
+            Err(err) => Err(Error::Io(err)),
             _ => unreachable!(),
         }
     }

tests/test.rs

@@ -1,3 +1,4 @@
+extern crate fallible_iterator;
 #[macro_use]
 extern crate postgres;
 extern crate url;
@@ -6,10 +7,7 @@ extern crate openssl;
 #[cfg(feature = "with-security-framework")]
 extern crate security_framework;
 
-use std::thread;
-use std::io;
-use std::time::Duration;
-
+use fallible_iterator::FallibleIterator;
 use postgres::{HandleNotice, Connection, GenericConnection, TlsMode};
 use postgres::transaction::{self, IsolationLevel};
 use postgres::error::{Error, ConnectError, DbError};
@@ -24,6 +22,9 @@ use postgres::error::ErrorPosition::Normal;
 use postgres::rows::RowIndex;
 use postgres::notification::Notification;
 use postgres::params::IntoConnectParams;
+use std::thread;
+use std::io;
+use std::time::Duration;
 
 macro_rules! or_panic {
     ($e:expr) => (
@@ -557,7 +558,7 @@ fn test_custom_notice_handler() {
 #[test]
 fn test_notification_iterator_none() {
     let conn = or_panic!(Connection::connect("postgres://postgres@localhost", TlsMode::None));
-    assert!(conn.notifications().iter().next().is_none());
+    assert!(conn.notifications().iter().next().unwrap().is_none());
 }
 
 fn check_notification(expected: Notification, actual: Notification) {
@@ -585,15 +586,15 @@ fn test_notification_iterator_some() {
         channel: "test_notification_iterator_one_channel2".to_string(),
         payload: "world".to_string()
     }, it.next().unwrap().unwrap());
-    assert!(it.next().is_none());
+    assert!(it.next().unwrap().is_none());
 
     or_panic!(conn.execute("NOTIFY test_notification_iterator_one_channel, '!'", &[]));
     check_notification(Notification {
         process_id: 0,
         channel: "test_notification_iterator_one_channel".to_string(),
         payload: "!".to_string()
     }, it.next().unwrap().unwrap());
-    assert!(it.next().is_none());
+    assert!(it.next().unwrap().is_none());
 }
 
 #[test]
@@ -612,7 +613,7 @@ fn test_notifications_next_block() {
         process_id: 0,
         channel: "test_notifications_next_block".to_string(),
         payload: "foo".to_string()
-    }, or_panic!(notifications.blocking_iter().next().unwrap()));
+    }, notifications.blocking_iter().next().unwrap().unwrap());
 }
 
 #[test]
@@ -634,9 +635,9 @@ fn test_notification_next_timeout() {
         process_id: 0,
         channel: "test_notifications_next_timeout".to_string(),
         payload: "foo".to_string()
-    }, or_panic!(it.next().unwrap()));
+    }, it.next().unwrap().unwrap());
 
-    assert!(it.next().is_none());
+    assert!(it.next().unwrap().is_none());
 }
 
 #[test]