|
| 1 | +//! Asynchronous notifications. |
| 2 | +
|
| 3 | +use crate::connection::ConnectionRef; |
| 4 | +use crate::{Error, Notification}; |
| 5 | +use fallible_iterator::FallibleIterator; |
| 6 | +use futures::{ready, FutureExt}; |
| 7 | +use std::task::Poll; |
| 8 | +use std::time::Duration; |
| 9 | +use tokio::time::{self, Delay, Instant}; |
| 10 | + |
| 11 | +/// Notifications from a PostgreSQL backend. |
| 12 | +pub struct Notifications<'a> { |
| 13 | + connection: ConnectionRef<'a>, |
| 14 | +} |
| 15 | + |
| 16 | +impl<'a> Notifications<'a> { |
| 17 | + pub(crate) fn new(connection: ConnectionRef<'a>) -> Notifications<'a> { |
| 18 | + Notifications { connection } |
| 19 | + } |
| 20 | + |
| 21 | + /// Returns the number of already buffered pending notifications. |
| 22 | + pub fn len(&self) -> usize { |
| 23 | + self.connection.notifications().len() |
| 24 | + } |
| 25 | + |
| 26 | + /// Determines if there are any already buffered pending notifications. |
| 27 | + pub fn is_empty(&self) -> bool { |
| 28 | + self.connection.notifications().is_empty() |
| 29 | + } |
| 30 | + |
| 31 | + /// Returns a nonblocking iterator over notifications. |
| 32 | + /// |
| 33 | + /// If there are no already buffered pending notifications, this iterator will poll the connection but will not |
| 34 | + /// block waiting on notifications over the network. A return value of `None` either indicates that there are no |
| 35 | + /// pending notifications or that the server has disconnected. |
| 36 | + /// |
| 37 | + /// # Note |
| 38 | + /// |
| 39 | + /// This iterator may start returning `Some` after previously returning `None` if more notifications are received. |
| 40 | + pub fn iter(&mut self) -> Iter<'_> { |
| 41 | + Iter { |
| 42 | + connection: self.connection.as_ref(), |
| 43 | + } |
| 44 | + } |
| 45 | + |
| 46 | + /// Returns a blocking iterator over notifications. |
| 47 | + /// |
| 48 | + /// If there are no already buffered pending notifications, this iterator will block indefinitely waiting on the |
| 49 | + /// PostgreSQL backend server to send one. It will only return `None` if the server has disconnected. |
| 50 | + pub fn blocking_iter(&mut self) -> BlockingIter<'_> { |
| 51 | + BlockingIter { |
| 52 | + connection: self.connection.as_ref(), |
| 53 | + } |
| 54 | + } |
| 55 | + |
| 56 | + /// Returns an iterator over notifications which blocks a limited amount of time. |
| 57 | + /// |
| 58 | + /// If there are no already buffered pending notifications, this iterator will block waiting on the PostgreSQL |
| 59 | + /// backend server to send one up to the provided timeout. A return value of `None` either indicates that there are |
| 60 | + /// no pending notifications or that the server has disconnected. |
| 61 | + /// |
| 62 | + /// # Note |
| 63 | + /// |
| 64 | + /// This iterator may start returning `Some` after previously returning `None` if more notifications are received. |
| 65 | + pub fn timeout_iter(&mut self, timeout: Duration) -> TimeoutIter<'_> { |
| 66 | + TimeoutIter { |
| 67 | + delay: self.connection.enter(|| time::delay_for(timeout)), |
| 68 | + timeout, |
| 69 | + connection: self.connection.as_ref(), |
| 70 | + } |
| 71 | + } |
| 72 | +} |
| 73 | + |
| 74 | +/// A nonblocking iterator over pending notifications. |
| 75 | +pub struct Iter<'a> { |
| 76 | + connection: ConnectionRef<'a>, |
| 77 | +} |
| 78 | + |
| 79 | +impl<'a> FallibleIterator for Iter<'a> { |
| 80 | + type Item = Notification; |
| 81 | + type Error = Error; |
| 82 | + |
| 83 | + fn next(&mut self) -> Result<Option<Self::Item>, Self::Error> { |
| 84 | + if let Some(notification) = self.connection.notifications_mut().pop_front() { |
| 85 | + return Ok(Some(notification)); |
| 86 | + } |
| 87 | + |
| 88 | + self.connection |
| 89 | + .poll_block_on(|_, notifications, _| Poll::Ready(Ok(notifications.pop_front()))) |
| 90 | + } |
| 91 | + |
| 92 | + fn size_hint(&self) -> (usize, Option<usize>) { |
| 93 | + (self.connection.notifications().len(), None) |
| 94 | + } |
| 95 | +} |
| 96 | + |
| 97 | +/// A blocking iterator over pending notifications. |
| 98 | +pub struct BlockingIter<'a> { |
| 99 | + connection: ConnectionRef<'a>, |
| 100 | +} |
| 101 | + |
| 102 | +impl<'a> FallibleIterator for BlockingIter<'a> { |
| 103 | + type Item = Notification; |
| 104 | + type Error = Error; |
| 105 | + |
| 106 | + fn next(&mut self) -> Result<Option<Self::Item>, Self::Error> { |
| 107 | + if let Some(notification) = self.connection.notifications_mut().pop_front() { |
| 108 | + return Ok(Some(notification)); |
| 109 | + } |
| 110 | + |
| 111 | + self.connection |
| 112 | + .poll_block_on(|_, notifications, done| match notifications.pop_front() { |
| 113 | + Some(notification) => Poll::Ready(Ok(Some(notification))), |
| 114 | + None if done => Poll::Ready(Ok(None)), |
| 115 | + None => Poll::Pending, |
| 116 | + }) |
| 117 | + } |
| 118 | + |
| 119 | + fn size_hint(&self) -> (usize, Option<usize>) { |
| 120 | + (self.connection.notifications().len(), None) |
| 121 | + } |
| 122 | +} |
| 123 | + |
| 124 | +/// A time-limited blocking iterator over pending notifications. |
| 125 | +pub struct TimeoutIter<'a> { |
| 126 | + connection: ConnectionRef<'a>, |
| 127 | + delay: Delay, |
| 128 | + timeout: Duration, |
| 129 | +} |
| 130 | + |
| 131 | +impl<'a> FallibleIterator for TimeoutIter<'a> { |
| 132 | + type Item = Notification; |
| 133 | + type Error = Error; |
| 134 | + |
| 135 | + fn next(&mut self) -> Result<Option<Self::Item>, Self::Error> { |
| 136 | + if let Some(notification) = self.connection.notifications_mut().pop_front() { |
| 137 | + self.delay.reset(Instant::now() + self.timeout); |
| 138 | + return Ok(Some(notification)); |
| 139 | + } |
| 140 | + |
| 141 | + let delay = &mut self.delay; |
| 142 | + let timeout = self.timeout; |
| 143 | + self.connection.poll_block_on(|cx, notifications, done| { |
| 144 | + match notifications.pop_front() { |
| 145 | + Some(notification) => { |
| 146 | + delay.reset(Instant::now() + timeout); |
| 147 | + return Poll::Ready(Ok(Some(notification))); |
| 148 | + } |
| 149 | + None if done => return Poll::Ready(Ok(None)), |
| 150 | + None => {} |
| 151 | + } |
| 152 | + |
| 153 | + ready!(delay.poll_unpin(cx)); |
| 154 | + Poll::Ready(Ok(None)) |
| 155 | + }) |
| 156 | + } |
| 157 | + |
| 158 | + fn size_hint(&self) -> (usize, Option<usize>) { |
| 159 | + (self.connection.notifications().len(), None) |
| 160 | + } |
| 161 | +} |
0 commit comments