Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Delegate Future implementation to Next struct #1372

Open
wants to merge 3 commits into
base: main
Choose a base branch
from
Open

Delegate Future implementation to Next struct #1372

Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
9b57356
Delegate Future implementation to Next struct
asonix Sep 16, 2021
da7eb26
Rename .next to .next_event to avoid conflict with Iterator
asonix Sep 16, 2021
bd3697f
Fix doctest
asonix Sep 16, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
100 changes: 58 additions & 42 deletions src/subscriber.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -107,9 +107,9 @@ type Senders = Map<usize, (Option<Waker>, SyncSender<OneShot<Option<Event>>>)>;
/// ```
/// Aynchronous, non-blocking subscriber:
///
/// `Subscription` implements `Future<Output=Option<Event>>`.
/// `Subscription` provides a `next` method which returns an `impl Future<Output=Option<Event>>`.
///
/// `while let Some(event) = (&mut subscriber).await { /* use it */ }`
/// `while let Some(event) = subscriber.next().await { /* use it */ }`
pub struct Subscriber {
id: usize,
rx: Receiver<OneShot<Option<Event>>>,
Expand All @@ -125,6 +125,44 @@ impl Drop for Subscriber {
}

impl Subscriber {
/// Creates a future that resolves to the next value of the
/// subscriber, or None if the backing `Db` shuts down
pub fn next(&mut self) -> impl Future<Output = Option<Event>> + '_ {
Copy link
Sponsor Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Although this is returning a concrete Next<'_> type in the implementation, we don't need to leak the new type in the public API in the event it needs to change later for some reason

Copy link
Sponsor Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Realizing now that next here conflicts with Iterator's next Maybe naming this something else would be better

Next { subscriber: self }
}

fn poll_next(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<Option<Event>> {
Copy link
Sponsor Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This API matches the current definition of the Stream trait in Futures 0.3, so if that ends up stabilizing as-is then a Stream implementation will be "free"

Otherwise some changes will need to be made later down the line, but it's a private method anyway

loop {
let mut future_rx = if let Some(future_rx) = self.existing.take() {
future_rx
} else {
match self.rx.try_recv() {
Ok(future_rx) => future_rx,
Err(TryRecvError::Empty) => break,
Err(TryRecvError::Disconnected) => {
return Poll::Ready(None)
}
}
};

match Future::poll(Pin::new(&mut future_rx), cx) {
Poll::Ready(Some(event)) => return Poll::Ready(event),
Poll::Ready(None) => continue,
Poll::Pending => {
self.existing = Some(future_rx);
return Poll::Pending;
}
}
}
let mut home = self.home.write();
let entry = home.get_mut(&self.id).unwrap();
entry.0 = Some(cx.waker().clone());
Poll::Pending
}

/// Attempts to wait for a value on this `Subscriber`, returning
/// an error if no event arrives within the provided `Duration`
/// or if the backing `Db` shuts down.
Expand Down Expand Up @@ -165,42 +203,6 @@ impl Subscriber {
}
}

impl Future for Subscriber {
type Output = Option<Event>;

fn poll(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<Self::Output> {
loop {
let mut future_rx = if let Some(future_rx) = self.existing.take() {
future_rx
} else {
match self.rx.try_recv() {
Ok(future_rx) => future_rx,
Err(TryRecvError::Empty) => break,
Err(TryRecvError::Disconnected) => {
return Poll::Ready(None)
}
}
};

match Future::poll(Pin::new(&mut future_rx), cx) {
Poll::Ready(Some(event)) => return Poll::Ready(event),
Poll::Ready(None) => continue,
Poll::Pending => {
self.existing = Some(future_rx);
return Poll::Pending;
}
}
}
let mut home = self.home.write();
let entry = home.get_mut(&self.id).unwrap();
entry.0 = Some(cx.waker().clone());
Poll::Pending
}
}

impl Iterator for Subscriber {
type Item = Event;

Expand All @@ -216,6 +218,23 @@ impl Iterator for Subscriber {
}
}

#[doc(hidden)]
#[must_use = "futures do nothing unless you `.await` or poll them"]
struct Next<'a> {
subscriber: &'a mut Subscriber,
}

impl<'a> Future for Next<'a> {
type Output = Option<Event>;

fn poll(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<Self::Output> {
Pin::new(&mut *self.subscriber).poll_next(cx)
Copy link
Sponsor Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

self.subscriber is an &mut Subscriber but it can't move, so &mut *self.subscriber to make it move-able.

}
}

#[derive(Debug, Default)]
pub(crate) struct Subscribers {
watched: RwLock<BTreeMap<Vec<u8>, Arc<RwLock<Senders>>>>,
Expand All @@ -239,10 +258,7 @@ impl Drop for Subscribers {
}

impl Subscribers {
pub(crate) fn register(
&self,
prefix: &[u8]
) -> Subscriber {
pub(crate) fn register(&self, prefix: &[u8]) -> Subscriber {
Copy link
Sponsor Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Unintentional formatting change. I can put this back if you want

self.ever_used.store(true, Relaxed);
let r_mu = {
let r_mu = self.watched.read();
Expand Down