finance_query/streaming/
news.rs

1//! Continuous RSS/Atom polling exposed as a `Stream`, for a source that's
2//! inherently pull-only (RSS/Atom has no server push): a background task
3//! polls the configured sources on an interval and broadcasts newly-seen
4//! entries (deduplicated by URL) to every subscriber.
5
6use std::collections::{HashSet, VecDeque};
7use std::pin::Pin;
8use std::task::{Context, Poll};
9use std::time::Duration;
10
11use futures::stream::Stream;
12use tokio::sync::{broadcast, mpsc};
13use tracing::warn;
14
15use super::subscription::Subscription;
16use crate::feeds::{FeedEntry, FeedSource, fetch_all};
17
18/// Default interval between polls of all subscribed sources.
19const DEFAULT_POLL_INTERVAL: Duration = Duration::from_secs(300);
20
21/// Channel capacity for news entries (much lower churn than price ticks).
22const CHANNEL_CAPACITY: usize = 256;
23
24/// Cap on remembered entry URLs used for new-vs-seen dedup, bounding memory
25/// for long-running subscriptions.
26const SEEN_CAP: usize = 2000;
27
28enum FeedCommand {
29    AddSources(Vec<FeedSource>),
30    RemoveSources(Vec<String>),
31    Close,
32}
33
34/// A continuous subscription to one or more RSS/Atom sources.
35///
36/// Polls the configured sources on an interval (5 minutes by default,
37/// configurable via [`NewsStreamBuilder::poll_interval`]) and yields entries
38/// as they're first seen: an initial batch on subscribe, then only new items
39/// on each subsequent poll. Backed by a broadcast channel, so
40/// [`resubscribe`](Self::resubscribe) supports multiple independent
41/// consumers of the same subscription.
42///
43/// # Example
44///
45/// ```no_run
46/// use finance_query::streaming::NewsStream;
47/// use finance_query::feeds::FeedSource;
48/// use futures::StreamExt;
49///
50/// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
51/// let mut stream =
52///     NewsStream::subscribe([FeedSource::Bloomberg, FeedSource::MarketWatch]).await;
53///
54/// while let Some(entry) = stream.next().await {
55///     println!("[{}] {}", entry.source, entry.title);
56/// }
57/// # Ok(())
58/// # }
59/// ```
60pub struct NewsStream {
61    inner: Subscription<FeedEntry, FeedCommand>,
62}
63
64impl NewsStream {
65    /// Subscribe to a continuous stream of new entries from the given
66    /// sources, polling every 5 minutes. Use [`NewsStreamBuilder`] to
67    /// customize the poll interval.
68    pub async fn subscribe(sources: impl IntoIterator<Item = FeedSource>) -> Self {
69        NewsStreamBuilder::new().sources(sources).build().await
70    }
71
72    async fn start(sources: Vec<FeedSource>, poll_interval: Duration) -> Self {
73        let inner = Subscription::start(CHANNEL_CAPACITY, 32, move |broadcast_tx, command_rx| {
74            run_feed_loop(sources, poll_interval, broadcast_tx, command_rx)
75        });
76        NewsStream { inner }
77    }
78
79    /// Create a new receiver for this stream.
80    ///
81    /// Useful when you need multiple consumers of the same news subscription.
82    pub fn resubscribe(&self) -> Self {
83        NewsStream {
84            inner: self.inner.resubscribe(),
85        }
86    }
87
88    /// Add more sources to the subscription.
89    ///
90    /// # Example
91    ///
92    /// ```no_run
93    /// use finance_query::streaming::NewsStream;
94    /// use finance_query::feeds::FeedSource;
95    ///
96    /// # async fn example() {
97    /// let stream = NewsStream::subscribe([FeedSource::Bloomberg]).await;
98    /// stream.add_sources([FeedSource::MarketWatch, FeedSource::WsjMarkets]).await;
99    /// # }
100    /// ```
101    pub async fn add_sources(&self, sources: impl IntoIterator<Item = FeedSource>) {
102        self.inner
103            .send(FeedCommand::AddSources(sources.into_iter().collect()))
104            .await;
105    }
106
107    /// Remove sources from the subscription (matched by [`FeedSource::url`]).
108    ///
109    /// # Example
110    ///
111    /// ```no_run
112    /// use finance_query::streaming::NewsStream;
113    /// use finance_query::feeds::FeedSource;
114    ///
115    /// # async fn example() {
116    /// let stream = NewsStream::subscribe([FeedSource::Bloomberg, FeedSource::MarketWatch]).await;
117    /// stream.remove_sources([FeedSource::MarketWatch]).await;
118    /// # }
119    /// ```
120    pub async fn remove_sources(&self, sources: impl IntoIterator<Item = FeedSource>) {
121        let urls = sources.into_iter().map(|s| s.url()).collect();
122        self.inner.send(FeedCommand::RemoveSources(urls)).await;
123    }
124
125    /// Stop polling and close the stream.
126    pub async fn close(&self) {
127        self.inner.send(FeedCommand::Close).await;
128    }
129}
130
131impl Stream for NewsStream {
132    type Item = FeedEntry;
133
134    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
135        Pin::new(&mut self.inner).poll_next(cx)
136    }
137}
138
139/// Builder for creating a [`NewsStream`] with custom configuration.
140pub struct NewsStreamBuilder {
141    sources: Vec<FeedSource>,
142    poll_interval: Duration,
143}
144
145impl NewsStreamBuilder {
146    /// Create a new builder with no sources and the default 5-minute poll interval.
147    pub fn new() -> Self {
148        Self {
149            sources: Vec::new(),
150            poll_interval: DEFAULT_POLL_INTERVAL,
151        }
152    }
153
154    /// Add sources to subscribe to.
155    pub fn sources(mut self, sources: impl IntoIterator<Item = FeedSource>) -> Self {
156        self.sources.extend(sources);
157        self
158    }
159
160    /// Set the interval between polls of all subscribed sources (default: 5 minutes).
161    pub fn poll_interval(mut self, interval: Duration) -> Self {
162        self.poll_interval = interval;
163        self
164    }
165
166    /// Start the news stream.
167    pub async fn build(self) -> NewsStream {
168        NewsStream::start(self.sources, self.poll_interval).await
169    }
170}
171
172impl Default for NewsStreamBuilder {
173    fn default() -> Self {
174        Self::new()
175    }
176}
177
178async fn run_feed_loop(
179    initial_sources: Vec<FeedSource>,
180    poll_interval: Duration,
181    broadcast_tx: broadcast::Sender<FeedEntry>,
182    mut command_rx: mpsc::Receiver<FeedCommand>,
183) {
184    let mut sources: Vec<(String, FeedSource)> =
185        initial_sources.into_iter().map(|s| (s.url(), s)).collect();
186    let mut seen: HashSet<String> = HashSet::new();
187    let mut seen_order: VecDeque<String> = VecDeque::new();
188
189    let mut ticker = tokio::time::interval(poll_interval);
190    ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
191
192    loop {
193        tokio::select! {
194            _ = ticker.tick() => {
195                if sources.is_empty() {
196                    continue;
197                }
198                let batch = sources.iter().map(|(_, s)| s.clone());
199                match fetch_all(batch).await {
200                    Ok(entries) => {
201                        for entry in entries {
202                            if seen.insert(entry.url.clone()) {
203                                seen_order.push_back(entry.url.clone());
204                                if seen_order.len() > SEEN_CAP
205                                    && let Some(old) = seen_order.pop_front()
206                                {
207                                    seen.remove(&old);
208                                }
209                                let _ = broadcast_tx.send(entry);
210                            }
211                        }
212                    }
213                    Err(e) => warn!("news stream poll failed: {e}"),
214                }
215            }
216            cmd = command_rx.recv() => {
217                match cmd {
218                    Some(FeedCommand::AddSources(new_sources)) => {
219                        for s in new_sources {
220                            let url = s.url();
221                            if let Some(existing) = sources.iter_mut().find(|(u, _)| *u == url) {
222                                existing.1 = s;
223                            } else {
224                                sources.push((url, s));
225                            }
226                        }
227                    }
228                    Some(FeedCommand::RemoveSources(urls)) => {
229                        sources.retain(|(u, _)| !urls.contains(u));
230                    }
231                    Some(FeedCommand::Close) | None => break,
232                }
233            }
234        }
235    }
236}
237
238#[cfg(test)]
239mod tests {
240    use super::*;
241    use futures::StreamExt;
242
243    #[tokio::test]
244    async fn add_and_remove_sources_are_accepted() {
245        // No network: sources list is empty so the poll tick is a no-op;
246        // this only exercises the command channel plumbing.
247        let stream = NewsStream::subscribe([]).await;
248        stream.add_sources([FeedSource::Bloomberg]).await;
249        stream.remove_sources([FeedSource::Bloomberg]).await;
250        stream.close().await;
251    }
252
253    #[tokio::test]
254    async fn resubscribe_gives_an_independent_receiver() {
255        let stream = NewsStream::subscribe([]).await;
256        let other = stream.resubscribe();
257        drop(other);
258        stream.close().await;
259    }
260
261    #[tokio::test]
262    async fn close_ends_the_stream() {
263        let mut stream = NewsStreamBuilder::new()
264            .poll_interval(Duration::from_millis(20))
265            .build()
266            .await;
267        stream.close().await;
268        let timeout = tokio::time::timeout(Duration::from_secs(2), stream.next()).await;
269        assert!(matches!(timeout, Ok(None)));
270    }
271}
finance_query/streaming/news.rs

finance_query/streaming/
news.rs
