rx_core_observable_merge 0.2.1

merge observable for rx_core
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79

use core::{marker::PhantomData, num::NonZero};

use rx_core_common::{
	ErasedObservable, ErasedObservables, Never, NeverMapIntoExtension, Observable,
	PhantomInvariant, RxObserver, SharedSubscription, Signal, Subscriber, TeardownCollection,
	TeardownCollectionExtension, UpgradeableObserver,
};
use rx_core_macro_observable_derive::RxObservable;
use rx_core_subscriber_higher_order_all::HigherOrderAllSubscriber;
use rx_core_subscriber_higher_order_concurrent::ConcurrentSubscriberProvider;

#[derive(RxObservable, Clone)]
#[rx_out(Out)]
#[rx_out_error(OutError)]
pub struct MergeObservable<Out, OutError, const SIZE: usize>
where
	Out: Signal,
	OutError: Signal,
{
	observables: ErasedObservables<Out, OutError, SIZE>,
	concurrency_limit: NonZero<usize>,
	_phantom_data: PhantomInvariant<(Out, OutError)>,
}

impl<Out, OutError, const SIZE: usize> MergeObservable<Out, OutError, SIZE>
where
	Out: Signal,
	OutError: Signal,
{
	pub fn new(
		observables: impl Into<ErasedObservables<Out, OutError, SIZE>>,
		concurrency_limit: usize,
	) -> Self {
		Self {
			observables: observables.into(),
			concurrency_limit: NonZero::new(concurrency_limit).unwrap_or(NonZero::<usize>::MIN),
			_phantom_data: PhantomData,
		}
	}
}

impl<Out, OutError, const SIZE: usize> Observable for MergeObservable<Out, OutError, SIZE>
where
	Out: Signal,
	OutError: Signal,
{
	type Subscription<Destination>
		= SharedSubscription
	where
		Destination: 'static + Subscriber<In = Self::Out, InError = Self::OutError>;

	fn subscribe<Destination>(
		&mut self,
		observer: Destination,
	) -> Self::Subscription<Destination::Upgraded>
	where
		Destination: 'static + UpgradeableObserver<In = Self::Out, InError = Self::OutError>,
	{
		let destination = observer.upgrade();

		let mut concat_subscriber = HigherOrderAllSubscriber::<
			ErasedObservable<Out, OutError>,
			Never,
			ConcurrentSubscriberProvider,
			_,
			<Destination as UpgradeableObserver>::Upgraded,
		>::new(destination, Never::map_into(), self.concurrency_limit);

		for next_observable in self.observables.iter().cloned() {
			concat_subscriber.next(next_observable);
		}
		concat_subscriber.complete();

		let mut subscription = SharedSubscription::default();
		concat_subscriber.add(subscription.clone());
		subscription.add_teardown(concat_subscriber.into());
		subscription
	}
}