oyashio

单个生产者、多个消费者流

• 消费者流可以被克隆
• 独立消费
• 数据位于中央弧
• 轮询和 await 风格消费
• 与常规迭代函数兼容

示例

extern crate oyashio;
use oyashio::{Stream,StreamRMerge};

fn main () {
    // stream stealing examples
    let (mut st, mut sr) = Stream::default();
    let mut sr2 = sr.clone();
    for n in (0..10) { st.send(n) }
    let r2 = sr2.take(4).collect::<Vec<&u8>>();
    let r = sr.take(4).collect::<Vec<&u8>>();
    println!("{:?},{:?}",r2,r); //[0, 1, 2, 3],[4, 5, 6, 7]


    // broadcast examples    
    let (mut st, mut sr) = Stream::new_broadcast();

    for n in (0..10) { st.send(n) }

    // regular iterator, may wait for values to be sent
    let r = sr.clone().filter(|x|*x%2==1)
              .take(4) //only wait on up to 4 odd numbers
              .collect::<Vec<&u8>>();
    println!("{:?}",r); //[1, 3, 5, 7]


    let mut vr = vec!();
    for (i,n) in sr.clone().enumerate() { //iterates til break
        vr.push(n);
        assert_eq!(*n,i as u8);
        if i == 9 {break;} //let's break out of this
    }
    println!("{:?}",vr); //[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]

    vr = vec!();
    // polling iterator, never waits
    for n in sr.poll() { vr.push(n); }
    println!("{:?}",vr); //[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]


    //stream merge example
    let (mut st2,mut sr2) = Stream::new_broadcast();
    let (mut st, mut sr) = Stream::new_broadcast();
    let (mut st3, mut sr3) = Stream::new_broadcast();

    for n in (0i32..2) { st.send(n); } //0,1

    let mut vr: Vec<i32> = vec!();        
    let mut vsr = vec![sr,sr2,sr3];

    let sm = StreamRMerge::new(vsr);

    for n in (2..4) { st2.send(n); } //2,3
    for n in (4..6) { st3.send(n); } //4,5

    for n in sm.clone() { vr.push(*n); }

    assert_eq!(vr,&[0,2,4,1,3,5]);
}								

待办事项

• 创建可选的有界大小流
• 使流的消费开始和停止而无需'移动'流变量（例如：stream->filter->take->collect，然后 stream->enumerate() 而无需必要的克隆）
• 在实际场景中尝试