TakeLastBuffer<TSource>
using System.Collections.Generic;
using System.Reactive.Concurrency;
namespace System.Reactive.Linq.ObservableImpl
{
internal static class TakeLastBuffer<TSource>
{
internal sealed class Count : Producer<IList<TSource>, Count._>
{
internal sealed class _ : Sink<IList<TSource>>, IObserver<TSource>
{
private readonly int _count;
private Queue<TSource> _queue;
public _(int count, IObserver<IList<TSource>> observer, IDisposable cancel)
: base(observer, cancel)
{
_count = count;
_queue = new Queue<TSource>();
}
public void OnNext(TSource value)
{
_queue.Enqueue(value);
if (_queue.Count > _count)
_queue.Dequeue();
}
public void OnError(Exception error)
{
_observer.OnError(error);
base.Dispose();
}
public void OnCompleted()
{
List<TSource> list = new List<TSource>(_queue.Count);
while (_queue.Count > 0) {
list.Add(_queue.Dequeue());
}
_observer.OnNext(list);
_observer.OnCompleted();
base.Dispose();
}
}
private readonly IObservable<TSource> _source;
private readonly int _count;
public Count(IObservable<TSource> source, int count)
{
_source = source;
_count = count;
}
protected override _ CreateSink(IObserver<IList<TSource>> observer, IDisposable cancel)
{
return new _(_count, observer, cancel);
}
protected override IDisposable Run(_ sink)
{
return ObservableExtensions.SubscribeSafe<TSource>(_source, (IObserver<TSource>)sink);
}
}
internal sealed class Time : Producer<IList<TSource>, Time._>
{
internal sealed class _ : Sink<IList<TSource>>, IObserver<TSource>
{
private readonly TimeSpan _duration;
private Queue<System.Reactive.TimeInterval<TSource>> _queue;
private IStopwatch _watch;
public _(TimeSpan duration, IObserver<IList<TSource>> observer, IDisposable cancel)
: base(observer, cancel)
{
_duration = duration;
_queue = new Queue<System.Reactive.TimeInterval<TSource>>();
}
public IDisposable Run(Time parent)
{
_watch = parent._scheduler.StartStopwatch();
return ObservableExtensions.SubscribeSafe<TSource>(parent._source, (IObserver<TSource>)this);
}
public void OnNext(TSource value)
{
TimeSpan elapsed = _watch.Elapsed;
_queue.Enqueue(new System.Reactive.TimeInterval<TSource>(value, elapsed));
Trim(elapsed);
}
public void OnError(Exception error)
{
_observer.OnError(error);
base.Dispose();
}
public void OnCompleted()
{
TimeSpan elapsed = _watch.Elapsed;
Trim(elapsed);
List<TSource> list = new List<TSource>(_queue.Count);
while (_queue.Count > 0) {
list.Add(_queue.Dequeue().Value);
}
_observer.OnNext(list);
_observer.OnCompleted();
base.Dispose();
}
private void Trim(TimeSpan now)
{
while (_queue.Count > 0 && now - _queue.Peek().Interval >= _duration) {
_queue.Dequeue();
}
}
}
private readonly IObservable<TSource> _source;
private readonly TimeSpan _duration;
private readonly IScheduler _scheduler;
public Time(IObservable<TSource> source, TimeSpan duration, IScheduler scheduler)
{
_source = source;
_duration = duration;
_scheduler = scheduler;
}
protected override _ CreateSink(IObserver<IList<TSource>> observer, IDisposable cancel)
{
return new _(_duration, observer, cancel);
}
protected override IDisposable Run(_ sink)
{
return sink.Run(this);
}
}
}
}