use std::fmt;
use std::time::Duration;
use {desynchronized, Result, Connection, NotificationsNew};
use message::BackendMessage::NotificationResponse;
use error::Error;
#[derive(Clone, Debug)]
pub struct Notification {
pub pid: u32,
pub channel: String,
pub payload: String,
}
pub struct Notifications<'conn> {
conn: &'conn Connection,
}
impl<'a> fmt::Debug for Notifications<'a> {
fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
fmt.debug_struct("Notifications")
.field("pending", &self.len())
.finish()
}
}
impl<'conn> Notifications<'conn> {
pub fn len(&self) -> usize {
self.conn.conn.borrow().notifications.len()
}
pub fn iter<'a>(&'a self) -> Iter<'a> {
Iter { conn: self.conn }
}
pub fn blocking_iter<'a>(&'a self) -> BlockingIter<'a> {
BlockingIter { conn: self.conn }
}
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> {
type Item = Result<Notification>;
type IntoIter = Iter<'a>;
fn into_iter(self) -> Iter<'a> {
self.iter()
}
}
impl<'conn> NotificationsNew<'conn> for Notifications<'conn> {
fn new(conn: &'conn Connection) -> Notifications<'conn> {
Notifications { conn: conn }
}
}
pub struct Iter<'a> {
conn: &'a Connection,
}
impl<'a> Iterator for Iter<'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::Io(desynchronized())));
}
match conn.read_message_with_notification_nonblocking() {
Ok(Some(NotificationResponse { pid, channel, payload })) => {
Some(Ok(Notification {
pid: pid,
channel: channel,
payload: payload,
}))
}
Ok(None) => None,
Err(err) => Some(Err(Error::Io(err))),
_ => unreachable!(),
}
}
fn size_hint(&self) -> (usize, Option<usize>) {
(self.conn.conn.borrow().notifications.len(), None)
}
}
pub struct BlockingIter<'a> {
conn: &'a Connection,
}
impl<'a> Iterator for BlockingIter<'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::Io(desynchronized())));
}
match conn.read_message_with_notification() {
Ok(NotificationResponse { pid, channel, payload }) => {
Some(Ok(Notification {
pid: pid,
channel: channel,
payload: payload,
}))
}
Err(err) => Some(Err(Error::Io(err))),
_ => unreachable!(),
}
}
fn size_hint(&self) -> (usize, Option<usize>) {
(usize::max_value(), None)
}
}
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::Io(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::Io(err))),
_ => unreachable!(),
}
}
fn size_hint(&self) -> (usize, Option<usize>) {
(self.conn.conn.borrow().notifications.len(), None)
}
}