amadeus_core/par_sink/
combine.rs

1#![allow(clippy::option_if_let_else)]
2
3use derive_new::new;
4use educe::Educe;
5use serde::{Deserialize, Serialize};
6use serde_closure::traits::FnMut;
7use std::marker::PhantomData;
8
9use super::{combiner_par_sink, FolderSync, FolderSyncReducer, ParallelPipe, ParallelSink};
10
11#[derive(Educe, Serialize, Deserialize, new)]
12#[educe(Clone(bound = "F: Clone"))]
13#[serde(
14	bound(serialize = "F: Serialize"),
15	bound(deserialize = "F: Deserialize<'de>")
16)]
17pub struct ReduceFn<F, A>(F, PhantomData<fn() -> A>);
18impl<F, A, Item> FolderSync<Item> for ReduceFn<F, A>
19where
20	F: FnMut<(A, A), Output = A>,
21	Item: Into<Option<A>>,
22{
23	type State = Option<A>;
24	type Done = Self::State;
25
26	fn zero(&mut self) -> Self::State {
27		None
28	}
29	fn push(&mut self, state: &mut Self::State, item: Item) {
30		if let Some(item) = item.into() {
31			*state = Some(if let Some(state) = state.take() {
32				self.0.call_mut((state, item))
33			} else {
34				item
35			});
36		}
37	}
38	fn done(&mut self, state: Self::State) -> Self::Done {
39		state
40	}
41}
42
43#[derive(new)]
44#[must_use]
45pub struct Combine<P, F> {
46	pipe: P,
47	f: F,
48}
49
50impl_par_dist! {
51	impl<P: ParallelPipe<Item>, Item, F> ParallelSink<Item> for Combine<P, F>
52	where
53		F: FnMut<(P::Output, P::Output), Output = P::Output> + Clone + Send + 'static,
54		P::Output: Send + 'static,
55	{
56		combiner_par_sink!(ReduceFn<F, P::Output>, self, ReduceFn::new(self.f));
57	}
58}
amadeus_core/par_sink/combine.rs

amadeus_core/par_sink/
combine.rs
