Rust Tokio 入门：Stream 与高级流处理：处理数据流的利器

trait Stream {
    type Item;
    fn poll_next(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>>;
}

[dependencies]
tokio = { version = "1", features = ["full"] }
tokio-stream = { version = "0.1", features = ["full"] }  # 开启 time 等特性

use tokio_stream::{self as stream, StreamExt};

#[tokio::main]
async fn main() {
    // 从迭代器创建 Stream
    let numbers = stream::iter(1..=5);

    // 消费 Stream（类似 for 循环）
    let mut s = numbers;
    while let Some(num) = s.next().await {
        println!("收到: {}", num);
    }
}

use tokio::sync::mpsc;
use tokio_stream::wrappers::ReceiverStream;

let (tx, rx) = mpsc::channel(32);
let mut stream = ReceiverStream::new(rx);

// 在其他 Task 中 tx.send(...) 

while let Some(msg) = stream.next().await {
    println!("处理消息: {}", msg);
}

use tokio_stream::{self as stream, StreamExt}; // 别名导入，方便使用 stream::iter
use std::time::Duration;

#[tokio::main]
async fn main() {
    let s = stream::iter(1..=10)
        .filter(|&n| n % 2 == 0)                   // 修正：同步过滤偶数
        .map(|n| n * n)                            // 映射平方
        .take(3)                                   // 取前 3 个
        .timeout(Duration::from_secs(1))           // 超时控制
        .filter_map(|res| {                        // 修正：同步处理 Result
            match res {
                Ok(v) => Some(v),
                Err(_) => None,
            }
        });

    // 提示：collect 需要引入 StreamExt
    let result: Vec<i32> = s.collect().await;
    println!("结果: {:?}", result);  // 输出: [4, 16, 36]
}

use tokio_stream::{self as stream, StreamExt};
use std::time::Duration;

#[tokio::main]
async fn main() {
    let s = stream::iter(1..=10)
        .filter(|&n| n % 2 == 0)
        .map(|n| n * n)
        // 使用 then 处理异步转换（如果内部有 await）
        .then(|n| async move {
            // 模拟异步操作，比如数据库查询
            tokio::time::sleep(Duration::from_millis(10)).await;
            n
        })
        .timeout(Duration::from_secs(1))
        .map(|res| res.ok()) // 将 Result 转为 Option，超时变 None
        .filter_map(|opt| opt); // 过滤掉 None

    let result: Vec<i32> = s.collect().await;
    println!("进阶结果: {:?}", result);
}

use tokio::sync::mpsc;
use tokio::time::{self, Duration};
use tokio_stream::{self as stream, StreamExt, StreamMap};
use tokio_stream::wrappers::{ReceiverStream, IntervalStream};
use std::pin::Pin;

// 专家技巧：定义一个类型别名，用于存放不同类型的流
// Pin<Box<dyn stream::Stream<Item = String> + Send>> 俗称 BoxStream
type BoxStream = Pin<Box<dyn stream::Stream<Item = String> + Send>>;

#[tokio::main]
async fn main() {
    // 1. 构造第一个流：来自 mpsc 通道（模拟用户输入）
    let (tx, rx) = mpsc::channel::<String>(10);
    let client1_stream = ReceiverStream::new(rx);

    // 2. 构造第二个流：来自 Interval 定时器（模拟心跳）
    let interval = time::interval(Duration::from_secs(2));
    let client2_stream = IntervalStream::new(interval)
        .map(|t| format!("心跳滴答: {:?}", t));

    // 3. 创建 StreamMap
    // 重点：由于两个流原始类型不同，需要通过 Box::pin 擦除类型（Type Erasure）
    let mut map = StreamMap::new();
    
    map.insert("User_Task", Box::pin(client1_stream) as BoxStream);
    map.insert("Heartbeat", Box::pin(client2_stream) as BoxStream);

    // 模拟异步发送数据给 client1
    tokio::spawn(async move {
        tx.send("你好，Tokio!".to_string()).await.unwrap();
        time::sleep(Duration::from_secs(1)).await;
        tx.send("来自客户端1的第二条消息".to_string()).await.unwrap();
    });

    println!("--- 开始多流监听 ---");

    // 4. 消费 StreamMap
    // StreamMap 本身实现了 Stream 特性，它会公平地轮询内部所有的流
    while let Some((key, value)) = map.next().await {
        println!("[{}] 接收到数据: {}", key, value);
    }
}

use std::time::Duration;
use tokio::sync::mpsc;
use tokio_stream::{self as stream, StreamExt};

#[tokio::main]
async fn main() {
    // 启动专家级日志流水线
    build_log_pipeline().await;
}
async fn build_log_pipeline() {
    // 1. 模拟数据源 A：来自 mpsc 通道
    let (tx, rx) = mpsc::channel::<String>(100);
    let log_stream_mpsc = tokio_stream::wrappers::ReceiverStream::new(rx);

    // 2. 模拟数据源 B：来自静态迭代器
    let log_stream_static = stream::iter(vec![
        "INFO: system initialized".to_string(),
        "DEBUG: memory check pass".to_string(),
        "INFO: setup complete".to_string(),
    ]);

    // 模拟生产者：异步向通道发送日志
    let tx_clone = tx.clone();
    tokio::spawn(async move {
        let logs = vec!["ERROR: disk full", "INFO: user_1 login", "WARN: high cpu"];
        for log in logs {
            let _ = tx_clone.send(log.to_string()).await;
            // 故意留点间隔，触发 chunks_timeout 的时间阈值
            tokio::time::sleep(Duration::from_millis(200)).await;
        }
        // 生产者完成后，tx 被 drop，rx 流会在处理完剩余数据后关闭
    });

    println!("🚀 日志聚合流水线已启动（使用 while let 模式）...");

    // 3. 构建组合子管道
    // 注意：这里我们不直接 await，而是先定义好管道
    let pipeline = log_stream_mpsc
        .merge(log_stream_static)
        .filter(|log| {
            let is_debug = log.contains("DEBUG");
            if is_debug {
                println!("── [过滤] 跳过 DEBUG 级日志");
            }
            !is_debug
        })
        .map(|log| log.to_uppercase())
        // 每凑够 3 条，或者超过 1 秒没新数据，就打包成 Vec<String> 产出
        .chunks_timeout(3, Duration::from_secs(1));

    // 4. 关键点：使用 tokio::pin! 固定流
    // 复杂的流结构在调用 .next() 时必须被固定在内存中
    tokio::pin!(pipeline);

    // 5. 消费管道：while let 模式比 for_each 更稳健，避开了 Trait 冲突
    while let Some(batch) = pipeline.next().await {
        println!("📦 --- 触发批量处理 ---");
        println!("内容: {:?}", batch);
        println!("批次大小: {}, 触发时间: {:?}", batch.len(), std::time::Instant::now());

        // 模拟异步数据库操作
        // db.insert_batch(batch).await;
    }

    println!("✅ 流水线运行结束，所有日志处理完毕。");
}