use super::plumbing::*; use super::*; /// `Flatten` turns each element to a parallel iterator, then flattens these iterators /// together. This struct is created by the [`flatten()`] method on [`ParallelIterator`]. /// /// [`flatten()`]: trait.ParallelIterator.html#method.flatten /// [`ParallelIterator`]: trait.ParallelIterator.html #[must_use = "iterator adaptors are lazy and do nothing unless consumed"] #[derive(Debug, Clone)] pub struct Flatten { base: I, } impl Flatten where I: ParallelIterator, I::Item: IntoParallelIterator, { /// Creates a new `Flatten` iterator. pub(super) fn new(base: I) -> Self { Flatten { base } } } impl ParallelIterator for Flatten where I: ParallelIterator, I::Item: IntoParallelIterator, { type Item = ::Item; fn drive_unindexed(self, consumer: C) -> C::Result where C: UnindexedConsumer, { let consumer = FlattenConsumer::new(consumer); self.base.drive_unindexed(consumer) } } /// //////////////////////////////////////////////////////////////////////// /// Consumer implementation struct FlattenConsumer { base: C, } impl FlattenConsumer { fn new(base: C) -> Self { FlattenConsumer { base } } } impl Consumer for FlattenConsumer where C: UnindexedConsumer, T: IntoParallelIterator, { type Folder = FlattenFolder; type Reducer = C::Reducer; type Result = C::Result; fn split_at(self, index: usize) -> (Self, Self, C::Reducer) { let (left, right, reducer) = self.base.split_at(index); ( FlattenConsumer::new(left), FlattenConsumer::new(right), reducer, ) } fn into_folder(self) -> Self::Folder { FlattenFolder { base: self.base, previous: None, } } fn full(&self) -> bool { self.base.full() } } impl UnindexedConsumer for FlattenConsumer where C: UnindexedConsumer, T: IntoParallelIterator, { fn split_off_left(&self) -> Self { FlattenConsumer::new(self.base.split_off_left()) } fn to_reducer(&self) -> Self::Reducer { self.base.to_reducer() } } struct FlattenFolder { base: C, previous: Option, } impl Folder for FlattenFolder where C: UnindexedConsumer, T: IntoParallelIterator, { type Result = C::Result; fn consume(self, item: T) -> Self { let par_iter = item.into_par_iter(); let consumer = self.base.split_off_left(); let result = par_iter.drive_unindexed(consumer); let previous = match self.previous { None => Some(result), Some(previous) => { let reducer = self.base.to_reducer(); Some(reducer.reduce(previous, result)) } }; FlattenFolder { base: self.base, previous, } } fn complete(self) -> Self::Result { match self.previous { Some(previous) => previous, None => self.base.into_folder().complete(), } } fn full(&self) -> bool { self.base.full() } }