Next<TSource>
using System.Reactive.Disposables;
using System.Threading;
using System.Reactive.Disposables;
using System.Reactive.Disposables;
namespace System.Reactive.Linq.ObservableImpl
{
internal sealed class Zip<T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, TResult> : Producer<TResult, Zip<T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, TResult>._>
{
internal sealed class _ : ZipSink<TResult>
{
private readonly Func<T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, TResult> _resultSelector = 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;
private ZipObserver<T8> _observer8;
private ZipObserver<T9> _observer9;
private ZipObserver<T10> _observer10;
private ZipObserver<T11> _observer11;
private ZipObserver<T12> _observer12;
private ZipObserver<T13> _observer13;
public _(Func<T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, TResult> resultSelector, IObserver<TResult> observer)
: base(13, observer)
{
}
public void Run(IObservable<T1> source1, IObservable<T2> source2, IObservable<T3> source3, IObservable<T4> source4, IObservable<T5> source5, IObservable<T6> source6, IObservable<T7> source7, IObservable<T8> source8, IObservable<T9> source9, IObservable<T10> source10, IObservable<T11> source11, IObservable<T12> source12, IObservable<T13> source13)
{
IDisposable[] array = new IDisposable[13];
_observer1 = new ZipObserver<T1>(_gate, this, 0);
IDisposable[] array;
array[0] = _observer1;
base.Queues[0] = _observer1.Values;
_observer2 = new ZipObserver<T2>(_gate, this, 1);
array[1] = _observer2;
base.Queues[1] = _observer2.Values;
_observer3 = new ZipObserver<T3>(_gate, this, 2);
array[2] = _observer3;
base.Queues[2] = _observer3.Values;
_observer4 = new ZipObserver<T4>(_gate, this, 3);
array[3] = _observer4;
base.Queues[3] = _observer4.Values;
_observer5 = new ZipObserver<T5>(_gate, this, 4);
array[4] = _observer5;
base.Queues[4] = _observer5.Values;
_observer6 = new ZipObserver<T6>(_gate, this, 5);
array[5] = _observer6;
base.Queues[5] = _observer6.Values;
_observer7 = new ZipObserver<T7>(_gate, this, 6);
array[6] = _observer7;
base.Queues[6] = _observer7.Values;
_observer8 = new ZipObserver<T8>(_gate, this, 7);
array[7] = _observer8;
base.Queues[7] = _observer8.Values;
_observer9 = new ZipObserver<T9>(_gate, this, 8);
array[8] = _observer9;
base.Queues[8] = _observer9.Values;
_observer10 = new ZipObserver<T10>(_gate, this, 9);
array[9] = _observer10;
base.Queues[9] = _observer10.Values;
_observer11 = new ZipObserver<T11>(_gate, this, 10);
array[10] = _observer11;
base.Queues[10] = _observer11.Values;
_observer12 = new ZipObserver<T12>(_gate, this, 11);
array[11] = _observer12;
base.Queues[11] = _observer12.Values;
_observer13 = new ZipObserver<T13>(_gate, this, 12);
array[12] = _observer13;
base.Queues[12] = _observer13.Values;
_observer1.SetResource(ObservableExtensions.SubscribeSafe<T1>(source1, (IObserver<T1>)_observer1));
_observer2.SetResource(ObservableExtensions.SubscribeSafe<T2>(source2, (IObserver<T2>)_observer2));
_observer3.SetResource(ObservableExtensions.SubscribeSafe<T3>(source3, (IObserver<T3>)_observer3));
_observer4.SetResource(ObservableExtensions.SubscribeSafe<T4>(source4, (IObserver<T4>)_observer4));
_observer5.SetResource(ObservableExtensions.SubscribeSafe<T5>(source5, (IObserver<T5>)_observer5));
_observer6.SetResource(ObservableExtensions.SubscribeSafe<T6>(source6, (IObserver<T6>)_observer6));
_observer7.SetResource(ObservableExtensions.SubscribeSafe<T7>(source7, (IObserver<T7>)_observer7));
_observer8.SetResource(ObservableExtensions.SubscribeSafe<T8>(source8, (IObserver<T8>)_observer8));
_observer9.SetResource(ObservableExtensions.SubscribeSafe<T9>(source9, (IObserver<T9>)_observer9));
_observer10.SetResource(ObservableExtensions.SubscribeSafe<T10>(source10, (IObserver<T10>)_observer10));
_observer11.SetResource(ObservableExtensions.SubscribeSafe<T11>(source11, (IObserver<T11>)_observer11));
_observer12.SetResource(ObservableExtensions.SubscribeSafe<T12>(source12, (IObserver<T12>)_observer12));
_observer13.SetResource(ObservableExtensions.SubscribeSafe<T13>(source13, (IObserver<T13>)_observer13));
SetUpstream(StableCompositeDisposable.CreateTrusted(array));
}
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(), _observer8.Values.Dequeue(), _observer9.Values.Dequeue(), _observer10.Values.Dequeue(), _observer11.Values.Dequeue(), _observer12.Values.Dequeue(), _observer13.Values.Dequeue());
}
}
private readonly IObservable<T1> _source1 = source1;
private readonly IObservable<T2> _source2 = source2;
private readonly IObservable<T3> _source3 = source3;
private readonly IObservable<T4> _source4 = source4;
private readonly IObservable<T5> _source5 = source5;
private readonly IObservable<T6> _source6 = source6;
private readonly IObservable<T7> _source7 = source7;
private readonly IObservable<T8> _source8 = source8;
private readonly IObservable<T9> _source9 = source9;
private readonly IObservable<T10> _source10 = source10;
private readonly IObservable<T11> _source11 = source11;
private readonly IObservable<T12> _source12 = source12;
private readonly IObservable<T13> _source13 = source13;
private readonly Func<T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, TResult> _resultSelector = resultSelector;
public Zip(IObservable<T1> source1, IObservable<T2> source2, IObservable<T3> source3, IObservable<T4> source4, IObservable<T5> source5, IObservable<T6> source6, IObservable<T7> source7, IObservable<T8> source8, IObservable<T9> source9, IObservable<T10> source10, IObservable<T11> source11, IObservable<T12> source12, IObservable<T13> source13, Func<T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, TResult> resultSelector)
{
}
protected override _ CreateSink(IObserver<TResult> observer)
{
return new _(_resultSelector, observer);
}
protected override void Run(_ sink)
{
sink.Run(_source1, _source2, _source3, _source4, _source5, _source6, _source7, _source8, _source9, _source10, _source11, _source12, _source13);
}
}
}
namespace System.Reactive.Linq.ObservableImpl
{
internal sealed class Next<TSource> : PushToPullAdapter<TSource, TSource>
{
private sealed class _ : PushToPullSink<TSource, TSource>
{
private readonly object _gate;
private readonly SemaphoreSlim _semaphore;
private bool _waiting;
private NotificationKind _kind;
private TSource _value;
private Exception _error;
public _()
{
_gate = new object();
_semaphore = new SemaphoreSlim(0, 1);
}
public override void OnNext(TSource value)
{
lock (_gate) {
if (_waiting) {
_value = value;
_kind = NotificationKind.OnNext;
_semaphore.Release();
}
_waiting = false;
}
}
public override void OnError(Exception error)
{
Dispose();
lock (_gate) {
_error = error;
_kind = NotificationKind.OnError;
if (_waiting)
_semaphore.Release();
_waiting = false;
}
}
public override void OnCompleted()
{
Dispose();
lock (_gate) {
_kind = NotificationKind.OnCompleted;
if (_waiting)
_semaphore.Release();
_waiting = false;
}
}
public override bool TryMoveNext(out TSource current)
{
bool flag = false;
lock (_gate) {
_waiting = true;
flag = (_kind != NotificationKind.OnNext);
}
if (!flag)
_semaphore.Wait();
switch (_kind) {
case NotificationKind.OnNext:
current = _value;
return true;
case NotificationKind.OnError:
_error.Throw();
break;
}
current = default(TSource);
return false;
}
}
public Next(IObservable<TSource> source)
: base(source)
{
}
protected override PushToPullSink<TSource, TSource> Run()
{
return new _();
}
}
}
namespace System.Reactive.Linq.ObservableImpl
{
internal sealed class Zip<T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, TResult> : Producer<TResult, Zip<T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, TResult>._>
{
internal sealed class _ : ZipSink<TResult>
{
private readonly Func<T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, TResult> _resultSelector = 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;
private ZipObserver<T8> _observer8;
private ZipObserver<T9> _observer9;
private ZipObserver<T10> _observer10;
private ZipObserver<T11> _observer11;
private ZipObserver<T12> _observer12;
private ZipObserver<T13> _observer13;
private ZipObserver<T14> _observer14;
private ZipObserver<T15> _observer15;
private ZipObserver<T16> _observer16;
public _(Func<T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, TResult> resultSelector, IObserver<TResult> observer)
: base(16, observer)
{
}
public void Run(IObservable<T1> source1, IObservable<T2> source2, IObservable<T3> source3, IObservable<T4> source4, IObservable<T5> source5, IObservable<T6> source6, IObservable<T7> source7, IObservable<T8> source8, IObservable<T9> source9, IObservable<T10> source10, IObservable<T11> source11, IObservable<T12> source12, IObservable<T13> source13, IObservable<T14> source14, IObservable<T15> source15, IObservable<T16> source16)
{
IDisposable[] array = new IDisposable[16];
_observer1 = new ZipObserver<T1>(_gate, this, 0);
IDisposable[] array;
IDisposable[] array;
array[0] = _observer1;
base.Queues[0] = _observer1.Values;
_observer2 = new ZipObserver<T2>(_gate, this, 1);
array[1] = _observer2;
base.Queues[1] = _observer2.Values;
_observer3 = new ZipObserver<T3>(_gate, this, 2);
array[2] = _observer3;
base.Queues[2] = _observer3.Values;
_observer4 = new ZipObserver<T4>(_gate, this, 3);
array[3] = _observer4;
base.Queues[3] = _observer4.Values;
_observer5 = new ZipObserver<T5>(_gate, this, 4);
array[4] = _observer5;
base.Queues[4] = _observer5.Values;
_observer6 = new ZipObserver<T6>(_gate, this, 5);
array[5] = _observer6;
base.Queues[5] = _observer6.Values;
_observer7 = new ZipObserver<T7>(_gate, this, 6);
array[6] = _observer7;
base.Queues[6] = _observer7.Values;
_observer8 = new ZipObserver<T8>(_gate, this, 7);
array[7] = _observer8;
base.Queues[7] = _observer8.Values;
_observer9 = new ZipObserver<T9>(_gate, this, 8);
array[8] = _observer9;
base.Queues[8] = _observer9.Values;
_observer10 = new ZipObserver<T10>(_gate, this, 9);
array[9] = _observer10;
base.Queues[9] = _observer10.Values;
_observer11 = new ZipObserver<T11>(_gate, this, 10);
array[10] = _observer11;
base.Queues[10] = _observer11.Values;
_observer12 = new ZipObserver<T12>(_gate, this, 11);
array[11] = _observer12;
base.Queues[11] = _observer12.Values;
_observer13 = new ZipObserver<T13>(_gate, this, 12);
array[12] = _observer13;
base.Queues[12] = _observer13.Values;
_observer14 = new ZipObserver<T14>(_gate, this, 13);
array[13] = _observer14;
base.Queues[13] = _observer14.Values;
_observer15 = new ZipObserver<T15>(_gate, this, 14);
array[14] = _observer15;
base.Queues[14] = _observer15.Values;
_observer16 = new ZipObserver<T16>(_gate, this, 15);
array[15] = _observer16;
base.Queues[15] = _observer16.Values;
_observer1.SetResource(ObservableExtensions.SubscribeSafe<T1>(source1, (IObserver<T1>)_observer1));
_observer2.SetResource(ObservableExtensions.SubscribeSafe<T2>(source2, (IObserver<T2>)_observer2));
_observer3.SetResource(ObservableExtensions.SubscribeSafe<T3>(source3, (IObserver<T3>)_observer3));
_observer4.SetResource(ObservableExtensions.SubscribeSafe<T4>(source4, (IObserver<T4>)_observer4));
_observer5.SetResource(ObservableExtensions.SubscribeSafe<T5>(source5, (IObserver<T5>)_observer5));
_observer6.SetResource(ObservableExtensions.SubscribeSafe<T6>(source6, (IObserver<T6>)_observer6));
_observer7.SetResource(ObservableExtensions.SubscribeSafe<T7>(source7, (IObserver<T7>)_observer7));
_observer8.SetResource(ObservableExtensions.SubscribeSafe<T8>(source8, (IObserver<T8>)_observer8));
_observer9.SetResource(ObservableExtensions.SubscribeSafe<T9>(source9, (IObserver<T9>)_observer9));
_observer10.SetResource(ObservableExtensions.SubscribeSafe<T10>(source10, (IObserver<T10>)_observer10));
_observer11.SetResource(ObservableExtensions.SubscribeSafe<T11>(source11, (IObserver<T11>)_observer11));
_observer12.SetResource(ObservableExtensions.SubscribeSafe<T12>(source12, (IObserver<T12>)_observer12));
_observer13.SetResource(ObservableExtensions.SubscribeSafe<T13>(source13, (IObserver<T13>)_observer13));
_observer14.SetResource(ObservableExtensions.SubscribeSafe<T14>(source14, (IObserver<T14>)_observer14));
_observer15.SetResource(ObservableExtensions.SubscribeSafe<T15>(source15, (IObserver<T15>)_observer15));
_observer16.SetResource(ObservableExtensions.SubscribeSafe<T16>(source16, (IObserver<T16>)_observer16));
SetUpstream(StableCompositeDisposable.CreateTrusted(array));
}
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(), _observer8.Values.Dequeue(), _observer9.Values.Dequeue(), _observer10.Values.Dequeue(), _observer11.Values.Dequeue(), _observer12.Values.Dequeue(), _observer13.Values.Dequeue(), _observer14.Values.Dequeue(), _observer15.Values.Dequeue(), _observer16.Values.Dequeue());
}
}
private readonly IObservable<T1> _source1 = source1;
private readonly IObservable<T2> _source2 = source2;
private readonly IObservable<T3> _source3 = source3;
private readonly IObservable<T4> _source4 = source4;
private readonly IObservable<T5> _source5 = source5;
private readonly IObservable<T6> _source6 = source6;
private readonly IObservable<T7> _source7 = source7;
private readonly IObservable<T8> _source8 = source8;
private readonly IObservable<T9> _source9 = source9;
private readonly IObservable<T10> _source10 = source10;
private readonly IObservable<T11> _source11 = source11;
private readonly IObservable<T12> _source12 = source12;
private readonly IObservable<T13> _source13 = source13;
private readonly IObservable<T14> _source14 = source14;
private readonly IObservable<T15> _source15 = source15;
private readonly IObservable<T16> _source16 = source16;
private readonly Func<T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, TResult> _resultSelector = resultSelector;
public Zip(IObservable<T1> source1, IObservable<T2> source2, IObservable<T3> source3, IObservable<T4> source4, IObservable<T5> source5, IObservable<T6> source6, IObservable<T7> source7, IObservable<T8> source8, IObservable<T9> source9, IObservable<T10> source10, IObservable<T11> source11, IObservable<T12> source12, IObservable<T13> source13, IObservable<T14> source14, IObservable<T15> source15, IObservable<T16> source16, Func<T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, TResult> resultSelector)
{
}
protected override _ CreateSink(IObserver<TResult> observer)
{
return new _(_resultSelector, observer);
}
protected override void Run(_ sink)
{
sink.Run(_source1, _source2, _source3, _source4, _source5, _source6, _source7, _source8, _source9, _source10, _source11, _source12, _source13, _source14, _source15, _source16);
}
}
}