如何在Rust中实现过滤升序值的有状态Stream过滤器？

要实现这种带状态的升序过滤器，用futures库中的scan操作符就能直接搞定——它天生用来处理需要维护状态的流转换逻辑，完全不需要额外组合filter。

核心逻辑是用scan携带当前的最大值状态，对每个流元素进行判断：如果元素大于当前最大值，就更新状态并保留该元素；否则直接过滤掉。

示例代码如下：

use futures::{stream::StreamExt, stream::iter};

#[tokio::main]
async fn main() {
    // 初始化测试流
    let source_stream = iter(vec![1, 2, 3, 2, 1, 6]);
    
    // 用scan实现有状态过滤
    let ascending_stream = source_stream.scan(i32::MIN, |current_max, value| async move {
        if value > *current_max {
            *current_max = value;
            Some(value)
        } else {
            None
        }
    });
    
    // 收集结果并打印
    let result: Vec<_> = ascending_stream.collect().await;
    println!("{:?}", result); // 输出: [1, 2, 3, 6]
}

代码说明 #

• 初始状态用i32::MIN而非0，是为了兼容负数开头的流，确保第一个元素无论正负都能正常通过。
• scan的闭包接收可变引用的状态和当前流元素，异步执行判断逻辑：返回Some(value)表示保留该元素，返回None则过滤掉。
• 这种方式完全替代了你设想的filter_stateful逻辑，一步完成状态维护和元素过滤。

内容的提问来源于stack exchange，提问作者Rol