Throttle<TSource, TThrottle>
using System.Reactive.Disposables;
using System.Runtime.CompilerServices;
namespace System.Reactive.Linq.ObservableImpl
{
[System.Runtime.CompilerServices.NullableContext(1)]
[System.Runtime.CompilerServices.Nullable(new byte[] {
0,
1,
1,
1,
1
})]
internal sealed class Throttle<[System.Runtime.CompilerServices.Nullable(2)] TSource, [System.Runtime.CompilerServices.Nullable(2)] TThrottle> : Producer<TSource, Throttle<TSource, TThrottle>._>
{
[System.Runtime.CompilerServices.Nullable(new byte[] {
0,
1
})]
internal sealed class _ : IdentitySink<TSource>
{
[System.Runtime.CompilerServices.Nullable(new byte[] {
0,
1
})]
private sealed class ThrottleObserver : SafeObserver<TThrottle>
{
[System.Runtime.CompilerServices.Nullable(new byte[] {
1,
0,
0
})]
private readonly _ _parent;
private readonly TSource _value;
private readonly ulong _currentid;
public ThrottleObserver([System.Runtime.CompilerServices.Nullable(new byte[] {
1,
0,
0
})] _ parent, TSource value, ulong currentid)
{
_parent = parent;
_value = value;
_currentid = currentid;
}
public override void OnNext(TThrottle value)
{
lock (_parent._gate) {
if (_parent._hasValue && _parent._id == _currentid)
_parent.ForwardOnNext(_value);
_parent._hasValue = false;
Dispose();
}
}
public override void OnError(Exception error)
{
lock (_parent._gate) {
_parent.ForwardOnError(error);
}
}
public override void OnCompleted()
{
lock (_parent._gate) {
if (_parent._hasValue && _parent._id == _currentid)
_parent.ForwardOnNext(_value);
_parent._hasValue = false;
Dispose();
}
}
}
private readonly object _gate = new object();
private readonly Func<TSource, IObservable<TThrottle>> _throttleSelector;
[System.Runtime.CompilerServices.Nullable(2)]
private TSource _value;
private bool _hasValue;
private SerialDisposableValue _serialCancelable;
private ulong _id;
public _(Throttle<TSource, TThrottle> parent, IObserver<TSource> observer)
: base(observer)
{
_throttleSelector = parent._throttleSelector;
}
protected override void Dispose(bool disposing)
{
if (disposing)
_serialCancelable.Dispose();
base.Dispose(disposing);
}
public override void OnNext(TSource value)
{
IObservable<TThrottle> source;
try {
source = _throttleSelector(value);
} catch (Exception error) {
lock (_gate) {
ForwardOnError(error);
}
return;
}
ulong id = default(ulong);
lock (_gate) {
_hasValue = true;
_value = value;
_id++;
id = _id;
}
_serialCancelable.Disposable = null;
ThrottleObserver throttleObserver = new ThrottleObserver(this, value, id);
throttleObserver.SetResource(ObservableExtensions.SubscribeSafe<TThrottle>(source, (IObserver<TThrottle>)throttleObserver));
_serialCancelable.Disposable = throttleObserver;
}
public override void OnError(Exception error)
{
_serialCancelable.Dispose();
lock (_gate) {
ForwardOnError(error);
_hasValue = false;
_id++;
}
}
public override void OnCompleted()
{
_serialCancelable.Dispose();
lock (_gate) {
if (_hasValue)
ForwardOnNext(_value);
ForwardOnCompleted();
_hasValue = false;
_id++;
}
}
}
private readonly IObservable<TSource> _source;
private readonly Func<TSource, IObservable<TThrottle>> _throttleSelector;
public Throttle(IObservable<TSource> source, Func<TSource, IObservable<TThrottle>> throttleSelector)
{
_source = source;
_throttleSelector = throttleSelector;
}
[return: System.Runtime.CompilerServices.Nullable(new byte[] {
1,
0,
0
})]
protected override _ CreateSink(IObserver<TSource> observer)
{
return new _(this, observer);
}
protected override void Run([System.Runtime.CompilerServices.Nullable(new byte[] {
1,
0,
0
})] _ sink)
{
sink.Run(_source);
}
}
}