pipeline.rs

extern crate rayon;

use rayon::prelude::*;
use rayon::iter::internal::*;

pub fn pipeline<I>(iter: I) -> Pipeline<I>
    where I: Iterator + Send, I::Item: Send,
{
    Pipeline { iter: iter }
}

pub struct Pipeline<I>
    where I: Iterator + Send, I::Item: Send,
{
    iter: I,
}

impl<I> ParallelIterator for Pipeline<I>
    where I: Iterator + Send, I::Item: Send,
{
    type Item = I::Item;

    fn drive_unindexed<C>(self, consumer: C) -> C::Result where C: UnindexedConsumer<Self::Item> {
        drive(self.iter, consumer)
    }
}

fn drive<I, C>(mut iter: I, consumer: C) -> C::Result
    where I: Iterator + Send, I::Item: Send, C: Consumer<I::Item>
{
    if let Some(item) = iter.next() {
        let (left, right, reducer) = consumer.split_at(1);
        let (r1, r2) = rayon::join(
            || left.into_folder().consume(item).complete(),
            || drive(iter, right));
        reducer.reduce(r1, r2)
    } else {
        consumer.into_folder().complete()
    }
}

pub fn main() {
    let x = vec![1, 2, 3, 4, 5];
    let y: Vec<u32> = pipeline(x.iter()).map(|&i| i + 1).collect();
    println!("x={:?}", x);
    println!("y={:?}", y);
}