Zip<T1, T2, T3, T4, T5, T6, T7, TResult>
sealed class Zip<T1, T2, T3, T4, T5, T6, T7, TResult> : Producer<TResult, _<T1, T2, T3, T4, T5, T6, T7, TResult>>
using System.Reactive.Disposables;
namespace System.Reactive.Linq.ObservableImpl
{
internal sealed class Zip<T1, T2, T3, T4, T5, T6, T7, TResult> : Producer<TResult, Zip<T1, T2, T3, T4, T5, T6, T7, TResult>._>
{
internal sealed class _ : ZipSink<TResult>
{
private readonly Func<T1, T2, T3, T4, T5, T6, T7, TResult> _resultSelector;
private ZipObserver<T1> _observer1;
private ZipObserver<T2> _observer2;
private ZipObserver<T3> _observer3;
private ZipObserver<T4> _observer4;
private ZipObserver<T5> _observer5;
private ZipObserver<T6> _observer6;
private ZipObserver<T7> _observer7;
public _(Func<T1, T2, T3, T4, T5, T6, T7, TResult> resultSelector, IObserver<TResult> observer, IDisposable cancel)
: base(7, observer, cancel)
{
_resultSelector = resultSelector;
}
public IDisposable Run(IObservable<T1> source1, IObservable<T2> source2, IObservable<T3> source3, IObservable<T4> source4, IObservable<T5> source5, IObservable<T6> source6, IObservable<T7> source7)
{
IDisposable[] obj = new IDisposable[8];
SingleAssignmentDisposable singleAssignmentDisposable = (SingleAssignmentDisposable)(obj[0] = new SingleAssignmentDisposable());
_observer1 = new ZipObserver<T1>(_gate, this, 0, singleAssignmentDisposable);
base.Queues[0] = _observer1.Values;
SingleAssignmentDisposable singleAssignmentDisposable2 = (SingleAssignmentDisposable)(obj[1] = new SingleAssignmentDisposable());
_observer2 = new ZipObserver<T2>(_gate, this, 1, singleAssignmentDisposable2);
base.Queues[1] = _observer2.Values;
SingleAssignmentDisposable singleAssignmentDisposable3 = (SingleAssignmentDisposable)(obj[2] = new SingleAssignmentDisposable());
_observer3 = new ZipObserver<T3>(_gate, this, 2, singleAssignmentDisposable3);
base.Queues[2] = _observer3.Values;
SingleAssignmentDisposable singleAssignmentDisposable4 = (SingleAssignmentDisposable)(obj[3] = new SingleAssignmentDisposable());
_observer4 = new ZipObserver<T4>(_gate, this, 3, singleAssignmentDisposable4);
base.Queues[3] = _observer4.Values;
SingleAssignmentDisposable singleAssignmentDisposable5 = (SingleAssignmentDisposable)(obj[4] = new SingleAssignmentDisposable());
_observer5 = new ZipObserver<T5>(_gate, this, 4, singleAssignmentDisposable5);
base.Queues[4] = _observer5.Values;
SingleAssignmentDisposable singleAssignmentDisposable6 = (SingleAssignmentDisposable)(obj[5] = new SingleAssignmentDisposable());
_observer6 = new ZipObserver<T6>(_gate, this, 5, singleAssignmentDisposable6);
base.Queues[5] = _observer6.Values;
SingleAssignmentDisposable singleAssignmentDisposable7 = (SingleAssignmentDisposable)(obj[6] = new SingleAssignmentDisposable());
_observer7 = new ZipObserver<T7>(_gate, this, 6, singleAssignmentDisposable7);
base.Queues[6] = _observer7.Values;
singleAssignmentDisposable.Disposable = ObservableExtensions.SubscribeSafe<T1>(source1, (IObserver<T1>)_observer1);
singleAssignmentDisposable2.Disposable = ObservableExtensions.SubscribeSafe<T2>(source2, (IObserver<T2>)_observer2);
singleAssignmentDisposable3.Disposable = ObservableExtensions.SubscribeSafe<T3>(source3, (IObserver<T3>)_observer3);
singleAssignmentDisposable4.Disposable = ObservableExtensions.SubscribeSafe<T4>(source4, (IObserver<T4>)_observer4);
singleAssignmentDisposable5.Disposable = ObservableExtensions.SubscribeSafe<T5>(source5, (IObserver<T5>)_observer5);
singleAssignmentDisposable6.Disposable = ObservableExtensions.SubscribeSafe<T6>(source6, (IObserver<T6>)_observer6);
singleAssignmentDisposable7.Disposable = ObservableExtensions.SubscribeSafe<T7>(source7, (IObserver<T7>)_observer7);
obj[7] = Disposable.Create(delegate {
_observer1.Values.Clear();
_observer2.Values.Clear();
_observer3.Values.Clear();
_observer4.Values.Clear();
_observer5.Values.Clear();
_observer6.Values.Clear();
_observer7.Values.Clear();
});
return StableCompositeDisposable.Create(obj);
}
protected override TResult GetResult()
{
return _resultSelector(_observer1.Values.Dequeue(), _observer2.Values.Dequeue(), _observer3.Values.Dequeue(), _observer4.Values.Dequeue(), _observer5.Values.Dequeue(), _observer6.Values.Dequeue(), _observer7.Values.Dequeue());
}
}
private readonly IObservable<T1> _source1;
private readonly IObservable<T2> _source2;
private readonly IObservable<T3> _source3;
private readonly IObservable<T4> _source4;
private readonly IObservable<T5> _source5;
private readonly IObservable<T6> _source6;
private readonly IObservable<T7> _source7;
private readonly Func<T1, T2, T3, T4, T5, T6, T7, TResult> _resultSelector;
public Zip(IObservable<T1> source1, IObservable<T2> source2, IObservable<T3> source3, IObservable<T4> source4, IObservable<T5> source5, IObservable<T6> source6, IObservable<T7> source7, Func<T1, T2, T3, T4, T5, T6, T7, TResult> resultSelector)
{
_source1 = source1;
_source2 = source2;
_source3 = source3;
_source4 = source4;
_source5 = source5;
_source6 = source6;
_source7 = source7;
_resultSelector = resultSelector;
}
protected override _ CreateSink(IObserver<TResult> observer, IDisposable cancel)
{
return new _(_resultSelector, observer, cancel);
}
protected override IDisposable Run(_ sink)
{
return sink.Run(_source1, _source2, _source3, _source4, _source5, _source6, _source7);
}
}
}