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