Crossbeam Bounded Channel Pipeline

use crossbeam_channel::{bounded, never, select};
use std::thread;
use std::time::Duration;

fn main() {
    let (raw_tx, raw_rx) = bounded::<String>(8);
    let (parsed_tx, parsed_rx) = bounded::<u64>(8);

    // 阶段一：读取与清洗
    thread::spawn(move || {
        for i in 0..20 {
            let line = format!("metric:{}:{}", i % 5, i * 10);
            if raw_tx.send(line).is_err() {
                break;
            }
            thread::sleep(Duration::from_millis(5));
        }
        drop(raw_tx);
    });

    // 阶段二：解析与转换（select 监听中断信号）
    thread::spawn(move || {
        let (stop_tx, stop_rx) = bounded::<()>(1);
        thread::spawn(move || {
            thread::sleep(Duration::from_millis(200));
            let _ = stop_tx.send(());
        });

        loop {
            select! {
                recv(raw_rx) -> line => match line {
                    Ok(text) => {
                        let value: u64 = text
                            .rsplit(':')
                            .next()
                            .and_then(|v| v.parse().ok())
                            .unwrap_or(0);
                        if parsed_tx.send(value).is_err() {
                            break;
                        }
                    }
                    Err(_) => break,
                },
                recv(stop_rx) -> _ => {
                    println!("interrupt requested, draining remaining input");
                    for leftover in raw_rx.try_iter() {
                        let _ = parsed_tx.send(parse_value(&leftover));
                    }
                    break;
                }
                recv(never()) -> _ => unreachable!(),
            }
        }
        drop(parsed_tx);
    });

    // 阶段三：聚合消费
    let mut total: u64 = 0;
    for value in parsed_rx.iter() {
        total += value;
    }
    println!("pipeline total: {total}");
}

fn parse_value(text: &str) -> u64 {
    text.rsplit(':').next().and_then(|v| v.parse().ok()).unwrap_or(0)
}