Merge pull request sfackler#320 from dunkelstern/feature/db-errors-in-notifications

Add handling of DB-Errors in Notification processing

Author: sfackler

postgres/src/notification.rs

@@ -3,9 +3,11 @@
 use fallible_iterator::{FallibleIterator, IntoFallibleIterator};
 use std::fmt;
 use std::time::Duration;
-use postgres_protocol::message::backend;
+use postgres_protocol::message::backend::{self, ErrorFields};
+use error::DbError;
 
 #[doc(inline)]
+use postgres_shared;
 pub use postgres_shared::Notification;
 
 use {desynchronized, Result, Connection};
@@ -110,6 +112,7 @@ impl<'a> FallibleIterator for Iter<'a> {
                     payload: body.message()?.to_owned(),
                 }))
             }
+            Ok(Some(backend::Message::ErrorResponse(body))) => Err(err(&mut body.fields())),
             Ok(None) => Ok(None),
             Err(err) => Err(err.into()),
             _ => unreachable!(),
@@ -149,6 +152,7 @@ impl<'a> FallibleIterator for BlockingIter<'a> {
                     payload: body.message()?.to_owned(),
                 }))
             }
+            Ok(backend::Message::ErrorResponse(body)) => Err(err(&mut body.fields())),
             Err(err) => Err(err.into()),
             _ => unreachable!(),
         }
@@ -185,6 +189,7 @@ impl<'a> FallibleIterator for TimeoutIter<'a> {
                     payload: body.message()?.to_owned(),
                 }))
             }
+            Ok(Some(backend::Message::ErrorResponse(body))) => Err(err(&mut body.fields())),
             Ok(None) => Ok(None),
             Err(err) => Err(err.into()),
             _ => unreachable!(),
@@ -195,3 +200,10 @@ impl<'a> FallibleIterator for TimeoutIter<'a> {
         (self.conn.0.borrow().notifications.len(), None)
     }
 }
+
+fn err(fields: &mut ErrorFields) -> Error {
+    match DbError::new(fields) {
+        Ok(err) => postgres_shared::error::db(err),
+        Err(err) => err.into(),
+    }
+}

postgres/tests/test.rs

@@ -901,6 +901,30 @@ fn test_notification_next_timeout() {
     assert!(it.next().unwrap().is_none());
 }
 
+#[test]
+fn test_notification_disconnect() {
+        let conn = or_panic!(Connection::connect(
+        "postgres://postgres@localhost:5433",
+        TlsMode::None,
+    ));
+    or_panic!(conn.execute("LISTEN test_notifications_disconnect", &[]));
+
+    let _t = thread::spawn(|| {
+        let conn = or_panic!(Connection::connect(
+            "postgres://postgres@localhost:5433",
+            TlsMode::None,
+        ));
+        thread::sleep(Duration::from_millis(500));
+        or_panic!(conn.execute(
+            "SELECT pg_terminate_backend(pid) FROM pg_stat_activity WHERE query = 'LISTEN test_notifications_disconnect'",
+            &[],
+        ));
+    });
+
+    let notifications = conn.notifications();
+    assert!(notifications.blocking_iter().next().is_err());
+}
+
 #[test]
 // This test is pretty sad, but I don't think there's a better way :(
 fn test_cancel_query() {