| 1 | use super::plumbing::*; |
| 2 | use super::*; |
| 3 | |
| 4 | use std::fmt::{self, Debug}; |
| 5 | |
| 6 | /// `Update` is an iterator that mutates the elements of an |
| 7 | /// underlying iterator before they are yielded. |
| 8 | /// |
| 9 | /// This struct is created by the [`update()`] method on [`ParallelIterator`] |
| 10 | /// |
| 11 | /// [`update()`]: trait.ParallelIterator.html#method.update |
| 12 | /// [`ParallelIterator`]: trait.ParallelIterator.html |
| 13 | #[must_use = "iterator adaptors are lazy and do nothing unless consumed" ] |
| 14 | #[derive (Clone)] |
| 15 | pub struct Update<I: ParallelIterator, F> { |
| 16 | base: I, |
| 17 | update_op: F, |
| 18 | } |
| 19 | |
| 20 | impl<I: ParallelIterator + Debug, F> Debug for Update<I, F> { |
| 21 | fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { |
| 22 | f.debug_struct("Update" ).field(name:"base" , &self.base).finish() |
| 23 | } |
| 24 | } |
| 25 | |
| 26 | impl<I, F> Update<I, F> |
| 27 | where |
| 28 | I: ParallelIterator, |
| 29 | { |
| 30 | /// Creates a new `Update` iterator. |
| 31 | pub(super) fn new(base: I, update_op: F) -> Self { |
| 32 | Update { base, update_op } |
| 33 | } |
| 34 | } |
| 35 | |
| 36 | impl<I, F> ParallelIterator for Update<I, F> |
| 37 | where |
| 38 | I: ParallelIterator, |
| 39 | F: Fn(&mut I::Item) + Send + Sync, |
| 40 | { |
| 41 | type Item = I::Item; |
| 42 | |
| 43 | fn drive_unindexed<C>(self, consumer: C) -> C::Result |
| 44 | where |
| 45 | C: UnindexedConsumer<Self::Item>, |
| 46 | { |
| 47 | let consumer1: UpdateConsumer<'_, C, F> = UpdateConsumer::new(base:consumer, &self.update_op); |
| 48 | self.base.drive_unindexed(consumer:consumer1) |
| 49 | } |
| 50 | |
| 51 | fn opt_len(&self) -> Option<usize> { |
| 52 | self.base.opt_len() |
| 53 | } |
| 54 | } |
| 55 | |
| 56 | impl<I, F> IndexedParallelIterator for Update<I, F> |
| 57 | where |
| 58 | I: IndexedParallelIterator, |
| 59 | F: Fn(&mut I::Item) + Send + Sync, |
| 60 | { |
| 61 | fn drive<C>(self, consumer: C) -> C::Result |
| 62 | where |
| 63 | C: Consumer<Self::Item>, |
| 64 | { |
| 65 | let consumer1 = UpdateConsumer::new(consumer, &self.update_op); |
| 66 | self.base.drive(consumer1) |
| 67 | } |
| 68 | |
| 69 | fn len(&self) -> usize { |
| 70 | self.base.len() |
| 71 | } |
| 72 | |
| 73 | fn with_producer<CB>(self, callback: CB) -> CB::Output |
| 74 | where |
| 75 | CB: ProducerCallback<Self::Item>, |
| 76 | { |
| 77 | return self.base.with_producer(Callback { |
| 78 | callback, |
| 79 | update_op: self.update_op, |
| 80 | }); |
| 81 | |
| 82 | struct Callback<CB, F> { |
| 83 | callback: CB, |
| 84 | update_op: F, |
| 85 | } |
| 86 | |
| 87 | impl<T, F, CB> ProducerCallback<T> for Callback<CB, F> |
| 88 | where |
| 89 | CB: ProducerCallback<T>, |
| 90 | F: Fn(&mut T) + Send + Sync, |
| 91 | { |
| 92 | type Output = CB::Output; |
| 93 | |
| 94 | fn callback<P>(self, base: P) -> CB::Output |
| 95 | where |
| 96 | P: Producer<Item = T>, |
| 97 | { |
| 98 | let producer = UpdateProducer { |
| 99 | base, |
| 100 | update_op: &self.update_op, |
| 101 | }; |
| 102 | self.callback.callback(producer) |
| 103 | } |
| 104 | } |
| 105 | } |
| 106 | } |
| 107 | |
| 108 | /// //////////////////////////////////////////////////////////////////////// |
| 109 | |
| 110 | struct UpdateProducer<'f, P, F> { |
| 111 | base: P, |
| 112 | update_op: &'f F, |
| 113 | } |
| 114 | |
| 115 | impl<'f, P, F> Producer for UpdateProducer<'f, P, F> |
| 116 | where |
| 117 | P: Producer, |
| 118 | F: Fn(&mut P::Item) + Send + Sync, |
| 119 | { |
| 120 | type Item = P::Item; |
| 121 | type IntoIter = UpdateSeq<P::IntoIter, &'f F>; |
| 122 | |
| 123 | fn into_iter(self) -> Self::IntoIter { |
| 124 | UpdateSeq { |
| 125 | base: self.base.into_iter(), |
| 126 | update_op: self.update_op, |
| 127 | } |
| 128 | } |
| 129 | |
| 130 | fn min_len(&self) -> usize { |
| 131 | self.base.min_len() |
| 132 | } |
| 133 | fn max_len(&self) -> usize { |
| 134 | self.base.max_len() |
| 135 | } |
| 136 | |
| 137 | fn split_at(self, index: usize) -> (Self, Self) { |
| 138 | let (left, right) = self.base.split_at(index); |
| 139 | ( |
| 140 | UpdateProducer { |
| 141 | base: left, |
| 142 | update_op: self.update_op, |
| 143 | }, |
| 144 | UpdateProducer { |
| 145 | base: right, |
| 146 | update_op: self.update_op, |
| 147 | }, |
| 148 | ) |
| 149 | } |
| 150 | |
| 151 | fn fold_with<G>(self, folder: G) -> G |
| 152 | where |
| 153 | G: Folder<Self::Item>, |
| 154 | { |
| 155 | let folder1 = UpdateFolder { |
| 156 | base: folder, |
| 157 | update_op: self.update_op, |
| 158 | }; |
| 159 | self.base.fold_with(folder1).base |
| 160 | } |
| 161 | } |
| 162 | |
| 163 | /// //////////////////////////////////////////////////////////////////////// |
| 164 | /// Consumer implementation |
| 165 | |
| 166 | struct UpdateConsumer<'f, C, F> { |
| 167 | base: C, |
| 168 | update_op: &'f F, |
| 169 | } |
| 170 | |
| 171 | impl<'f, C, F> UpdateConsumer<'f, C, F> { |
| 172 | fn new(base: C, update_op: &'f F) -> Self { |
| 173 | UpdateConsumer { base, update_op } |
| 174 | } |
| 175 | } |
| 176 | |
| 177 | impl<'f, T, C, F> Consumer<T> for UpdateConsumer<'f, C, F> |
| 178 | where |
| 179 | C: Consumer<T>, |
| 180 | F: Fn(&mut T) + Send + Sync, |
| 181 | { |
| 182 | type Folder = UpdateFolder<'f, C::Folder, F>; |
| 183 | type Reducer = C::Reducer; |
| 184 | type Result = C::Result; |
| 185 | |
| 186 | fn split_at(self, index: usize) -> (Self, Self, Self::Reducer) { |
| 187 | let (left, right, reducer) = self.base.split_at(index); |
| 188 | ( |
| 189 | UpdateConsumer::new(left, self.update_op), |
| 190 | UpdateConsumer::new(right, self.update_op), |
| 191 | reducer, |
| 192 | ) |
| 193 | } |
| 194 | |
| 195 | fn into_folder(self) -> Self::Folder { |
| 196 | UpdateFolder { |
| 197 | base: self.base.into_folder(), |
| 198 | update_op: self.update_op, |
| 199 | } |
| 200 | } |
| 201 | |
| 202 | fn full(&self) -> bool { |
| 203 | self.base.full() |
| 204 | } |
| 205 | } |
| 206 | |
| 207 | impl<'f, T, C, F> UnindexedConsumer<T> for UpdateConsumer<'f, C, F> |
| 208 | where |
| 209 | C: UnindexedConsumer<T>, |
| 210 | F: Fn(&mut T) + Send + Sync, |
| 211 | { |
| 212 | fn split_off_left(&self) -> Self { |
| 213 | UpdateConsumer::new(self.base.split_off_left(), self.update_op) |
| 214 | } |
| 215 | |
| 216 | fn to_reducer(&self) -> Self::Reducer { |
| 217 | self.base.to_reducer() |
| 218 | } |
| 219 | } |
| 220 | |
| 221 | struct UpdateFolder<'f, C, F> { |
| 222 | base: C, |
| 223 | update_op: &'f F, |
| 224 | } |
| 225 | |
| 226 | fn apply<T>(update_op: impl Fn(&mut T)) -> impl Fn(T) -> T { |
| 227 | move |mut item: T| { |
| 228 | update_op(&mut item); |
| 229 | item |
| 230 | } |
| 231 | } |
| 232 | |
| 233 | impl<'f, T, C, F> Folder<T> for UpdateFolder<'f, C, F> |
| 234 | where |
| 235 | C: Folder<T>, |
| 236 | F: Fn(&mut T), |
| 237 | { |
| 238 | type Result = C::Result; |
| 239 | |
| 240 | fn consume(self, mut item: T) -> Self { |
| 241 | (self.update_op)(&mut item); |
| 242 | |
| 243 | UpdateFolder { |
| 244 | base: self.base.consume(item), |
| 245 | update_op: self.update_op, |
| 246 | } |
| 247 | } |
| 248 | |
| 249 | fn consume_iter<I>(mut self, iter: I) -> Self |
| 250 | where |
| 251 | I: IntoIterator<Item = T>, |
| 252 | { |
| 253 | let update_op = self.update_op; |
| 254 | self.base = self |
| 255 | .base |
| 256 | .consume_iter(iter.into_iter().map(apply(update_op))); |
| 257 | self |
| 258 | } |
| 259 | |
| 260 | fn complete(self) -> C::Result { |
| 261 | self.base.complete() |
| 262 | } |
| 263 | |
| 264 | fn full(&self) -> bool { |
| 265 | self.base.full() |
| 266 | } |
| 267 | } |
| 268 | |
| 269 | /// Standard Update adaptor, based on `itertools::adaptors::Update` |
| 270 | #[must_use = "iterator adaptors are lazy and do nothing unless consumed" ] |
| 271 | #[derive (Debug, Clone)] |
| 272 | struct UpdateSeq<I, F> { |
| 273 | base: I, |
| 274 | update_op: F, |
| 275 | } |
| 276 | |
| 277 | impl<I, F> Iterator for UpdateSeq<I, F> |
| 278 | where |
| 279 | I: Iterator, |
| 280 | F: Fn(&mut I::Item), |
| 281 | { |
| 282 | type Item = I::Item; |
| 283 | |
| 284 | fn next(&mut self) -> Option<Self::Item> { |
| 285 | let mut v = self.base.next()?; |
| 286 | (self.update_op)(&mut v); |
| 287 | Some(v) |
| 288 | } |
| 289 | |
| 290 | fn size_hint(&self) -> (usize, Option<usize>) { |
| 291 | self.base.size_hint() |
| 292 | } |
| 293 | |
| 294 | fn fold<Acc, G>(self, init: Acc, g: G) -> Acc |
| 295 | where |
| 296 | G: FnMut(Acc, Self::Item) -> Acc, |
| 297 | { |
| 298 | self.base.map(apply(self.update_op)).fold(init, g) |
| 299 | } |
| 300 | |
| 301 | // if possible, re-use inner iterator specializations in collect |
| 302 | fn collect<C>(self) -> C |
| 303 | where |
| 304 | C: ::std::iter::FromIterator<Self::Item>, |
| 305 | { |
| 306 | self.base.map(apply(self.update_op)).collect() |
| 307 | } |
| 308 | } |
| 309 | |
| 310 | impl<I, F> ExactSizeIterator for UpdateSeq<I, F> |
| 311 | where |
| 312 | I: ExactSizeIterator, |
| 313 | F: Fn(&mut I::Item), |
| 314 | { |
| 315 | } |
| 316 | |
| 317 | impl<I, F> DoubleEndedIterator for UpdateSeq<I, F> |
| 318 | where |
| 319 | I: DoubleEndedIterator, |
| 320 | F: Fn(&mut I::Item), |
| 321 | { |
| 322 | fn next_back(&mut self) -> Option<Self::Item> { |
| 323 | let mut v: ::Item = self.base.next_back()?; |
| 324 | (self.update_op)(&mut v); |
| 325 | Some(v) |
| 326 | } |
| 327 | } |
| 328 | |