ActivePlan<T1, T2, T3>
namespace System.
Reactive.
Joins
{
internal class ActivePlan<
T1,
T2,
T3> :
ActivePlan
{
private readonly Action<
T1,
T2,
T3>
onNext;
private readonly Action onCompleted;
private readonly JoinObserver<
T1>
first;
private readonly JoinObserver<
T2>
second;
private readonly JoinObserver<
T3>
third;
internal ActivePlan(
JoinObserver<
T1>
first,
JoinObserver<
T2>
second,
JoinObserver<
T3>
third,
Action<
T1,
T2,
T3>
onNext,
Action onCompleted)
{
this.
onNext =
onNext;
this.
onCompleted =
onCompleted;
this.
first =
first;
this.
second =
second;
this.
third =
third;
AddJoinObserver(
first);
AddJoinObserver(
second);
AddJoinObserver(
third);
}
internal override void Match()
{
if (
first.
Queue.
Count >
0 &&
second.
Queue.
Count >
0 &&
third.
Queue.
Count >
0) {
Notification<
T1>
notification =
first.
Queue.
Peek();
Notification<
T2>
notification2 =
second.
Queue.
Peek();
Notification<
T3>
notification3 =
third.
Queue.
Peek();
if (
notification.
Kind ==
NotificationKind.
OnCompleted ||
notification2.
Kind ==
NotificationKind.
OnCompleted ||
notification3.
Kind ==
NotificationKind.
OnCompleted)
onCompleted();
else {
Dequeue();
onNext(
notification.
Value,
notification2.
Value,
notification3.
Value);
}
}
}
}
}