Add support for event streaming

Signed-off-by: Yuki Kishimoto <yukikishimoto@protonmail.com>

Author: yukibtc

Cargo.lock

@@ -23,10 +23,12 @@ uniffi-cli = ["uniffi/cli"] # required for the `uniffi-bindgen` binary
 [dependencies]
 async-trait = "0.1"
 async-wsocket = "0.13"
+futures-util = "0.3.31"
 nostr = { version = "0.41.0", features = ["std", "all-nips"] }
 nostr-connect = "0.41.0"
 nostr-sdk  = { version = "0.41.0", default-features = false, features = ["all-nips"] }
 nwc = "0.41.0"
+tokio = { version = "1", features = ["sync"] }
 tracing = { version = "0.1", features = ["std"] }
 tracing-subscriber = "0.3"
 uniffi = { version = "=0.28.3", features = ["tokio"] }

Cargo.toml

@@ -7,6 +7,7 @@ use std::ops::Deref;
 use std::sync::Arc;
 use std::time::Duration;
 
+use nostr::RelayUrl;
 use nostr_sdk::client::Client as ClientSdk;
 use nostr_sdk::pool::RelayPoolNotification as RelayPoolNotificationSdk;
 use nostr_sdk::SubscriptionId;
@@ -32,6 +33,7 @@ use crate::protocol::nips::nip59::UnwrappedGift;
 use crate::protocol::signer::NostrSigner;
 use crate::relay::options::{SubscribeAutoCloseOptions, SyncOptions};
 use crate::relay::{Relay, RelayOptions};
+use crate::stream::EventStream;
 
 #[derive(Object)]
 pub struct Client {
@@ -407,6 +409,70 @@ impl Client {
             .into())
     }
 
+    /// Stream events from relays
+    ///
+    /// # Overview
+    ///
+    /// This is an **auto-closing subscription** and will be closed automatically on `EOSE`.
+    /// For long-lived subscriptions, check [`Client::subscribe`].
+    ///
+    /// # Gossip
+    ///
+    /// If `gossip` is enabled the events will be streamed also from
+    /// NIP65 relays (automatically discovered) of public keys included in filters (if any).
+    pub async fn stream_events(&self, filter: &Filter, timeout: Duration) -> Result<EventStream> {
+        let stream = self
+            .inner
+            .stream_events(filter.deref().clone(), timeout)
+            .await?;
+        Ok(stream.into())
+    }
+
+    /// Stream events from specific relays
+    ///
+    /// # Overview
+    ///
+    /// This is an **auto-closing subscription** and will be closed automatically on `EOSE`.
+    /// For long-lived subscriptions, check [`Client::subscribe_to`].
+    pub async fn stream_events_from(
+        &self,
+        urls: Vec<String>,
+        filter: &Filter,
+        timeout: Duration,
+    ) -> Result<EventStream> {
+        let stream = self
+            .inner
+            .stream_events_from(urls, filter.deref().clone(), timeout)
+            .await?;
+        Ok(stream.into())
+    }
+
+    /// Stream events from specific relays with specific filters
+    ///
+    /// # Overview
+    ///
+    /// This is an **auto-closing subscription** and will be closed automatically on `EOSE`.
+    pub async fn stream_events_targeted(
+        &self,
+        targets: HashMap<String, Arc<Filter>>,
+        timeout: Duration,
+    ) -> Result<EventStream> {
+        let mut new_targets: HashMap<RelayUrl, nostr::Filter> =
+            HashMap::with_capacity(targets.len());
+
+        for (url, filter) in targets.into_iter() {
+            let url: RelayUrl = RelayUrl::parse(&url)?;
+            let filter: nostr::Filter = filter.as_ref().deref().clone();
+            new_targets.insert(url, filter);
+        }
+
+        let stream = self
+            .inner
+            .stream_events_targeted(new_targets, timeout)
+            .await?;
+        Ok(stream.into())
+    }
+
     pub async fn send_msg_to(&self, urls: Vec<String>, msg: &ClientMessage) -> Result<Output> {
         Ok(self
             .inner

src/client/mod.rs

@@ -16,6 +16,7 @@ pub mod nwc;
 pub mod policy;
 pub mod protocol;
 pub mod relay;
+pub mod stream;
 pub mod transport;
 mod util;
 

src/lib.rs

@@ -0,0 +1,34 @@
+// Copyright (c) 2022-2023 Yuki Kishimoto
+// Copyright (c) 2023-2025 Rust Nostr Developers
+// Distributed under the MIT software license
+
+use std::sync::Arc;
+
+use futures_util::StreamExt;
+use nostr_sdk::pool::stream::ReceiverStream;
+use tokio::sync::Mutex;
+use uniffi::Object;
+
+use crate::protocol::event::Event;
+
+#[derive(Object)]
+pub struct EventStream {
+    stream: Mutex<ReceiverStream<nostr::Event>>,
+}
+
+impl From<ReceiverStream<nostr::Event>> for EventStream {
+    fn from(stream: ReceiverStream<nostr::Event>) -> Self {
+        Self {
+            stream: Mutex::new(stream),
+        }
+    }
+}
+
+#[uniffi::export(async_runtime = "tokio")]
+impl EventStream {
+    pub async fn next(&self) -> Option<Arc<Event>> {
+        let mut stream = self.stream.lock().await;
+        let event: nostr::Event = stream.next().await?;
+        Some(Arc::new(event.into()))
+    }
+}