Leverage new AsyncFn traits in stream module

This commit is contained in:
Héctor Ramón Jiménez 2025-02-21 01:48:09 +01:00
parent c12beecd38
commit 0f9934b1a7
No known key found for this signature in database
GPG key ID: 7CC46565708259A7
2 changed files with 7 additions and 13 deletions

View file

@ -8,13 +8,10 @@ use futures::stream::{self, Stream, StreamExt};
/// This is a more ergonomic [`stream::unfold`], which allows you to go /// This is a more ergonomic [`stream::unfold`], which allows you to go
/// from the "world of futures" to the "world of streams" by simply looping /// from the "world of futures" to the "world of streams" by simply looping
/// and publishing to an async channel from inside a [`Future`]. /// and publishing to an async channel from inside a [`Future`].
pub fn channel<T, F>( pub fn channel<T>(
size: usize, size: usize,
f: impl FnOnce(mpsc::Sender<T>) -> F, f: impl AsyncFnOnce(mpsc::Sender<T>),
) -> impl Stream<Item = T> ) -> impl Stream<Item = T> {
where
F: Future<Output = ()>,
{
let (sender, receiver) = mpsc::channel(size); let (sender, receiver) = mpsc::channel(size);
let runner = stream::once(f(sender)).filter_map(|_| async { None }); let runner = stream::once(f(sender)).filter_map(|_| async { None });
@ -24,13 +21,10 @@ where
/// Creates a new [`Stream`] that produces the items sent from a [`Future`] /// Creates a new [`Stream`] that produces the items sent from a [`Future`]
/// that can fail to the [`mpsc::Sender`] provided to the closure. /// that can fail to the [`mpsc::Sender`] provided to the closure.
pub fn try_channel<T, E, F>( pub fn try_channel<T, E>(
size: usize, size: usize,
f: impl FnOnce(mpsc::Sender<T>) -> F, f: impl AsyncFnOnce(mpsc::Sender<T>) -> Result<(), E>,
) -> impl Stream<Item = Result<T, E>> ) -> impl Stream<Item = Result<T, E>> {
where
F: Future<Output = Result<(), E>>,
{
let (sender, receiver) = mpsc::channel(size); let (sender, receiver) = mpsc::channel(size);
let runner = stream::once(f(sender)).filter_map(|result| async { let runner = stream::once(f(sender)).filter_map(|result| async {

View file

@ -161,7 +161,7 @@ impl<T> Subscription<T> {
/// } /// }
/// ///
/// fn some_worker() -> impl Stream<Item = Event> { /// fn some_worker() -> impl Stream<Item = Event> {
/// stream::channel(100, |mut output| async move { /// stream::channel(100, async |mut output| {
/// // Create channel /// // Create channel
/// let (sender, mut receiver) = mpsc::channel(100); /// let (sender, mut receiver) = mpsc::channel(100);
/// ///