Move notification stuff to its own module

This commit is contained in:
Steven Fackler 2015-09-18 20:55:01 -07:00
parent ff4248c737
commit 008f14b459
3 changed files with 86 additions and 64 deletions

View File

@ -78,6 +78,7 @@ use message::BackendMessage::*;
use message::FrontendMessage::*; use message::FrontendMessage::*;
use message::{FrontendMessage, BackendMessage, RowDescriptionEntry}; use message::{FrontendMessage, BackendMessage, RowDescriptionEntry};
use message::{WriteMessage, ReadMessage}; use message::{WriteMessage, ReadMessage};
use notification::{Notifications, Notification};
use rows::{Rows, LazyRows}; use rows::{Rows, LazyRows};
use stmt::{Statement, Column}; use stmt::{Statement, Column};
use types::{IsNull, Kind, Type, SessionInfo, Oid, Other}; use types::{IsNull, Kind, Type, SessionInfo, Oid, Other};
@ -97,6 +98,7 @@ pub mod io;
pub mod rows; pub mod rows;
pub mod stmt; pub mod stmt;
pub mod types; pub mod types;
pub mod notification;
const TYPEINFO_QUERY: &'static str = "t"; const TYPEINFO_QUERY: &'static str = "t";
@ -231,68 +233,6 @@ impl HandleNotice for LoggingNoticeHandler {
} }
} }
/// An asynchronous notification.
#[derive(Clone, Debug)]
pub struct Notification {
/// The process ID of the notifying backend process.
pub pid: u32,
/// The name of the channel that the notify has been raised on.
pub channel: String,
/// The "payload" string passed from the notifying process.
pub payload: String,
}
/// An iterator over asynchronous notifications.
pub struct Notifications<'conn> {
conn: &'conn Connection
}
impl<'a> fmt::Debug for Notifications<'a> {
fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
DebugStruct::new(fmt, "Notifications")
.field("pending", &self.conn.conn.borrow().notifications.len())
.finish()
}
}
impl<'conn> Iterator for Notifications<'conn> {
type Item = Notification;
/// Returns the oldest pending notification or `None` if there are none.
///
/// ## Note
///
/// `next` may return `Some` notification after returning `None` if a new
/// notification was received.
fn next(&mut self) -> Option<Notification> {
self.conn.conn.borrow_mut().notifications.pop_front()
}
}
impl<'conn> Notifications<'conn> {
/// Returns the oldest pending notification.
///
/// If no notifications are pending, blocks until one arrives.
pub fn next_block(&mut self) -> Result<Notification> {
if let Some(notification) = self.next() {
return Ok(notification);
}
let mut conn = self.conn.conn.borrow_mut();
check_desync!(conn);
match try!(conn.read_message_with_notification()) {
NotificationResponse { pid, channel, payload } => {
Ok(Notification {
pid: pid,
channel: channel,
payload: payload
})
}
_ => unreachable!()
}
}
}
/// Contains information necessary to cancel queries for a session. /// Contains information necessary to cancel queries for a session.
#[derive(Copy, Clone, Debug)] #[derive(Copy, Clone, Debug)]
pub struct CancelData { pub struct CancelData {
@ -947,7 +887,7 @@ impl Connection {
/// ///
/// Use the `LISTEN` command to register this connection for notifications. /// Use the `LISTEN` command to register this connection for notifications.
pub fn notifications<'a>(&'a self) -> Notifications<'a> { pub fn notifications<'a>(&'a self) -> Notifications<'a> {
Notifications { conn: self } Notifications::new(self)
} }
/// Creates a new prepared statement. /// Creates a new prepared statement.
@ -1444,3 +1384,7 @@ trait StatementInternals<'conn> {
trait ColumnNew { trait ColumnNew {
fn new(name: String, type_: Type) -> Column; fn new(name: String, type_: Type) -> Column;
} }
trait NotificationsNew<'conn> {
fn new(conn: &'conn Connection) -> Notifications<'conn>;
}

78
src/notification.rs Normal file
View File

@ -0,0 +1,78 @@
//! Asynchronous notifications.
use debug_builders::DebugStruct;
use std::fmt;
use {Result, Connection, NotificationsNew};
use message::BackendMessage::NotificationResponse;
/// An asynchronous notification.
#[derive(Clone, Debug)]
pub struct Notification {
/// The process ID of the notifying backend process.
pub pid: u32,
/// The name of the channel that the notify has been raised on.
pub channel: String,
/// The "payload" string passed from the notifying process.
pub payload: String,
}
/// An iterator over asynchronous notifications.
pub struct Notifications<'conn> {
conn: &'conn Connection
}
impl<'a> fmt::Debug for Notifications<'a> {
fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
DebugStruct::new(fmt, "Notifications")
.field("pending", &self.conn.conn.borrow().notifications.len())
.finish()
}
}
impl<'conn> NotificationsNew<'conn> for Notifications<'conn> {
fn new(conn: &'conn Connection) -> Notifications<'conn> {
Notifications {
conn: conn,
}
}
}
impl<'conn> Iterator for Notifications<'conn> {
type Item = Notification;
/// Returns the oldest pending notification or `None` if there are none.
///
/// ## Note
///
/// `next` may return `Some` notification after returning `None` if a new
/// notification was received.
fn next(&mut self) -> Option<Notification> {
self.conn.conn.borrow_mut().notifications.pop_front()
}
}
impl<'conn> Notifications<'conn> {
/// Returns the oldest pending notification.
///
/// If no notifications are pending, blocks until one arrives.
pub fn next_block(&mut self) -> Result<Notification> {
if let Some(notification) = self.next() {
return Ok(notification);
}
let mut conn = self.conn.conn.borrow_mut();
check_desync!(conn);
match try!(conn.read_message_with_notification()) {
NotificationResponse { pid, channel, payload } => {
Ok(Notification {
pid: pid,
channel: channel,
payload: payload
})
}
_ => unreachable!()
}
}
}

View File

@ -11,7 +11,6 @@ use std::io;
use std::io::prelude::*; use std::io::prelude::*;
use postgres::{HandleNotice, use postgres::{HandleNotice,
Notification,
Connection, Connection,
GenericConnection, GenericConnection,
SslMode, SslMode,
@ -27,6 +26,7 @@ use postgres::error::SqlState::{SyntaxError,
CardinalityViolation}; CardinalityViolation};
use postgres::error::ErrorPosition::Normal; use postgres::error::ErrorPosition::Normal;
use postgres::rows::RowIndex; use postgres::rows::RowIndex;
use postgres::notification::Notification;
macro_rules! or_panic { macro_rules! or_panic {
($e:expr) => ( ($e:expr) => (