|
|
@@ -8,697 +8,709 @@ using System.Reactive.Disposables;
|
|
|
|
|
|
namespace System.Reactive.Linq.ObservableImpl
|
|
|
{
|
|
|
- internal sealed class Buffer<TSource> : Producer<IList<TSource>>
|
|
|
+ internal static class Buffer<TSource>
|
|
|
{
|
|
|
- private readonly IObservable<TSource> _source;
|
|
|
- private readonly int _count;
|
|
|
- private readonly int _skip;
|
|
|
-
|
|
|
- private readonly TimeSpan _timeSpan;
|
|
|
- private readonly TimeSpan _timeShift;
|
|
|
- private readonly IScheduler _scheduler;
|
|
|
-
|
|
|
- public Buffer(IObservable<TSource> source, int count, int skip)
|
|
|
- {
|
|
|
- _source = source;
|
|
|
- _count = count;
|
|
|
- _skip = skip;
|
|
|
- }
|
|
|
-
|
|
|
- public Buffer(IObservable<TSource> source, TimeSpan timeSpan, TimeSpan timeShift, IScheduler scheduler)
|
|
|
+ internal sealed class Count : Producer<IList<TSource>>
|
|
|
{
|
|
|
- _source = source;
|
|
|
- _timeSpan = timeSpan;
|
|
|
- _timeShift = timeShift;
|
|
|
- _scheduler = scheduler;
|
|
|
- }
|
|
|
+ private readonly IObservable<TSource> _source;
|
|
|
+ private readonly int _count;
|
|
|
+ private readonly int _skip;
|
|
|
|
|
|
- public Buffer(IObservable<TSource> source, TimeSpan timeSpan, int count, IScheduler scheduler)
|
|
|
- {
|
|
|
- _source = source;
|
|
|
- _timeSpan = timeSpan;
|
|
|
- _count = count;
|
|
|
- _scheduler = scheduler;
|
|
|
- }
|
|
|
-
|
|
|
- protected override IDisposable Run(IObserver<IList<TSource>> observer, IDisposable cancel, Action<IDisposable> setSink)
|
|
|
- {
|
|
|
- if (_scheduler == null)
|
|
|
+ public Count(IObservable<TSource> source, int count, int skip)
|
|
|
{
|
|
|
- var sink = new _(this, observer, cancel);
|
|
|
- setSink(sink);
|
|
|
- return sink.Run();
|
|
|
+ _source = source;
|
|
|
+ _count = count;
|
|
|
+ _skip = skip;
|
|
|
}
|
|
|
- else if (_count > 0)
|
|
|
+
|
|
|
+ protected override IDisposable Run(IObserver<IList<TSource>> observer, IDisposable cancel, Action<IDisposable> setSink)
|
|
|
{
|
|
|
- var sink = new Impl(this, observer, cancel);
|
|
|
+ var sink = new _(this, observer, cancel);
|
|
|
setSink(sink);
|
|
|
- return sink.Run();
|
|
|
- }
|
|
|
- else
|
|
|
- {
|
|
|
- if (_timeSpan == _timeShift)
|
|
|
- {
|
|
|
- var sink = new BufferTimeShift(this, observer, cancel);
|
|
|
- setSink(sink);
|
|
|
- return sink.Run();
|
|
|
- }
|
|
|
- else
|
|
|
- {
|
|
|
- var sink = new BufferImpl(this, observer, cancel);
|
|
|
- setSink(sink);
|
|
|
- return sink.Run();
|
|
|
- }
|
|
|
+ return sink.Run(_source);
|
|
|
}
|
|
|
- }
|
|
|
-
|
|
|
- class _ : Sink<IList<TSource>>, IObserver<TSource>
|
|
|
- {
|
|
|
- private readonly Buffer<TSource> _parent;
|
|
|
|
|
|
- public _(Buffer<TSource> parent, IObserver<IList<TSource>> observer, IDisposable cancel)
|
|
|
- : base(observer, cancel)
|
|
|
+ private sealed class _ : Sink<IList<TSource>>, IObserver<TSource>
|
|
|
{
|
|
|
- _parent = parent;
|
|
|
- }
|
|
|
+ private readonly Queue<IList<TSource>> _queue = new Queue<IList<TSource>>();
|
|
|
|
|
|
- private Queue<IList<TSource>> _queue;
|
|
|
- private int _n;
|
|
|
+ private readonly int _count;
|
|
|
+ private readonly int _skip;
|
|
|
|
|
|
- public IDisposable Run()
|
|
|
- {
|
|
|
- _queue = new Queue<IList<TSource>>();
|
|
|
- _n = 0;
|
|
|
+ public _(Count parent, IObserver<IList<TSource>> observer, IDisposable cancel)
|
|
|
+ : base(observer, cancel)
|
|
|
+ {
|
|
|
+ _count = parent._count;
|
|
|
+ _skip = parent._skip;
|
|
|
+ }
|
|
|
|
|
|
- CreateWindow();
|
|
|
- return _parent._source.SubscribeSafe(this);
|
|
|
- }
|
|
|
+ private int _n;
|
|
|
|
|
|
- private void CreateWindow()
|
|
|
- {
|
|
|
- var s = new List<TSource>();
|
|
|
- _queue.Enqueue(s);
|
|
|
- }
|
|
|
+ public IDisposable Run(IObservable<TSource> source)
|
|
|
+ {
|
|
|
+ _n = 0;
|
|
|
|
|
|
- public void OnNext(TSource value)
|
|
|
- {
|
|
|
- foreach (var s in _queue)
|
|
|
- s.Add(value);
|
|
|
+ CreateWindow();
|
|
|
+ return source.SubscribeSafe(this);
|
|
|
+ }
|
|
|
|
|
|
- var c = _n - _parent._count + 1;
|
|
|
- if (c >= 0 && c % _parent._skip == 0)
|
|
|
+ private void CreateWindow()
|
|
|
{
|
|
|
- var s = _queue.Dequeue();
|
|
|
- if (s.Count > 0)
|
|
|
- base._observer.OnNext(s);
|
|
|
+ var s = new List<TSource>();
|
|
|
+ _queue.Enqueue(s);
|
|
|
}
|
|
|
|
|
|
- _n++;
|
|
|
- if (_n % _parent._skip == 0)
|
|
|
- CreateWindow();
|
|
|
- }
|
|
|
+ public void OnNext(TSource value)
|
|
|
+ {
|
|
|
+ foreach (var s in _queue)
|
|
|
+ s.Add(value);
|
|
|
|
|
|
- public void OnError(Exception error)
|
|
|
- {
|
|
|
- while (_queue.Count > 0)
|
|
|
- _queue.Dequeue().Clear();
|
|
|
+ var c = _n - _count + 1;
|
|
|
+ if (c >= 0 && c % _skip == 0)
|
|
|
+ {
|
|
|
+ var s = _queue.Dequeue();
|
|
|
+ if (s.Count > 0)
|
|
|
+ base._observer.OnNext(s);
|
|
|
+ }
|
|
|
|
|
|
- base._observer.OnError(error);
|
|
|
- base.Dispose();
|
|
|
- }
|
|
|
+ _n++;
|
|
|
+ if (_n % _skip == 0)
|
|
|
+ CreateWindow();
|
|
|
+ }
|
|
|
|
|
|
- public void OnCompleted()
|
|
|
- {
|
|
|
- while (_queue.Count > 0)
|
|
|
+ public void OnError(Exception error)
|
|
|
{
|
|
|
- var s = _queue.Dequeue();
|
|
|
- if (s.Count > 0)
|
|
|
- base._observer.OnNext(s);
|
|
|
+ while (_queue.Count > 0)
|
|
|
+ _queue.Dequeue().Clear();
|
|
|
+
|
|
|
+ base._observer.OnError(error);
|
|
|
+ base.Dispose();
|
|
|
}
|
|
|
|
|
|
- base._observer.OnCompleted();
|
|
|
- base.Dispose();
|
|
|
+ public void OnCompleted()
|
|
|
+ {
|
|
|
+ while (_queue.Count > 0)
|
|
|
+ {
|
|
|
+ var s = _queue.Dequeue();
|
|
|
+ if (s.Count > 0)
|
|
|
+ base._observer.OnNext(s);
|
|
|
+ }
|
|
|
+
|
|
|
+ base._observer.OnCompleted();
|
|
|
+ base.Dispose();
|
|
|
+ }
|
|
|
}
|
|
|
}
|
|
|
|
|
|
- class BufferImpl : Sink<IList<TSource>>, IObserver<TSource>
|
|
|
+ internal sealed class TimeSliding : Producer<IList<TSource>>
|
|
|
{
|
|
|
- private readonly Buffer<TSource> _parent;
|
|
|
+ private readonly IObservable<TSource> _source;
|
|
|
+
|
|
|
+ private readonly TimeSpan _timeSpan;
|
|
|
+ private readonly TimeSpan _timeShift;
|
|
|
+ private readonly IScheduler _scheduler;
|
|
|
|
|
|
- public BufferImpl(Buffer<TSource> parent, IObserver<IList<TSource>> observer, IDisposable cancel)
|
|
|
- : base(observer, cancel)
|
|
|
+ public TimeSliding(IObservable<TSource> source, TimeSpan timeSpan, TimeSpan timeShift, IScheduler scheduler)
|
|
|
{
|
|
|
- _parent = parent;
|
|
|
+ _source = source;
|
|
|
+ _timeSpan = timeSpan;
|
|
|
+ _timeShift = timeShift;
|
|
|
+ _scheduler = scheduler;
|
|
|
}
|
|
|
|
|
|
- private TimeSpan _totalTime;
|
|
|
- private TimeSpan _nextShift;
|
|
|
- private TimeSpan _nextSpan;
|
|
|
-
|
|
|
- private object _gate;
|
|
|
- private Queue<List<TSource>> _q;
|
|
|
-
|
|
|
- private SerialDisposable _timerD;
|
|
|
+ protected override IDisposable Run(IObserver<IList<TSource>> observer, IDisposable cancel, Action<IDisposable> setSink)
|
|
|
+ {
|
|
|
+ var sink = new _(this, observer, cancel);
|
|
|
+ setSink(sink);
|
|
|
+ return sink.Run(this);
|
|
|
+ }
|
|
|
|
|
|
- public IDisposable Run()
|
|
|
+ private sealed class _ : Sink<IList<TSource>>, IObserver<TSource>
|
|
|
{
|
|
|
- _totalTime = TimeSpan.Zero;
|
|
|
- _nextShift = _parent._timeShift;
|
|
|
- _nextSpan = _parent._timeSpan;
|
|
|
+ private readonly TimeSpan _timeShift;
|
|
|
+ private readonly IScheduler _scheduler;
|
|
|
|
|
|
- _gate = new object();
|
|
|
- _q = new Queue<List<TSource>>();
|
|
|
+ private readonly object _gate = new object();
|
|
|
+ private readonly Queue<List<TSource>> _q = new Queue<List<TSource>>();
|
|
|
+ private readonly SerialDisposable _timerD = new SerialDisposable();
|
|
|
|
|
|
- _timerD = new SerialDisposable();
|
|
|
+ public _(TimeSliding parent, IObserver<IList<TSource>> observer, IDisposable cancel)
|
|
|
+ : base(observer, cancel)
|
|
|
+ {
|
|
|
+ _timeShift = parent._timeShift;
|
|
|
+ _scheduler = parent._scheduler;
|
|
|
+ }
|
|
|
|
|
|
- CreateWindow();
|
|
|
- CreateTimer();
|
|
|
+ private TimeSpan _totalTime;
|
|
|
+ private TimeSpan _nextShift;
|
|
|
+ private TimeSpan _nextSpan;
|
|
|
|
|
|
- var subscription = _parent._source.SubscribeSafe(this);
|
|
|
+ public IDisposable Run(TimeSliding parent)
|
|
|
+ {
|
|
|
+ _totalTime = TimeSpan.Zero;
|
|
|
+ _nextShift = parent._timeShift;
|
|
|
+ _nextSpan = parent._timeSpan;
|
|
|
|
|
|
- return StableCompositeDisposable.Create(_timerD, subscription);
|
|
|
- }
|
|
|
+ CreateWindow();
|
|
|
+ CreateTimer();
|
|
|
|
|
|
- private void CreateWindow()
|
|
|
- {
|
|
|
- var s = new List<TSource>();
|
|
|
- _q.Enqueue(s);
|
|
|
- }
|
|
|
+ var subscription = parent._source.SubscribeSafe(this);
|
|
|
|
|
|
- private void CreateTimer()
|
|
|
- {
|
|
|
- var m = new SingleAssignmentDisposable();
|
|
|
- _timerD.Disposable = m;
|
|
|
+ return StableCompositeDisposable.Create(_timerD, subscription);
|
|
|
+ }
|
|
|
|
|
|
- var isSpan = false;
|
|
|
- var isShift = false;
|
|
|
- if (_nextSpan == _nextShift)
|
|
|
+ private void CreateWindow()
|
|
|
{
|
|
|
- isSpan = true;
|
|
|
- isShift = true;
|
|
|
+ var s = new List<TSource>();
|
|
|
+ _q.Enqueue(s);
|
|
|
}
|
|
|
- else if (_nextSpan < _nextShift)
|
|
|
- isSpan = true;
|
|
|
- else
|
|
|
- isShift = true;
|
|
|
|
|
|
- var newTotalTime = isSpan ? _nextSpan : _nextShift;
|
|
|
- var ts = newTotalTime - _totalTime;
|
|
|
- _totalTime = newTotalTime;
|
|
|
+ private void CreateTimer()
|
|
|
+ {
|
|
|
+ var m = new SingleAssignmentDisposable();
|
|
|
+ _timerD.Disposable = m;
|
|
|
|
|
|
- if (isSpan)
|
|
|
- _nextSpan += _parent._timeShift;
|
|
|
- if (isShift)
|
|
|
- _nextShift += _parent._timeShift;
|
|
|
+ var isSpan = false;
|
|
|
+ var isShift = false;
|
|
|
+ if (_nextSpan == _nextShift)
|
|
|
+ {
|
|
|
+ isSpan = true;
|
|
|
+ isShift = true;
|
|
|
+ }
|
|
|
+ else if (_nextSpan < _nextShift)
|
|
|
+ isSpan = true;
|
|
|
+ else
|
|
|
+ isShift = true;
|
|
|
|
|
|
- m.Disposable = _parent._scheduler.Schedule(new State { isSpan = isSpan, isShift = isShift }, ts, Tick);
|
|
|
- }
|
|
|
+ var newTotalTime = isSpan ? _nextSpan : _nextShift;
|
|
|
+ var ts = newTotalTime - _totalTime;
|
|
|
+ _totalTime = newTotalTime;
|
|
|
|
|
|
- struct State
|
|
|
- {
|
|
|
- public bool isSpan;
|
|
|
- public bool isShift;
|
|
|
- }
|
|
|
+ if (isSpan)
|
|
|
+ _nextSpan += _timeShift;
|
|
|
+ if (isShift)
|
|
|
+ _nextShift += _timeShift;
|
|
|
|
|
|
- private IDisposable Tick(IScheduler self, State state)
|
|
|
- {
|
|
|
- lock (_gate)
|
|
|
+ m.Disposable = _scheduler.Schedule(new State { isSpan = isSpan, isShift = isShift }, ts, Tick);
|
|
|
+ }
|
|
|
+
|
|
|
+ private struct State
|
|
|
{
|
|
|
- //
|
|
|
- // Before v2, the two operations below were reversed. This doesn't have an observable
|
|
|
- // difference for Buffer, but is done to keep code consistent with Window, where we
|
|
|
- // took a breaking change in v2 to ensure consistency across overloads. For more info,
|
|
|
- // see the comment in Tick for Window.
|
|
|
- //
|
|
|
- if (state.isSpan)
|
|
|
- {
|
|
|
- var s = _q.Dequeue();
|
|
|
- base._observer.OnNext(s);
|
|
|
- }
|
|
|
+ public bool isSpan;
|
|
|
+ public bool isShift;
|
|
|
+ }
|
|
|
|
|
|
- if (state.isShift)
|
|
|
+ private IDisposable Tick(IScheduler self, State state)
|
|
|
+ {
|
|
|
+ lock (_gate)
|
|
|
{
|
|
|
- CreateWindow();
|
|
|
+ //
|
|
|
+ // Before v2, the two operations below were reversed. This doesn't have an observable
|
|
|
+ // difference for Buffer, but is done to keep code consistent with Window, where we
|
|
|
+ // took a breaking change in v2 to ensure consistency across overloads. For more info,
|
|
|
+ // see the comment in Tick for Window.
|
|
|
+ //
|
|
|
+ if (state.isSpan)
|
|
|
+ {
|
|
|
+ var s = _q.Dequeue();
|
|
|
+ base._observer.OnNext(s);
|
|
|
+ }
|
|
|
+
|
|
|
+ if (state.isShift)
|
|
|
+ {
|
|
|
+ CreateWindow();
|
|
|
+ }
|
|
|
}
|
|
|
- }
|
|
|
|
|
|
- CreateTimer();
|
|
|
+ CreateTimer();
|
|
|
|
|
|
- return Disposable.Empty;
|
|
|
- }
|
|
|
+ return Disposable.Empty;
|
|
|
+ }
|
|
|
|
|
|
- public void OnNext(TSource value)
|
|
|
- {
|
|
|
- lock (_gate)
|
|
|
+ public void OnNext(TSource value)
|
|
|
{
|
|
|
- foreach (var s in _q)
|
|
|
- s.Add(value);
|
|
|
+ lock (_gate)
|
|
|
+ {
|
|
|
+ foreach (var s in _q)
|
|
|
+ s.Add(value);
|
|
|
+ }
|
|
|
}
|
|
|
- }
|
|
|
|
|
|
- public void OnError(Exception error)
|
|
|
- {
|
|
|
- lock (_gate)
|
|
|
+ public void OnError(Exception error)
|
|
|
{
|
|
|
- while (_q.Count > 0)
|
|
|
- _q.Dequeue().Clear();
|
|
|
+ lock (_gate)
|
|
|
+ {
|
|
|
+ while (_q.Count > 0)
|
|
|
+ _q.Dequeue().Clear();
|
|
|
|
|
|
- base._observer.OnError(error);
|
|
|
- base.Dispose();
|
|
|
+ base._observer.OnError(error);
|
|
|
+ base.Dispose();
|
|
|
+ }
|
|
|
}
|
|
|
- }
|
|
|
|
|
|
- public void OnCompleted()
|
|
|
- {
|
|
|
- lock (_gate)
|
|
|
+ public void OnCompleted()
|
|
|
{
|
|
|
- while (_q.Count > 0)
|
|
|
- base._observer.OnNext(_q.Dequeue());
|
|
|
+ lock (_gate)
|
|
|
+ {
|
|
|
+ while (_q.Count > 0)
|
|
|
+ base._observer.OnNext(_q.Dequeue());
|
|
|
|
|
|
- base._observer.OnCompleted();
|
|
|
- base.Dispose();
|
|
|
+ base._observer.OnCompleted();
|
|
|
+ base.Dispose();
|
|
|
+ }
|
|
|
}
|
|
|
}
|
|
|
}
|
|
|
|
|
|
- class BufferTimeShift : Sink<IList<TSource>>, IObserver<TSource>
|
|
|
+ internal sealed class TimeHopping : Producer<IList<TSource>>
|
|
|
{
|
|
|
- private readonly Buffer<TSource> _parent;
|
|
|
+ private readonly IObservable<TSource> _source;
|
|
|
|
|
|
- public BufferTimeShift(Buffer<TSource> parent, IObserver<IList<TSource>> observer, IDisposable cancel)
|
|
|
- : base(observer, cancel)
|
|
|
+ private readonly TimeSpan _timeSpan;
|
|
|
+ private readonly IScheduler _scheduler;
|
|
|
+
|
|
|
+ public TimeHopping(IObservable<TSource> source, TimeSpan timeSpan, IScheduler scheduler)
|
|
|
{
|
|
|
- _parent = parent;
|
|
|
+ _source = source;
|
|
|
+ _timeSpan = timeSpan;
|
|
|
+ _scheduler = scheduler;
|
|
|
}
|
|
|
|
|
|
- private object _gate;
|
|
|
- private List<TSource> _list;
|
|
|
+ protected override IDisposable Run(IObserver<IList<TSource>> observer, IDisposable cancel, Action<IDisposable> setSink)
|
|
|
+ {
|
|
|
+ var sink = new _( observer, cancel);
|
|
|
+ setSink(sink);
|
|
|
+ return sink.Run(this);
|
|
|
+ }
|
|
|
|
|
|
- public IDisposable Run()
|
|
|
+ private sealed class _ : Sink<IList<TSource>>, IObserver<TSource>
|
|
|
{
|
|
|
- _gate = new object();
|
|
|
- _list = new List<TSource>();
|
|
|
+ private readonly object _gate = new object();
|
|
|
|
|
|
- var d = _parent._scheduler.SchedulePeriodic(_parent._timeSpan, Tick);
|
|
|
- var s = _parent._source.SubscribeSafe(this);
|
|
|
+ public _(IObserver<IList<TSource>> observer, IDisposable cancel)
|
|
|
+ : base(observer, cancel)
|
|
|
+ {
|
|
|
+ }
|
|
|
|
|
|
- return StableCompositeDisposable.Create(d, s);
|
|
|
- }
|
|
|
+ private List<TSource> _list;
|
|
|
|
|
|
- private void Tick()
|
|
|
- {
|
|
|
- lock (_gate)
|
|
|
+ public IDisposable Run(TimeHopping parent)
|
|
|
{
|
|
|
- base._observer.OnNext(_list);
|
|
|
_list = new List<TSource>();
|
|
|
+
|
|
|
+ var d = parent._scheduler.SchedulePeriodic(parent._timeSpan, Tick);
|
|
|
+ var s = parent._source.SubscribeSafe(this);
|
|
|
+
|
|
|
+ return StableCompositeDisposable.Create(d, s);
|
|
|
}
|
|
|
- }
|
|
|
|
|
|
- public void OnNext(TSource value)
|
|
|
- {
|
|
|
- lock (_gate)
|
|
|
+ private void Tick()
|
|
|
{
|
|
|
- _list.Add(value);
|
|
|
+ lock (_gate)
|
|
|
+ {
|
|
|
+ base._observer.OnNext(_list);
|
|
|
+ _list = new List<TSource>();
|
|
|
+ }
|
|
|
}
|
|
|
- }
|
|
|
|
|
|
- public void OnError(Exception error)
|
|
|
- {
|
|
|
- lock (_gate)
|
|
|
+ public void OnNext(TSource value)
|
|
|
{
|
|
|
- _list.Clear();
|
|
|
+ lock (_gate)
|
|
|
+ {
|
|
|
+ _list.Add(value);
|
|
|
+ }
|
|
|
+ }
|
|
|
|
|
|
- base._observer.OnError(error);
|
|
|
- base.Dispose();
|
|
|
+ public void OnError(Exception error)
|
|
|
+ {
|
|
|
+ lock (_gate)
|
|
|
+ {
|
|
|
+ _list.Clear();
|
|
|
+
|
|
|
+ base._observer.OnError(error);
|
|
|
+ base.Dispose();
|
|
|
+ }
|
|
|
}
|
|
|
- }
|
|
|
|
|
|
- public void OnCompleted()
|
|
|
- {
|
|
|
- lock (_gate)
|
|
|
+ public void OnCompleted()
|
|
|
{
|
|
|
- base._observer.OnNext(_list);
|
|
|
- base._observer.OnCompleted();
|
|
|
- base.Dispose();
|
|
|
+ lock (_gate)
|
|
|
+ {
|
|
|
+ base._observer.OnNext(_list);
|
|
|
+ base._observer.OnCompleted();
|
|
|
+ base.Dispose();
|
|
|
+ }
|
|
|
}
|
|
|
}
|
|
|
}
|
|
|
|
|
|
- class Impl : Sink<IList<TSource>>, IObserver<TSource>
|
|
|
+ internal sealed class Ferry : Producer<IList<TSource>>
|
|
|
{
|
|
|
- private readonly Buffer<TSource> _parent;
|
|
|
+ private readonly IObservable<TSource> _source;
|
|
|
+ private readonly int _count;
|
|
|
+ private readonly TimeSpan _timeSpan;
|
|
|
+ private readonly IScheduler _scheduler;
|
|
|
|
|
|
- public Impl(Buffer<TSource> parent, IObserver<IList<TSource>> observer, IDisposable cancel)
|
|
|
- : base(observer, cancel)
|
|
|
+ public Ferry(IObservable<TSource> source, TimeSpan timeSpan, int count, IScheduler scheduler)
|
|
|
{
|
|
|
- _parent = parent;
|
|
|
+ _source = source;
|
|
|
+ _timeSpan = timeSpan;
|
|
|
+ _count = count;
|
|
|
+ _scheduler = scheduler;
|
|
|
}
|
|
|
|
|
|
- private object _gate;
|
|
|
- private IList<TSource> _s;
|
|
|
- private int _n;
|
|
|
- private int _windowId;
|
|
|
-
|
|
|
- private SerialDisposable _timerD;
|
|
|
-
|
|
|
- public IDisposable Run()
|
|
|
+ protected override IDisposable Run(IObserver<IList<TSource>> observer, IDisposable cancel, Action<IDisposable> setSink)
|
|
|
{
|
|
|
- _gate = new object();
|
|
|
- _s = default(IList<TSource>);
|
|
|
- _n = 0;
|
|
|
- _windowId = 0;
|
|
|
-
|
|
|
- _timerD = new SerialDisposable();
|
|
|
-
|
|
|
- _s = new List<TSource>();
|
|
|
- CreateTimer(0);
|
|
|
-
|
|
|
- var subscription = _parent._source.SubscribeSafe(this);
|
|
|
-
|
|
|
- return StableCompositeDisposable.Create(_timerD, subscription);
|
|
|
+ var sink = new _(this, observer, cancel);
|
|
|
+ setSink(sink);
|
|
|
+ return sink.Run();
|
|
|
}
|
|
|
|
|
|
- private void CreateTimer(int id)
|
|
|
+ private sealed class _ : Sink<IList<TSource>>, IObserver<TSource>
|
|
|
{
|
|
|
- var m = new SingleAssignmentDisposable();
|
|
|
- _timerD.Disposable = m;
|
|
|
+ private readonly Ferry _parent;
|
|
|
|
|
|
- m.Disposable = _parent._scheduler.Schedule(id, _parent._timeSpan, Tick);
|
|
|
- }
|
|
|
+ private readonly object _gate = new object();
|
|
|
+ private readonly SerialDisposable _timerD = new SerialDisposable();
|
|
|
|
|
|
- private IDisposable Tick(IScheduler self, int id)
|
|
|
- {
|
|
|
- var d = Disposable.Empty;
|
|
|
-
|
|
|
- var newId = 0;
|
|
|
- lock (_gate)
|
|
|
+ public _(Ferry parent, IObserver<IList<TSource>> observer, IDisposable cancel)
|
|
|
+ : base(observer, cancel)
|
|
|
{
|
|
|
- if (id != _windowId)
|
|
|
- return d;
|
|
|
+ _parent = parent;
|
|
|
+ }
|
|
|
|
|
|
- _n = 0;
|
|
|
- newId = ++_windowId;
|
|
|
+ private IList<TSource> _s;
|
|
|
+ private int _n;
|
|
|
+ private int _windowId;
|
|
|
|
|
|
- var res = _s;
|
|
|
+ public IDisposable Run()
|
|
|
+ {
|
|
|
_s = new List<TSource>();
|
|
|
- base._observer.OnNext(res);
|
|
|
+ _n = 0;
|
|
|
+ _windowId = 0;
|
|
|
+
|
|
|
+ CreateTimer(0);
|
|
|
+
|
|
|
+ var subscription = _parent._source.SubscribeSafe(this);
|
|
|
|
|
|
- CreateTimer(newId);
|
|
|
+ return StableCompositeDisposable.Create(_timerD, subscription);
|
|
|
}
|
|
|
|
|
|
- return d;
|
|
|
- }
|
|
|
+ private void CreateTimer(int id)
|
|
|
+ {
|
|
|
+ var m = new SingleAssignmentDisposable();
|
|
|
+ _timerD.Disposable = m;
|
|
|
|
|
|
- public void OnNext(TSource value)
|
|
|
- {
|
|
|
- var newWindow = false;
|
|
|
- var newId = 0;
|
|
|
+ m.Disposable = _parent._scheduler.Schedule(id, _parent._timeSpan, Tick);
|
|
|
+ }
|
|
|
|
|
|
- lock (_gate)
|
|
|
+ private IDisposable Tick(IScheduler self, int id)
|
|
|
{
|
|
|
- _s.Add(value);
|
|
|
+ var d = Disposable.Empty;
|
|
|
|
|
|
- _n++;
|
|
|
- if (_n == _parent._count)
|
|
|
+ var newId = 0;
|
|
|
+ lock (_gate)
|
|
|
{
|
|
|
- newWindow = true;
|
|
|
+ if (id != _windowId)
|
|
|
+ return d;
|
|
|
+
|
|
|
_n = 0;
|
|
|
newId = ++_windowId;
|
|
|
|
|
|
var res = _s;
|
|
|
_s = new List<TSource>();
|
|
|
base._observer.OnNext(res);
|
|
|
- }
|
|
|
|
|
|
- if (newWindow)
|
|
|
CreateTimer(newId);
|
|
|
+ }
|
|
|
+
|
|
|
+ return d;
|
|
|
}
|
|
|
- }
|
|
|
|
|
|
- public void OnError(Exception error)
|
|
|
- {
|
|
|
- lock (_gate)
|
|
|
+ public void OnNext(TSource value)
|
|
|
{
|
|
|
- _s.Clear();
|
|
|
- base._observer.OnError(error);
|
|
|
- base.Dispose();
|
|
|
+ var newWindow = false;
|
|
|
+ var newId = 0;
|
|
|
+
|
|
|
+ lock (_gate)
|
|
|
+ {
|
|
|
+ _s.Add(value);
|
|
|
+
|
|
|
+ _n++;
|
|
|
+ if (_n == _parent._count)
|
|
|
+ {
|
|
|
+ newWindow = true;
|
|
|
+ _n = 0;
|
|
|
+ newId = ++_windowId;
|
|
|
+
|
|
|
+ var res = _s;
|
|
|
+ _s = new List<TSource>();
|
|
|
+ base._observer.OnNext(res);
|
|
|
+ }
|
|
|
+
|
|
|
+ if (newWindow)
|
|
|
+ CreateTimer(newId);
|
|
|
+ }
|
|
|
}
|
|
|
- }
|
|
|
|
|
|
- public void OnCompleted()
|
|
|
- {
|
|
|
- lock (_gate)
|
|
|
+ public void OnError(Exception error)
|
|
|
{
|
|
|
- base._observer.OnNext(_s);
|
|
|
- base._observer.OnCompleted();
|
|
|
- base.Dispose();
|
|
|
+ lock (_gate)
|
|
|
+ {
|
|
|
+ _s.Clear();
|
|
|
+ base._observer.OnError(error);
|
|
|
+ base.Dispose();
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ public void OnCompleted()
|
|
|
+ {
|
|
|
+ lock (_gate)
|
|
|
+ {
|
|
|
+ base._observer.OnNext(_s);
|
|
|
+ base._observer.OnCompleted();
|
|
|
+ base.Dispose();
|
|
|
+ }
|
|
|
}
|
|
|
}
|
|
|
}
|
|
|
}
|
|
|
|
|
|
- internal sealed class Buffer<TSource, TBufferClosing> : Producer<IList<TSource>>
|
|
|
+ internal static class Buffer<TSource, TBufferClosing>
|
|
|
{
|
|
|
- private readonly IObservable<TSource> _source;
|
|
|
- private readonly Func<IObservable<TBufferClosing>> _bufferClosingSelector;
|
|
|
- private readonly IObservable<TBufferClosing> _bufferBoundaries;
|
|
|
-
|
|
|
- public Buffer(IObservable<TSource> source, Func<IObservable<TBufferClosing>> bufferClosingSelector)
|
|
|
+ internal sealed class Selector : Producer<IList<TSource>>
|
|
|
{
|
|
|
- _source = source;
|
|
|
- _bufferClosingSelector = bufferClosingSelector;
|
|
|
- }
|
|
|
+ private readonly IObservable<TSource> _source;
|
|
|
+ private readonly Func<IObservable<TBufferClosing>> _bufferClosingSelector;
|
|
|
|
|
|
- public Buffer(IObservable<TSource> source, IObservable<TBufferClosing> bufferBoundaries)
|
|
|
- {
|
|
|
- _source = source;
|
|
|
- _bufferBoundaries = bufferBoundaries;
|
|
|
- }
|
|
|
-
|
|
|
- protected override IDisposable Run(IObserver<IList<TSource>> observer, IDisposable cancel, Action<IDisposable> setSink)
|
|
|
- {
|
|
|
- if (_bufferClosingSelector != null)
|
|
|
+ public Selector(IObservable<TSource> source, Func<IObservable<TBufferClosing>> bufferClosingSelector)
|
|
|
{
|
|
|
- var sink = new _(this, observer, cancel);
|
|
|
- setSink(sink);
|
|
|
- return sink.Run();
|
|
|
+ _source = source;
|
|
|
+ _bufferClosingSelector = bufferClosingSelector;
|
|
|
}
|
|
|
- else
|
|
|
+
|
|
|
+ protected override IDisposable Run(IObserver<IList<TSource>> observer, IDisposable cancel, Action<IDisposable> setSink)
|
|
|
{
|
|
|
- var sink = new Beta(this, observer, cancel);
|
|
|
+ var sink = new _(this, observer, cancel);
|
|
|
setSink(sink);
|
|
|
- return sink.Run();
|
|
|
+ return sink.Run(_source);
|
|
|
}
|
|
|
- }
|
|
|
|
|
|
- class _ : Sink<IList<TSource>>, IObserver<TSource>
|
|
|
- {
|
|
|
- private readonly Buffer<TSource, TBufferClosing> _parent;
|
|
|
-
|
|
|
- public _(Buffer<TSource, TBufferClosing> parent, IObserver<IList<TSource>> observer, IDisposable cancel)
|
|
|
- : base(observer, cancel)
|
|
|
+ private sealed class _ : Sink<IList<TSource>>, IObserver<TSource>
|
|
|
{
|
|
|
- _parent = parent;
|
|
|
- }
|
|
|
+ private readonly object _gate = new object();
|
|
|
+ private readonly AsyncLock _bufferGate = new AsyncLock();
|
|
|
+ private readonly SerialDisposable _bufferClosingSubscription = new SerialDisposable();
|
|
|
|
|
|
- private IList<TSource> _buffer;
|
|
|
- private object _gate;
|
|
|
- private AsyncLock _bufferGate;
|
|
|
+ private readonly Func<IObservable<TBufferClosing>> _bufferClosingSelector;
|
|
|
|
|
|
- private SerialDisposable _m;
|
|
|
+ public _(Selector parent, IObserver<IList<TSource>> observer, IDisposable cancel)
|
|
|
+ : base(observer, cancel)
|
|
|
+ {
|
|
|
+ _bufferClosingSelector = parent._bufferClosingSelector;
|
|
|
+ }
|
|
|
|
|
|
- public IDisposable Run()
|
|
|
- {
|
|
|
- _buffer = new List<TSource>();
|
|
|
- _gate = new object();
|
|
|
- _bufferGate = new AsyncLock();
|
|
|
+ private IList<TSource> _buffer;
|
|
|
|
|
|
- _m = new SerialDisposable();
|
|
|
- var groupDisposable = new CompositeDisposable(2) { _m };
|
|
|
+ public IDisposable Run(IObservable<TSource> source)
|
|
|
+ {
|
|
|
+ _buffer = new List<TSource>();
|
|
|
|
|
|
- groupDisposable.Add(_parent._source.SubscribeSafe(this));
|
|
|
+ var groupDisposable = StableCompositeDisposable.Create(_bufferClosingSubscription, source.SubscribeSafe(this));
|
|
|
|
|
|
- _bufferGate.Wait(CreateBufferClose);
|
|
|
+ _bufferGate.Wait(CreateBufferClose);
|
|
|
|
|
|
- return groupDisposable;
|
|
|
- }
|
|
|
+ return groupDisposable;
|
|
|
+ }
|
|
|
|
|
|
- private void CreateBufferClose()
|
|
|
- {
|
|
|
- var bufferClose = default(IObservable<TBufferClosing>);
|
|
|
- try
|
|
|
+ private void CreateBufferClose()
|
|
|
{
|
|
|
- bufferClose = _parent._bufferClosingSelector();
|
|
|
+ var bufferClose = default(IObservable<TBufferClosing>);
|
|
|
+ try
|
|
|
+ {
|
|
|
+ bufferClose = _bufferClosingSelector();
|
|
|
+ }
|
|
|
+ catch (Exception exception)
|
|
|
+ {
|
|
|
+ lock (_gate)
|
|
|
+ {
|
|
|
+ base._observer.OnError(exception);
|
|
|
+ base.Dispose();
|
|
|
+ }
|
|
|
+ return;
|
|
|
+ }
|
|
|
+
|
|
|
+ var closingSubscription = new SingleAssignmentDisposable();
|
|
|
+ _bufferClosingSubscription.Disposable = closingSubscription;
|
|
|
+ closingSubscription.Disposable = bufferClose.SubscribeSafe(new BufferClosingObserver(this, closingSubscription));
|
|
|
}
|
|
|
- catch (Exception exception)
|
|
|
+
|
|
|
+ private void CloseBuffer(IDisposable closingSubscription)
|
|
|
{
|
|
|
+ closingSubscription.Dispose();
|
|
|
+
|
|
|
lock (_gate)
|
|
|
{
|
|
|
- base._observer.OnError(exception);
|
|
|
- base.Dispose();
|
|
|
+ var res = _buffer;
|
|
|
+ _buffer = new List<TSource>();
|
|
|
+ base._observer.OnNext(res);
|
|
|
}
|
|
|
- return;
|
|
|
- }
|
|
|
|
|
|
- var closingSubscription = new SingleAssignmentDisposable();
|
|
|
- _m.Disposable = closingSubscription;
|
|
|
- closingSubscription.Disposable = bufferClose.SubscribeSafe(new Omega(this, closingSubscription));
|
|
|
- }
|
|
|
-
|
|
|
- private void CloseBuffer(IDisposable closingSubscription)
|
|
|
- {
|
|
|
- closingSubscription.Dispose();
|
|
|
+ _bufferGate.Wait(CreateBufferClose);
|
|
|
+ }
|
|
|
|
|
|
- lock (_gate)
|
|
|
+ private sealed class BufferClosingObserver : IObserver<TBufferClosing>
|
|
|
{
|
|
|
- var res = _buffer;
|
|
|
- _buffer = new List<TSource>();
|
|
|
- base._observer.OnNext(res);
|
|
|
- }
|
|
|
+ private readonly _ _parent;
|
|
|
+ private readonly IDisposable _self;
|
|
|
|
|
|
- _bufferGate.Wait(CreateBufferClose);
|
|
|
- }
|
|
|
+ public BufferClosingObserver(_ parent, IDisposable self)
|
|
|
+ {
|
|
|
+ _parent = parent;
|
|
|
+ _self = self;
|
|
|
+ }
|
|
|
|
|
|
- class Omega : IObserver<TBufferClosing>
|
|
|
- {
|
|
|
- private readonly _ _parent;
|
|
|
- private readonly IDisposable _self;
|
|
|
+ public void OnNext(TBufferClosing value)
|
|
|
+ {
|
|
|
+ _parent.CloseBuffer(_self);
|
|
|
+ }
|
|
|
|
|
|
- public Omega(_ parent, IDisposable self)
|
|
|
- {
|
|
|
- _parent = parent;
|
|
|
- _self = self;
|
|
|
+ public void OnError(Exception error)
|
|
|
+ {
|
|
|
+ _parent.OnError(error);
|
|
|
+ }
|
|
|
+
|
|
|
+ public void OnCompleted()
|
|
|
+ {
|
|
|
+ _parent.CloseBuffer(_self);
|
|
|
+ }
|
|
|
}
|
|
|
|
|
|
- public void OnNext(TBufferClosing value)
|
|
|
+ public void OnNext(TSource value)
|
|
|
{
|
|
|
- _parent.CloseBuffer(_self);
|
|
|
+ lock (_gate)
|
|
|
+ {
|
|
|
+ _buffer.Add(value);
|
|
|
+ }
|
|
|
}
|
|
|
|
|
|
public void OnError(Exception error)
|
|
|
{
|
|
|
- _parent.OnError(error);
|
|
|
+ lock (_gate)
|
|
|
+ {
|
|
|
+ _buffer.Clear();
|
|
|
+ base._observer.OnError(error);
|
|
|
+ base.Dispose();
|
|
|
+ }
|
|
|
}
|
|
|
|
|
|
public void OnCompleted()
|
|
|
{
|
|
|
- _parent.CloseBuffer(_self);
|
|
|
+ lock (_gate)
|
|
|
+ {
|
|
|
+ base._observer.OnNext(_buffer);
|
|
|
+ base._observer.OnCompleted();
|
|
|
+ base.Dispose();
|
|
|
+ }
|
|
|
}
|
|
|
}
|
|
|
+ }
|
|
|
|
|
|
- public void OnNext(TSource value)
|
|
|
+ internal sealed class Boundaries : Producer<IList<TSource>>
|
|
|
+ {
|
|
|
+ private readonly IObservable<TSource> _source;
|
|
|
+ private readonly IObservable<TBufferClosing> _bufferBoundaries;
|
|
|
+
|
|
|
+ public Boundaries(IObservable<TSource> source, IObservable<TBufferClosing> bufferBoundaries)
|
|
|
{
|
|
|
- lock (_gate)
|
|
|
- {
|
|
|
- _buffer.Add(value);
|
|
|
- }
|
|
|
+ _source = source;
|
|
|
+ _bufferBoundaries = bufferBoundaries;
|
|
|
}
|
|
|
|
|
|
- public void OnError(Exception error)
|
|
|
+ protected override IDisposable Run(IObserver<IList<TSource>> observer, IDisposable cancel, Action<IDisposable> setSink)
|
|
|
{
|
|
|
- lock (_gate)
|
|
|
- {
|
|
|
- _buffer.Clear();
|
|
|
- base._observer.OnError(error);
|
|
|
- base.Dispose();
|
|
|
- }
|
|
|
+ var sink = new _(observer, cancel);
|
|
|
+ setSink(sink);
|
|
|
+ return sink.Run(this);
|
|
|
}
|
|
|
|
|
|
- public void OnCompleted()
|
|
|
+ private sealed class _ : Sink<IList<TSource>>, IObserver<TSource>
|
|
|
{
|
|
|
- lock (_gate)
|
|
|
+ private readonly object _gate = new object();
|
|
|
+
|
|
|
+ public _(IObserver<IList<TSource>> observer, IDisposable cancel)
|
|
|
+ : base(observer, cancel)
|
|
|
{
|
|
|
- base._observer.OnNext(_buffer);
|
|
|
- base._observer.OnCompleted();
|
|
|
- base.Dispose();
|
|
|
}
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- class Beta : Sink<IList<TSource>>, IObserver<TSource>
|
|
|
- {
|
|
|
- private readonly Buffer<TSource, TBufferClosing> _parent;
|
|
|
|
|
|
- public Beta(Buffer<TSource, TBufferClosing> parent, IObserver<IList<TSource>> observer, IDisposable cancel)
|
|
|
- : base(observer, cancel)
|
|
|
- {
|
|
|
- _parent = parent;
|
|
|
- }
|
|
|
+ private IList<TSource> _buffer;
|
|
|
|
|
|
- private IList<TSource> _buffer;
|
|
|
- private object _gate;
|
|
|
+ public IDisposable Run(Boundaries parent)
|
|
|
+ {
|
|
|
+ _buffer = new List<TSource>();
|
|
|
|
|
|
- private RefCountDisposable _refCountDisposable;
|
|
|
+ var sourceSubscription = parent._source.SubscribeSafe(this);
|
|
|
+ var boundariesSubscription = parent._bufferBoundaries.SubscribeSafe(new BufferClosingObserver(this));
|
|
|
|
|
|
- public IDisposable Run()
|
|
|
- {
|
|
|
- _buffer = new List<TSource>();
|
|
|
- _gate = new object();
|
|
|
+ return StableCompositeDisposable.Create(sourceSubscription, boundariesSubscription);
|
|
|
+ }
|
|
|
|
|
|
- var d = new CompositeDisposable(2);
|
|
|
- _refCountDisposable = new RefCountDisposable(d);
|
|
|
+ private sealed class BufferClosingObserver : IObserver<TBufferClosing>
|
|
|
+ {
|
|
|
+ private readonly _ _parent;
|
|
|
|
|
|
- d.Add(_parent._source.SubscribeSafe(this));
|
|
|
- d.Add(_parent._bufferBoundaries.SubscribeSafe(new Omega(this)));
|
|
|
+ public BufferClosingObserver(_ parent)
|
|
|
+ {
|
|
|
+ _parent = parent;
|
|
|
+ }
|
|
|
|
|
|
- return _refCountDisposable;
|
|
|
- }
|
|
|
+ public void OnNext(TBufferClosing value)
|
|
|
+ {
|
|
|
+ lock (_parent._gate)
|
|
|
+ {
|
|
|
+ var res = _parent._buffer;
|
|
|
+ _parent._buffer = new List<TSource>();
|
|
|
+ _parent._observer.OnNext(res);
|
|
|
+ }
|
|
|
+ }
|
|
|
|
|
|
- class Omega : IObserver<TBufferClosing>
|
|
|
- {
|
|
|
- private readonly Beta _parent;
|
|
|
+ public void OnError(Exception error)
|
|
|
+ {
|
|
|
+ _parent.OnError(error);
|
|
|
+ }
|
|
|
|
|
|
- public Omega(Beta parent)
|
|
|
- {
|
|
|
- _parent = parent;
|
|
|
+ public void OnCompleted()
|
|
|
+ {
|
|
|
+ _parent.OnCompleted();
|
|
|
+ }
|
|
|
}
|
|
|
|
|
|
- public void OnNext(TBufferClosing value)
|
|
|
+ public void OnNext(TSource value)
|
|
|
{
|
|
|
- lock (_parent._gate)
|
|
|
+ lock (_gate)
|
|
|
{
|
|
|
- var res = _parent._buffer;
|
|
|
- _parent._buffer = new List<TSource>();
|
|
|
- _parent._observer.OnNext(res);
|
|
|
+ _buffer.Add(value);
|
|
|
}
|
|
|
}
|
|
|
|
|
|
public void OnError(Exception error)
|
|
|
{
|
|
|
- _parent.OnError(error);
|
|
|
+ lock (_gate)
|
|
|
+ {
|
|
|
+ _buffer.Clear();
|
|
|
+ base._observer.OnError(error);
|
|
|
+ base.Dispose();
|
|
|
+ }
|
|
|
}
|
|
|
|
|
|
public void OnCompleted()
|
|
|
{
|
|
|
- _parent.OnCompleted();
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- public void OnNext(TSource value)
|
|
|
- {
|
|
|
- lock (_gate)
|
|
|
- {
|
|
|
- _buffer.Add(value);
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- public void OnError(Exception error)
|
|
|
- {
|
|
|
- lock (_gate)
|
|
|
- {
|
|
|
- _buffer.Clear();
|
|
|
- base._observer.OnError(error);
|
|
|
- base.Dispose();
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- public void OnCompleted()
|
|
|
- {
|
|
|
- lock (_gate)
|
|
|
- {
|
|
|
- base._observer.OnNext(_buffer);
|
|
|
- base._observer.OnCompleted();
|
|
|
- base.Dispose();
|
|
|
+ lock (_gate)
|
|
|
+ {
|
|
|
+ base._observer.OnNext(_buffer);
|
|
|
+ base._observer.OnCompleted();
|
|
|
+ base.Dispose();
|
|
|
+ }
|
|
|
}
|
|
|
}
|
|
|
}
|
Bart De Smet