Window<TSource>
using System.Collections.Generic;
using System.Reactive.Concurrency;
using System.Reactive.Disposables;
using System.Reactive.Subjects;
using System.Runtime.CompilerServices;
namespace System.Reactive.Linq.ObservableImpl
{
[System.Runtime.CompilerServices.NullableContext(1)]
[System.Runtime.CompilerServices.Nullable(0)]
internal static class Window<[System.Runtime.CompilerServices.Nullable(2)] TSource>
{
[System.Runtime.CompilerServices.Nullable(new byte[] {
0,
1,
1,
1,
0
})]
internal sealed class Count : Producer<IObservable<TSource>, Count._>
{
[System.Runtime.CompilerServices.Nullable(new byte[] {
0,
1,
1,
1
})]
internal sealed class _ : Sink<TSource, IObservable<TSource>>
{
private readonly Queue<ISubject<TSource>> _queue = new Queue<ISubject<TSource>>();
private readonly SingleAssignmentDisposable _m = new SingleAssignmentDisposable();
private readonly RefCountDisposable _refCountDisposable;
private readonly int _count;
private readonly int _skip;
private int _n;
public _([System.Runtime.CompilerServices.Nullable(new byte[] {
1,
0
})] Count parent, IObserver<IObservable<TSource>> observer)
: base(observer)
{
_refCountDisposable = new RefCountDisposable(_m);
_count = parent._count;
_skip = parent._skip;
}
public override void Run(IObservable<TSource> source)
{
IObservable<TSource> value = CreateWindow();
ForwardOnNext(value);
_m.Disposable = ObservableExtensions.SubscribeSafe<TSource>(source, (IObserver<TSource>)this);
SetUpstream(_refCountDisposable);
}
private IObservable<TSource> CreateWindow()
{
Subject<TSource> subject = new Subject<TSource>();
_queue.Enqueue(subject);
return new WindowObservable<TSource>(subject, _refCountDisposable);
}
public override void OnNext(TSource value)
{
foreach (ISubject<TSource> item in _queue) {
item.OnNext(value);
}
int num = _n - _count + 1;
if (num >= 0 && num % _skip == 0)
_queue.Dequeue().OnCompleted();
_n++;
if (_n % _skip == 0) {
IObservable<TSource> value2 = CreateWindow();
ForwardOnNext(value2);
}
}
public override void OnError(Exception error)
{
while (_queue.Count > 0) {
_queue.Dequeue().OnError(error);
}
ForwardOnError(error);
}
public override void OnCompleted()
{
while (_queue.Count > 0) {
_queue.Dequeue().OnCompleted();
}
ForwardOnCompleted();
}
}
private readonly IObservable<TSource> _source;
private readonly int _count;
private readonly int _skip;
public Count(IObservable<TSource> source, int count, int skip)
{
_source = source;
_count = count;
_skip = skip;
}
[return: System.Runtime.CompilerServices.Nullable(new byte[] {
1,
0
})]
protected override _ CreateSink(IObserver<IObservable<TSource>> observer)
{
return new _(this, observer);
}
protected override void Run([System.Runtime.CompilerServices.Nullable(new byte[] {
1,
0
})] _ sink)
{
sink.Run(_source);
}
}
[System.Runtime.CompilerServices.Nullable(new byte[] {
0,
1,
1,
1,
0
})]
internal sealed class TimeSliding : Producer<IObservable<TSource>, TimeSliding._>
{
[System.Runtime.CompilerServices.Nullable(new byte[] {
0,
1,
1,
1
})]
internal sealed class _ : Sink<TSource, IObservable<TSource>>
{
private readonly object _gate = new object();
private readonly Queue<ISubject<TSource>> _q = new Queue<ISubject<TSource>>();
private readonly SerialDisposable _timerD = new SerialDisposable();
private readonly IScheduler _scheduler;
private readonly TimeSpan _timeShift;
[System.Runtime.CompilerServices.Nullable(2)]
private RefCountDisposable _refCountDisposable;
private TimeSpan _totalTime;
private TimeSpan _nextShift;
private TimeSpan _nextSpan;
public _([System.Runtime.CompilerServices.Nullable(new byte[] {
1,
0
})] TimeSliding parent, IObserver<IObservable<TSource>> observer)
: base(observer)
{
_scheduler = parent._scheduler;
_timeShift = parent._timeShift;
}
public void Run([System.Runtime.CompilerServices.Nullable(new byte[] {
1,
0
})] TimeSliding parent)
{
_totalTime = TimeSpan.Zero;
_nextShift = parent._timeShift;
_nextSpan = parent._timeSpan;
CompositeDisposable compositeDisposable = new CompositeDisposable(2) {
_timerD
};
_refCountDisposable = new RefCountDisposable(compositeDisposable);
CreateWindow();
CreateTimer();
compositeDisposable.Add(ObservableExtensions.SubscribeSafe<TSource>(parent._source, (IObserver<TSource>)this));
SetUpstream(_refCountDisposable);
}
private void CreateWindow()
{
Subject<TSource> subject = new Subject<TSource>();
_q.Enqueue(subject);
ForwardOnNext(new WindowObservable<TSource>(subject, _refCountDisposable));
}
private void CreateTimer()
{
SingleAssignmentDisposable singleAssignmentDisposable = new SingleAssignmentDisposable();
_timerD.Disposable = singleAssignmentDisposable;
bool flag = false;
bool flag2 = false;
if (_nextSpan == _nextShift) {
flag = true;
flag2 = true;
} else if (_nextSpan < _nextShift) {
flag = true;
} else {
flag2 = true;
}
TimeSpan timeSpan = flag ? _nextSpan : _nextShift;
TimeSpan dueTime = timeSpan - _totalTime;
_totalTime = timeSpan;
if (flag)
_nextSpan += _timeShift;
if (flag2)
_nextShift += _timeShift;
singleAssignmentDisposable.Disposable = Scheduler.ScheduleAction<(_, bool, bool)>(_scheduler, (this, flag, flag2), dueTime, (Action<(_, bool, bool)>)delegate((_ this, bool isSpan, bool isShift) tuple) {
tuple.this.Tick(tuple.isSpan, tuple.isShift);
});
}
private void Tick(bool isSpan, bool isShift)
{
lock (_gate) {
if (isSpan)
_q.Dequeue().OnCompleted();
if (isShift)
CreateWindow();
}
CreateTimer();
}
public override void OnNext(TSource value)
{
lock (_gate) {
foreach (ISubject<TSource> item in _q) {
item.OnNext(value);
}
}
}
public override void OnError(Exception error)
{
lock (_gate) {
foreach (ISubject<TSource> item in _q) {
item.OnError(error);
}
ForwardOnError(error);
}
}
public override void OnCompleted()
{
lock (_gate) {
foreach (ISubject<TSource> item in _q) {
item.OnCompleted();
}
ForwardOnCompleted();
}
}
}
private readonly IObservable<TSource> _source;
private readonly TimeSpan _timeSpan;
private readonly TimeSpan _timeShift;
private readonly IScheduler _scheduler;
public TimeSliding(IObservable<TSource> source, TimeSpan timeSpan, TimeSpan timeShift, IScheduler scheduler)
{
_source = source;
_timeSpan = timeSpan;
_timeShift = timeShift;
_scheduler = scheduler;
}
[return: System.Runtime.CompilerServices.Nullable(new byte[] {
1,
0
})]
protected override _ CreateSink(IObserver<IObservable<TSource>> observer)
{
return new _(this, observer);
}
protected override void Run([System.Runtime.CompilerServices.Nullable(new byte[] {
1,
0
})] _ sink)
{
sink.Run(this);
}
}
[System.Runtime.CompilerServices.Nullable(new byte[] {
0,
1,
1,
1,
0
})]
internal sealed class TimeHopping : Producer<IObservable<TSource>, TimeHopping._>
{
[System.Runtime.CompilerServices.Nullable(new byte[] {
0,
1,
1,
1
})]
internal sealed class _ : Sink<TSource, IObservable<TSource>>
{
private readonly object _gate = new object();
private Subject<TSource> _subject;
[System.Runtime.CompilerServices.Nullable(2)]
private RefCountDisposable _refCountDisposable;
public _(IObserver<IObservable<TSource>> observer)
: base(observer)
{
_subject = new Subject<TSource>();
}
public void Run([System.Runtime.CompilerServices.Nullable(new byte[] {
1,
0
})] TimeHopping parent)
{
CompositeDisposable compositeDisposable = new CompositeDisposable(2);
_refCountDisposable = new RefCountDisposable(compositeDisposable);
NextWindow();
compositeDisposable.Add(Scheduler.SchedulePeriodic<_>(parent._scheduler, this, parent._timeSpan, (Action<_>)delegate(_ this) {
this.Tick();
}));
compositeDisposable.Add(ObservableExtensions.SubscribeSafe<TSource>(parent._source, (IObserver<TSource>)this));
SetUpstream(_refCountDisposable);
}
private void Tick()
{
lock (_gate) {
_subject.OnCompleted();
_subject = new Subject<TSource>();
NextWindow();
}
}
private void NextWindow()
{
ForwardOnNext(new WindowObservable<TSource>(_subject, _refCountDisposable));
}
public override void OnNext(TSource value)
{
lock (_gate) {
_subject.OnNext(value);
}
}
public override void OnError(Exception error)
{
lock (_gate) {
_subject.OnError(error);
ForwardOnError(error);
}
}
public override void OnCompleted()
{
lock (_gate) {
_subject.OnCompleted();
ForwardOnCompleted();
}
}
}
private readonly IObservable<TSource> _source;
private readonly TimeSpan _timeSpan;
private readonly IScheduler _scheduler;
public TimeHopping(IObservable<TSource> source, TimeSpan timeSpan, IScheduler scheduler)
{
_source = source;
_timeSpan = timeSpan;
_scheduler = scheduler;
}
[return: System.Runtime.CompilerServices.Nullable(new byte[] {
1,
0
})]
protected override _ CreateSink(IObserver<IObservable<TSource>> observer)
{
return new _(observer);
}
protected override void Run([System.Runtime.CompilerServices.Nullable(new byte[] {
1,
0
})] _ sink)
{
sink.Run(this);
}
}
[System.Runtime.CompilerServices.Nullable(new byte[] {
0,
1,
1,
1,
0
})]
internal sealed class Ferry : Producer<IObservable<TSource>, Ferry._>
{
[System.Runtime.CompilerServices.Nullable(new byte[] {
0,
1,
1,
1
})]
internal sealed class _ : Sink<TSource, IObservable<TSource>>
{
private readonly object _gate = new object();
private readonly SerialDisposable _timerD = new SerialDisposable();
private readonly int _count;
private readonly TimeSpan _timeSpan;
private readonly IScheduler _scheduler;
private Subject<TSource> _s;
private int _n;
[System.Runtime.CompilerServices.Nullable(2)]
private RefCountDisposable _refCountDisposable;
public _([System.Runtime.CompilerServices.Nullable(new byte[] {
1,
0
})] Ferry parent, IObserver<IObservable<TSource>> observer)
: base(observer)
{
_count = parent._count;
_timeSpan = parent._timeSpan;
_scheduler = parent._scheduler;
_s = new Subject<TSource>();
}
public override void Run(IObservable<TSource> source)
{
CompositeDisposable compositeDisposable = new CompositeDisposable(2) {
_timerD
};
_refCountDisposable = new RefCountDisposable(compositeDisposable);
NextWindow();
CreateTimer(_s);
compositeDisposable.Add(ObservableExtensions.SubscribeSafe<TSource>(source, (IObserver<TSource>)this));
SetUpstream(_refCountDisposable);
}
private void CreateTimer(Subject<TSource> window)
{
SingleAssignmentDisposable singleAssignmentDisposable = new SingleAssignmentDisposable();
_timerD.Disposable = singleAssignmentDisposable;
singleAssignmentDisposable.Disposable = Scheduler.ScheduleAction<(_, Subject<TSource>)>(_scheduler, (this, window), _timeSpan, (Action<(_, Subject<TSource>)>)delegate((_ this, Subject<TSource> window) tuple) {
tuple.this.Tick(tuple.window);
});
}
private void NextWindow()
{
ForwardOnNext(new WindowObservable<TSource>(_s, _refCountDisposable));
}
private void Tick(Subject<TSource> window)
{
Subject<TSource> subject = default(Subject<TSource>);
lock (_gate) {
if (window != _s)
return;
_n = 0;
subject = new Subject<TSource>();
_s.OnCompleted();
_s = subject;
NextWindow();
}
CreateTimer(subject);
}
public override void OnNext(TSource value)
{
Subject<TSource> subject = null;
lock (_gate) {
_s.OnNext(value);
_n++;
if (_n == _count) {
_n = 0;
subject = new Subject<TSource>();
_s.OnCompleted();
_s = subject;
NextWindow();
}
}
if (subject != null)
CreateTimer(subject);
}
public override void OnError(Exception error)
{
lock (_gate) {
_s.OnError(error);
ForwardOnError(error);
}
}
public override void OnCompleted()
{
lock (_gate) {
_s.OnCompleted();
ForwardOnCompleted();
}
}
}
private readonly IObservable<TSource> _source;
private readonly int _count;
private readonly TimeSpan _timeSpan;
private readonly IScheduler _scheduler;
public Ferry(IObservable<TSource> source, TimeSpan timeSpan, int count, IScheduler scheduler)
{
_source = source;
_timeSpan = timeSpan;
_count = count;
_scheduler = scheduler;
}
[return: System.Runtime.CompilerServices.Nullable(new byte[] {
1,
0
})]
protected override _ CreateSink(IObserver<IObservable<TSource>> observer)
{
return new _(this, observer);
}
protected override void Run([System.Runtime.CompilerServices.Nullable(new byte[] {
1,
0
})] _ sink)
{
sink.Run(_source);
}
}
}
}