From e5073165f05ab97156c6918afaaf17d7a5ceff87 Mon Sep 17 00:00:00 2001 From: Jason Volk Date: Mon, 24 Nov 2025 17:05:18 +0000 Subject: [PATCH] Cleanup TryIgnore/TryExpect stream extensions related. Signed-off-by: Jason Volk --- src/core/utils/future/mod.rs | 14 ++++++++------ src/core/utils/stream/expect.rs | 21 ++++++++++++-------- src/core/utils/stream/ignore.rs | 29 ++++++++++++++-------------- src/core/utils/stream/mod.rs | 34 +++++++++++++++++---------------- 4 files changed, 54 insertions(+), 44 deletions(-) diff --git a/src/core/utils/future/mod.rs b/src/core/utils/future/mod.rs index c80ccdd6..073b2f0b 100644 --- a/src/core/utils/future/mod.rs +++ b/src/core/utils/future/mod.rs @@ -5,9 +5,11 @@ mod option_stream; mod ready_eq_ext; mod try_ext_ext; -pub use bool_ext::{BoolExt, and, and4, and5, and6, and7, or}; -pub use ext_ext::ExtExt; -pub use option_ext::OptionExt; -pub use option_stream::OptionStream; -pub use ready_eq_ext::ReadyEqExt; -pub use try_ext_ext::TryExtExt; +pub use self::{ + bool_ext::{BoolExt, and, and4, and5, and6, and7, or}, + ext_ext::ExtExt, + option_ext::OptionExt, + option_stream::OptionStream, + ready_eq_ext::ReadyEqExt, + try_ext_ext::TryExtExt, +}; diff --git a/src/core/utils/stream/expect.rs b/src/core/utils/stream/expect.rs index ec572714..92fe6a0b 100644 --- a/src/core/utils/stream/expect.rs +++ b/src/core/utils/stream/expect.rs @@ -2,25 +2,30 @@ use futures::{Stream, StreamExt, TryStream}; use crate::Result; -pub trait TryExpect<'a, Item> { - fn expect_ok(self) -> impl Stream + Send + 'a; +pub trait TryExpect +where + Item: Send, + Self: Send + Sized, +{ + fn expect_ok(self) -> impl Stream + Send; - fn map_expect(self, msg: &'a str) -> impl Stream + Send + 'a; + fn map_expect(self, msg: &str) -> impl Stream + Send; } -impl<'a, T, Item> TryExpect<'a, Item> for T +impl TryExpect for S where - T: Stream> + Send + TryStream + 'a, - Item: 'a, + S: Stream> + Send + TryStream, + Item: Send, + Self: Send + Sized, { #[inline] - fn expect_ok(self: T) -> impl Stream + Send + 'a { + fn expect_ok(self: S) -> impl Stream + Send { self.map_expect("stream expectation failure") } //TODO: move to impl MapExpect #[inline] - fn map_expect(self, msg: &'a str) -> impl Stream + Send + 'a { + fn map_expect(self, msg: &str) -> impl Stream + Send { self.map(|res| res.expect(msg)) } } diff --git a/src/core/utils/stream/ignore.rs b/src/core/utils/stream/ignore.rs index 37c89d9a..1a94f709 100644 --- a/src/core/utils/stream/ignore.rs +++ b/src/core/utils/stream/ignore.rs @@ -1,34 +1,35 @@ use futures::{Stream, StreamExt, TryStream, future::ready}; -use crate::{Error, Result}; +use crate::{Error, Result, utils::stream::TryExpect}; -pub trait TryIgnore<'a, Item> { - fn ignore_err(self) -> impl Stream + Send + 'a; +pub trait TryIgnore +where + Item: Send, + Self: Send + Sized, +{ + fn ignore_err(self) -> impl Stream + Send; - fn ignore_ok(self) -> impl Stream + Send + 'a; + fn ignore_ok(self) -> impl Stream + Send; } -impl<'a, T, Item> TryIgnore<'a, Item> for T +impl TryIgnore for S where - T: Stream> + TryStream + Send + 'a, - Item: Send + 'a, + S: Stream> + Send + TryStream + TryExpect, + Item: Send, + Self: Send + Sized, { #[cfg(debug_assertions)] #[inline] - fn ignore_err(self: T) -> impl Stream + Send + 'a { - use super::TryExpect; - - self.expect_ok() - } + fn ignore_err(self: S) -> impl Stream + Send { self.expect_ok() } #[cfg(not(debug_assertions))] #[inline] - fn ignore_err(self: T) -> impl Stream + Send + 'a { + fn ignore_err(self: S) -> impl Stream + Send { self.filter_map(|res| ready(res.ok())) } #[inline] - fn ignore_ok(self: T) -> impl Stream + Send + 'a { + fn ignore_ok(self: S) -> impl Stream + Send { self.filter_map(|res| ready(res.err())) } } diff --git a/src/core/utils/stream/mod.rs b/src/core/utils/stream/mod.rs index a356f05f..9eb4e64f 100644 --- a/src/core/utils/stream/mod.rs +++ b/src/core/utils/stream/mod.rs @@ -13,20 +13,22 @@ mod try_tools; mod try_wideband; mod wideband; -pub use band::{ - AMPLIFICATION_LIMIT, WIDTH_LIMIT, automatic_amplification, automatic_width, - set_amplification, set_width, +pub use self::{ + band::{ + AMPLIFICATION_LIMIT, WIDTH_LIMIT, automatic_amplification, automatic_width, + set_amplification, set_width, + }, + broadband::BroadbandExt, + cloned::Cloned, + expect::TryExpect, + ignore::TryIgnore, + iter_stream::IterStream, + ready::ReadyExt, + tools::Tools, + try_broadband::TryBroadbandExt, + try_parallel::TryParallelExt, + try_ready::TryReadyExt, + try_tools::TryTools, + try_wideband::TryWidebandExt, + wideband::WidebandExt, }; -pub use broadband::BroadbandExt; -pub use cloned::Cloned; -pub use expect::TryExpect; -pub use ignore::TryIgnore; -pub use iter_stream::IterStream; -pub use ready::ReadyExt; -pub use tools::Tools; -pub use try_broadband::TryBroadbandExt; -pub use try_parallel::TryParallelExt; -pub use try_ready::TryReadyExt; -pub use try_tools::TryTools; -pub use try_wideband::TryWidebandExt; -pub use wideband::WidebandExt;