| 1 | use super::plumbing::*; |
| 2 | use super::*; |
| 3 | |
| 4 | use std::fmt::{self, Debug}; |
| 5 | |
| 6 | /// `Filter` takes a predicate `filter_op` and filters out elements that match. |
| 7 | /// This struct is created by the [`filter()`] method on [`ParallelIterator`] |
| 8 | /// |
| 9 | /// [`filter()`]: trait.ParallelIterator.html#method.filter |
| 10 | /// [`ParallelIterator`]: trait.ParallelIterator.html |
| 11 | #[must_use = "iterator adaptors are lazy and do nothing unless consumed" ] |
| 12 | #[derive (Clone)] |
| 13 | pub struct Filter<I: ParallelIterator, P> { |
| 14 | base: I, |
| 15 | filter_op: P, |
| 16 | } |
| 17 | |
| 18 | impl<I: ParallelIterator + Debug, P> Debug for Filter<I, P> { |
| 19 | fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { |
| 20 | f.debug_struct("Filter" ).field(name:"base" , &self.base).finish() |
| 21 | } |
| 22 | } |
| 23 | |
| 24 | impl<I, P> Filter<I, P> |
| 25 | where |
| 26 | I: ParallelIterator, |
| 27 | { |
| 28 | /// Creates a new `Filter` iterator. |
| 29 | pub(super) fn new(base: I, filter_op: P) -> Self { |
| 30 | Filter { base, filter_op } |
| 31 | } |
| 32 | } |
| 33 | |
| 34 | impl<I, P> ParallelIterator for Filter<I, P> |
| 35 | where |
| 36 | I: ParallelIterator, |
| 37 | P: Fn(&I::Item) -> bool + Sync + Send, |
| 38 | { |
| 39 | type Item = I::Item; |
| 40 | |
| 41 | fn drive_unindexed<C>(self, consumer: C) -> C::Result |
| 42 | where |
| 43 | C: UnindexedConsumer<Self::Item>, |
| 44 | { |
| 45 | let consumer1: FilterConsumer<'_, C, P> = FilterConsumer::new(base:consumer, &self.filter_op); |
| 46 | self.base.drive_unindexed(consumer:consumer1) |
| 47 | } |
| 48 | } |
| 49 | |
| 50 | /// //////////////////////////////////////////////////////////////////////// |
| 51 | /// Consumer implementation |
| 52 | |
| 53 | struct FilterConsumer<'p, C, P> { |
| 54 | base: C, |
| 55 | filter_op: &'p P, |
| 56 | } |
| 57 | |
| 58 | impl<'p, C, P> FilterConsumer<'p, C, P> { |
| 59 | fn new(base: C, filter_op: &'p P) -> Self { |
| 60 | FilterConsumer { base, filter_op } |
| 61 | } |
| 62 | } |
| 63 | |
| 64 | impl<'p, T, C, P: 'p> Consumer<T> for FilterConsumer<'p, C, P> |
| 65 | where |
| 66 | C: Consumer<T>, |
| 67 | P: Fn(&T) -> bool + Sync, |
| 68 | { |
| 69 | type Folder = FilterFolder<'p, C::Folder, P>; |
| 70 | type Reducer = C::Reducer; |
| 71 | type Result = C::Result; |
| 72 | |
| 73 | fn split_at(self, index: usize) -> (Self, Self, C::Reducer) { |
| 74 | let (left, right, reducer) = self.base.split_at(index); |
| 75 | ( |
| 76 | FilterConsumer::new(left, self.filter_op), |
| 77 | FilterConsumer::new(right, self.filter_op), |
| 78 | reducer, |
| 79 | ) |
| 80 | } |
| 81 | |
| 82 | fn into_folder(self) -> Self::Folder { |
| 83 | FilterFolder { |
| 84 | base: self.base.into_folder(), |
| 85 | filter_op: self.filter_op, |
| 86 | } |
| 87 | } |
| 88 | |
| 89 | fn full(&self) -> bool { |
| 90 | self.base.full() |
| 91 | } |
| 92 | } |
| 93 | |
| 94 | impl<'p, T, C, P: 'p> UnindexedConsumer<T> for FilterConsumer<'p, C, P> |
| 95 | where |
| 96 | C: UnindexedConsumer<T>, |
| 97 | P: Fn(&T) -> bool + Sync, |
| 98 | { |
| 99 | fn split_off_left(&self) -> Self { |
| 100 | FilterConsumer::new(self.base.split_off_left(), self.filter_op) |
| 101 | } |
| 102 | |
| 103 | fn to_reducer(&self) -> Self::Reducer { |
| 104 | self.base.to_reducer() |
| 105 | } |
| 106 | } |
| 107 | |
| 108 | struct FilterFolder<'p, C, P> { |
| 109 | base: C, |
| 110 | filter_op: &'p P, |
| 111 | } |
| 112 | |
| 113 | impl<'p, C, P, T> Folder<T> for FilterFolder<'p, C, P> |
| 114 | where |
| 115 | C: Folder<T>, |
| 116 | P: Fn(&T) -> bool + 'p, |
| 117 | { |
| 118 | type Result = C::Result; |
| 119 | |
| 120 | fn consume(self, item: T) -> Self { |
| 121 | let filter_op = self.filter_op; |
| 122 | if filter_op(&item) { |
| 123 | let base = self.base.consume(item); |
| 124 | FilterFolder { base, filter_op } |
| 125 | } else { |
| 126 | self |
| 127 | } |
| 128 | } |
| 129 | |
| 130 | // This cannot easily specialize `consume_iter` to be better than |
| 131 | // the default, because that requires checking `self.base.full()` |
| 132 | // during a call to `self.base.consume_iter()`. (#632) |
| 133 | |
| 134 | fn complete(self) -> Self::Result { |
| 135 | self.base.complete() |
| 136 | } |
| 137 | |
| 138 | fn full(&self) -> bool { |
| 139 | self.base.full() |
| 140 | } |
| 141 | } |
| 142 | |