diff --git a/nostr-sdk/CHANGELOG.md b/nostr-sdk/CHANGELOG.md index d6e392cda..49b0eb89b 100644 --- a/nostr-sdk/CHANGELOG.md +++ b/nostr-sdk/CHANGELOG.md @@ -32,12 +32,14 @@ ### Breaking change - Redesign `Relay::count_events` API +- Return `NotificationStream` from client and relay `notifications()` methods instead of a boxed stream. ### Added - Add `LocalRelay::connections_left` (https://github.com/nostrdevkit/nostr/pull/1459) - Add `LocalRelayBuilderNip42::relay_url` to make it possible to configure a custom `relay_url` for the nip42 challenge validation (https://github.com/nostrdevkit/nostr/pull/1476) - Add `LocalRelayBuilder::new_event_channel_size` for customizing the size of the channel used to notify new received events +- Add opt-in receiver gap reporting with `NotificationStream::with_gaps()`. ### Fixed diff --git a/nostr-sdk/src/client/mod.rs b/nostr-sdk/src/client/mod.rs index 7910123e1..839aeac3f 100644 --- a/nostr-sdk/src/client/mod.rs +++ b/nostr-sdk/src/client/mod.rs @@ -6,14 +6,11 @@ use std::borrow::Cow; use std::collections::HashMap; -use std::pin::Pin; use std::sync::{Arc, Weak}; use std::time::Duration; -use futures::{Stream, StreamExt}; use nostr::prelude::*; use nostr_database::prelude::*; -use tokio::sync::oneshot; mod api; mod builder; @@ -194,32 +191,45 @@ impl Client { /// /// The stream terminates when the client shutdowns. /// + /// Notifications lost when this receiver falls behind are skipped. + /// Use [`NotificationStream::with_gaps`] to observe such losses. + /// ///
When you call this method, you subscribe to the notifications channel from that precise moment. Anything received by relay/s before that moment is not included in the channel!
+ /// + /// # Examples + /// + /// ## Report notification gaps + /// + /// ```rust,no_run + /// # use nostr_sdk::prelude::*; + /// # async fn example(client: &Client) -> Result<(), Box> { + /// // Create the stream before subscribing so it can observe the response. + /// let mut notifications = client.notifications().with_gaps(); + /// client.subscribe(Filter::new().kind(Kind::TextNote)).await?; + /// + /// while let Some(update) = notifications.next().await { + /// match update { + /// Ok(ClientNotification::Event { event, .. }) => println!("{}", event.id), + /// Ok(ClientNotification::Shutdown) => break, + /// Ok(_) => {} + /// Err(gap) => { + /// eprintln!("Skipped {} notifications", gap.skipped); + /// // Reacquire any subscription coverage the application requires. + /// } + /// } + /// } + /// # Ok(()) + /// # } + /// ``` #[inline] - pub fn notifications(&self) -> Pin + Send>> { + pub fn notifications(&self) -> NotificationStream { if self.is_shutdown() { - return Box::pin(futures::stream::empty()); + return NotificationStream::empty(); } // Subscribe to notifications let rx = self.pool().notifications(); - - // Create a oneshot channel - let (tx, rx_done) = oneshot::channel(); - let mut tx: Option> = Some(tx); - - Box::pin( - NotificationStream::new(rx) - .inspect(move |notification| { - if let ClientNotification::Shutdown = ¬ification { - // Take the sender and send the oneshot notification - if let Some(tx) = tx.take() { - let _ = tx.send(()); - } - } - }) - .take_until(rx_done), - ) + NotificationStream::new(rx, |notification| notification.is_shutdown()) } /// Get relays from the relay pool. @@ -1190,6 +1200,7 @@ impl Client { #[cfg(test)] mod tests { + use futures::StreamExt; use nostr_gossip_memory::prelude::*; use super::*; diff --git a/nostr-sdk/src/client/notification.rs b/nostr-sdk/src/client/notification.rs index 34ad5970f..f2aafe4ab 100644 --- a/nostr-sdk/src/client/notification.rs +++ b/nostr-sdk/src/client/notification.rs @@ -39,3 +39,10 @@ pub enum ClientNotification { /// This notification variant is sent after [`Client::shutdown`](super::Client::shutdown) method is called and all connections have been closed. Shutdown, } + +impl ClientNotification { + #[inline] + pub(super) fn is_shutdown(&self) -> bool { + matches!(self, Self::Shutdown) + } +} diff --git a/nostr-sdk/src/lib.rs b/nostr-sdk/src/lib.rs index 0b82cc81d..284fb06da 100644 --- a/nostr-sdk/src/lib.rs +++ b/nostr-sdk/src/lib.rs @@ -20,7 +20,7 @@ pub mod prelude; pub mod proxy; pub mod relay; mod shared; -mod stream; +pub mod stream; #[cfg(test)] mod test_utils; pub mod transport; diff --git a/nostr-sdk/src/prelude.rs b/nostr-sdk/src/prelude.rs index 9683a3916..a24ade4ad 100644 --- a/nostr-sdk/src/prelude.rs +++ b/nostr-sdk/src/prelude.rs @@ -23,4 +23,5 @@ pub use crate::policy::*; #[cfg(not(target_arch = "wasm32"))] pub use crate::proxy::{self, *}; pub use crate::relay::{self, *}; +pub use crate::stream::{self, *}; pub use crate::*; diff --git a/nostr-sdk/src/relay/mod.rs b/nostr-sdk/src/relay/mod.rs index 153271c9e..cea7291da 100644 --- a/nostr-sdk/src/relay/mod.rs +++ b/nostr-sdk/src/relay/mod.rs @@ -4,15 +4,13 @@ use std::cmp; use std::collections::HashMap; #[cfg(not(target_arch = "wasm32"))] use std::net::SocketAddr; -use std::pin::Pin; use std::sync::Arc; use std::sync::atomic::{AtomicUsize, Ordering}; use std::time::Duration; use async_utility::time; -use futures::{Stream, StreamExt}; use nostr_database::prelude::*; -use tokio::sync::{broadcast, oneshot}; +use tokio::sync::broadcast; mod api; mod builder; @@ -211,35 +209,24 @@ impl Relay { /// /// The stream terminates when the relay shutdowns or is banned. /// + /// Notifications lost when this receiver falls behind are skipped. + /// Use [`NotificationStream::with_gaps`] to observe such losses. + /// ///
When you call this method, you subscribe to the notifications channel from that precise moment. Anything received by relay/s before that moment is not included in the channel!
#[inline] - pub fn notifications(&self) -> Pin + Send>> { + pub fn notifications(&self) -> NotificationStream { // If the relay is permanently unusable, return an empty stream let status: RelayStatus = self.status(); if status.is_banned() || status.is_shutdown() { - return Box::pin(futures::stream::empty()); + return NotificationStream::empty(); } // Subscribe to notifications let rx = self.inner.internal_notification_sender.subscribe(); - // Create a oneshot channel - let (tx, rx_done) = oneshot::channel(); - let mut tx: Option> = Some(tx); - - Box::pin( - NotificationStream::new(rx) - .inspect(move |notification| { - if let RelayNotification::RelayStatus { status } = ¬ification { - if status.is_banned() || status.is_shutdown() { - // Take the sender and send the oneshot notification - if let Some(tx) = tx.take() { - let _ = tx.send(()); - } - } - } - }) - .take_until(rx_done), + NotificationStream::new( + rx, + |notification| matches!(notification, RelayNotification::RelayStatus { status } if status.is_banned() || status.is_shutdown()), ) } @@ -432,6 +419,7 @@ mod tests { use std::sync::Arc; use async_utility::time; + use futures::StreamExt; use super::*; use crate::error::{Error, ErrorKind}; diff --git a/nostr-sdk/src/stream.rs b/nostr-sdk/src/stream.rs deleted file mode 100644 index af0ff33d7..000000000 --- a/nostr-sdk/src/stream.rs +++ /dev/null @@ -1,61 +0,0 @@ -use std::pin::Pin; -use std::task::{Context, Poll}; - -use futures::{Stream, StreamExt}; -use tokio::sync::broadcast; -use tokio::sync::mpsc::Receiver; -use tokio_stream::wrappers::BroadcastStream; - -pub(crate) struct ReceiverStream { - inner: Receiver, -} - -impl ReceiverStream { - #[inline] - pub(crate) fn new(recv: Receiver) -> Self { - Self { inner: recv } - } -} - -impl Stream for ReceiverStream { - type Item = T; - - fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - self.inner.poll_recv(cx) - } -} - -pub(crate) struct NotificationStream { - inner: BroadcastStream, -} - -impl NotificationStream -where - T: Clone + Send + 'static, -{ - #[inline] - pub(crate) fn new(inner: broadcast::Receiver) -> Self { - Self { - inner: BroadcastStream::new(inner), - } - } -} - -impl Stream for NotificationStream -where - T: Clone + Send + 'static, -{ - type Item = T; - - fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - loop { - match self.inner.poll_next_unpin(cx) { - Poll::Ready(Some(Ok(notification))) => return Poll::Ready(Some(notification)), - // Skip errors for now - Poll::Ready(Some(Err(..))) => continue, - Poll::Ready(None) => return Poll::Ready(None), - Poll::Pending => return Poll::Pending, - } - } - } -} diff --git a/nostr-sdk/src/stream/mod.rs b/nostr-sdk/src/stream/mod.rs new file mode 100644 index 000000000..ed1b78fa1 --- /dev/null +++ b/nostr-sdk/src/stream/mod.rs @@ -0,0 +1,30 @@ +//! Streams + +use std::pin::Pin; +use std::task::{Context, Poll}; + +use futures::Stream; +use tokio::sync::mpsc::Receiver; + +mod notification; + +pub use self::notification::*; + +pub(crate) struct ReceiverStream { + inner: Receiver, +} + +impl ReceiverStream { + #[inline] + pub(crate) fn new(recv: Receiver) -> Self { + Self { inner: recv } + } +} + +impl Stream for ReceiverStream { + type Item = T; + + fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + self.inner.poll_recv(cx) + } +} diff --git a/nostr-sdk/src/stream/notification.rs b/nostr-sdk/src/stream/notification.rs new file mode 100644 index 000000000..44f49d787 --- /dev/null +++ b/nostr-sdk/src/stream/notification.rs @@ -0,0 +1,219 @@ +use std::pin::Pin; +use std::task::{Context, Poll}; + +use futures::{Stream, StreamExt}; +use tokio::sync::broadcast; +use tokio_stream::wrappers::BroadcastStream; +use tokio_stream::wrappers::errors::BroadcastStreamRecvError; + +/// Notifications skipped by one notification stream receiver. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[non_exhaustive] +pub struct NotificationGap { + /// Number of notifications skipped by this receiver. + pub skipped: u64, +} + +/// A stream of notifications. +/// +/// By default, notifications lost when this receiver falls behind are skipped. +/// Use [`NotificationStream::with_gaps`] to observe such losses. +pub struct NotificationStream { + inner: Option>, + is_terminal: fn(&T) -> bool, +} + +impl NotificationStream +where + T: Clone + Send + 'static, +{ + #[inline] + pub(crate) fn new(inner: broadcast::Receiver, is_terminal: fn(&T) -> bool) -> Self { + Self { + inner: Some(BroadcastStream::new(inner)), + is_terminal, + } + } + + #[inline] + pub(crate) fn empty() -> Self { + Self { + inner: None, + is_terminal: |_| false, + } + } + + fn poll_next_with_err( + &mut self, + cx: &mut Context<'_>, + ) -> Poll>> { + let Some(inner) = &mut self.inner else { + return Poll::Ready(None); + }; + + match inner.poll_next_unpin(cx) { + Poll::Ready(Some(Ok(notification))) => { + if (self.is_terminal)(¬ification) { + self.inner = None; + } + Poll::Ready(Some(Ok(notification))) + } + Poll::Ready(Some(Err(e))) => Poll::Ready(Some(Err(e))), + Poll::Ready(None) => { + self.inner = None; + Poll::Ready(None) + } + Poll::Pending => Poll::Pending, + } + } + + /// Report notification loss to this receiver. + /// + /// An error reports the number of notifications skipped by this receiver. + /// The stream continues after the error. If no loss occurs, every item is `Ok`. + #[inline] + pub fn with_gaps(self) -> NotificationStreamWithGaps { + NotificationStreamWithGaps(self) + } +} + +impl Stream for NotificationStream +where + T: Clone + Send + Unpin + 'static, +{ + type Item = T; + + fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + let stream = self.get_mut(); + loop { + match stream.poll_next_with_err(cx) { + Poll::Ready(Some(Ok(notification))) => return Poll::Ready(Some(notification)), + Poll::Ready(Some(Err(..))) => continue, + Poll::Ready(None) => return Poll::Ready(None), + Poll::Pending => return Poll::Pending, + } + } + } +} + +/// A notification stream that reports losses to its receiver. +/// +/// A gap is emitted as an error, after which the stream continues. +pub struct NotificationStreamWithGaps(NotificationStream); + +impl Stream for NotificationStreamWithGaps +where + T: Clone + Send + Unpin + 'static, +{ + type Item = Result; + + fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + let stream: &mut Self = self.get_mut(); + match stream.0.poll_next_with_err(cx) { + Poll::Ready(Some(Ok(notification))) => Poll::Ready(Some(Ok(notification))), + Poll::Ready(Some(Err(BroadcastStreamRecvError::Lagged(skipped)))) => { + Poll::Ready(Some(Err(NotificationGap { skipped }))) + } + Poll::Ready(None) => Poll::Ready(None), + Poll::Pending => Poll::Pending, + } + } +} + +#[cfg(test)] +mod tests { + use futures::StreamExt; + use tokio::sync::broadcast; + + use super::{NotificationGap, NotificationStream}; + + #[tokio::test] + async fn reports_repeated_gaps_and_continues() { + let (sender, receiver) = broadcast::channel(2); + + let mut stream = NotificationStream::new(receiver, |value| *value == 99).with_gaps(); + + for value in 0..8 { + sender.send(value).unwrap(); + } + + assert_eq!( + stream.next().await, + Some(Err(NotificationGap { skipped: 6 })) + ); + assert_eq!(stream.next().await, Some(Ok(6))); + assert_eq!(stream.next().await, Some(Ok(7))); + + for value in 8..12 { + sender.send(value).unwrap(); + } + + assert_eq!( + stream.next().await, + Some(Err(NotificationGap { skipped: 2 })) + ); + assert_eq!(stream.next().await, Some(Ok(10))); + assert_eq!(stream.next().await, Some(Ok(11))); + + sender.send(99).unwrap(); + + assert_eq!(stream.next().await, Some(Ok(99))); + assert_eq!(stream.next().await, None); + } + + #[tokio::test] + async fn gap_is_local_to_the_slow_receiver() { + let (sender, receiver) = broadcast::channel(2); + + let mut fast = NotificationStream::new(receiver, |_| false).with_gaps(); + let mut slow = NotificationStream::new(sender.subscribe(), |_| false).with_gaps(); + + for value in 0..8 { + sender.send(value).unwrap(); + assert_eq!(fast.next().await, Some(Ok(value))); + } + + assert_eq!(slow.next().await, Some(Err(NotificationGap { skipped: 6 }))); + assert_eq!(slow.next().await, Some(Ok(6))); + assert_eq!(slow.next().await, Some(Ok(7))); + } + + #[tokio::test] + async fn ordinary_stream_skips_gaps_and_stops_after_terminal_notification() { + let (sender, receiver) = broadcast::channel(2); + + let mut stream = NotificationStream::new(receiver, |value| *value == 99); + + for value in 0..8 { + sender.send(value).unwrap(); + } + + assert_eq!(stream.next().await, Some(6)); + assert_eq!(stream.next().await, Some(7)); + + sender.send(99).unwrap(); + + assert_eq!(stream.next().await, Some(99)); + assert_eq!(stream.next().await, None); + } + + #[tokio::test] + async fn switching_to_gap_reporting_keeps_the_existing_receiver() { + let (sender, receiver) = broadcast::channel(2); + + let stream = NotificationStream::new(receiver, |_| false); + + for value in 0..4 { + sender.send(value).unwrap(); + } + + let mut stream = stream.with_gaps(); + + assert_eq!( + stream.next().await, + Some(Err(NotificationGap { skipped: 2 })) + ); + assert_eq!(stream.next().await, Some(Ok(2))); + assert_eq!(stream.next().await, Some(Ok(3))); + } +} diff --git a/signer/nostr-connect/src/client.rs b/signer/nostr-connect/src/client.rs index 94202e37b..f0aa870e1 100644 --- a/signer/nostr-connect/src/client.rs +++ b/signer/nostr-connect/src/client.rs @@ -12,7 +12,6 @@ use std::sync::Arc; use std::time::Duration; use async_utility::time; -use futures_core::stream::BoxStream; use nostr::nips::nip04::AsyncNip04; use nostr::nips::nip44::{self, AsyncNip44}; use nostr::nips::nip46::{ @@ -155,7 +154,7 @@ impl NostrConnect { Ok(remote_signer_public_key) } - async fn subscribe(&self) -> Result, Error> { + async fn subscribe(&self) -> Result, Error> { let public_key: PublicKey = self.client_keys.public_key(); let filter = Filter::new() @@ -163,7 +162,7 @@ impl NostrConnect { .kind(Kind::NostrConnect) .limit(0); - let notifications = self.client.notifications(); + let notifications: NotificationStream = self.client.notifications(); // Subscribe self.client.subscribe(filter).await?; @@ -392,7 +391,7 @@ impl NostrConnect { async fn get_remote_signer_public_key( client_keys: &Keys, expected_secret: &str, - mut notifications: BoxStream<'_, ClientNotification>, + mut notifications: NotificationStream, timeout: Duration, ) -> Result { time::timeout(Some(timeout), async {