using System.Reactive.Concurrency; 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 })] internal sealed class Sample<[System.Runtime.CompilerServices.Nullable(2)] TSource> : Producer<TSource, Sample<TSource>._> { [System.Runtime.CompilerServices.Nullable(new byte[] { 0, 1 })] internal sealed class _ : IdentitySink<TSource> { private readonly object _gate = new object(); private SingleAssignmentDisposableValue _sourceDisposable; private bool _hasValue; [System.Runtime.CompilerServices.Nullable(2)] private TSource _value; private bool _atEnd; public _(IObserver<TSource> observer) : base(observer) { } public void Run(Sample<TSource> parent) { _sourceDisposable.Disposable = ObservableExtensions.SubscribeSafe<TSource>(parent._source, (IObserver<TSource>)this); SetUpstream(parent._scheduler.SchedulePeriodic(parent._interval, Tick)); } protected override void Dispose(bool disposing) { if (disposing) _sourceDisposable.Dispose(); base.Dispose(disposing); } private void Tick() { lock (_gate) { if (_hasValue) { _hasValue = false; ForwardOnNext(_value); } if (_atEnd) ForwardOnCompleted(); } } public override void OnNext(TSource value) { lock (_gate) { _hasValue = true; _value = value; } } public override void OnError(Exception error) { lock (_gate) { ForwardOnError(error); } } public override void OnCompleted() { lock (_gate) { _atEnd = true; _sourceDisposable.Dispose(); } } } private readonly IObservable<TSource> _source; private readonly TimeSpan _interval; private readonly IScheduler _scheduler; public Sample(IObservable<TSource> source, TimeSpan interval, IScheduler scheduler) { _source = source; _interval = interval; _scheduler = scheduler; } [return: System.Runtime.CompilerServices.Nullable(new byte[] { 1, 0 })] protected override _ CreateSink(IObserver<TSource> observer) { return new _(observer); } protected override void Run([System.Runtime.CompilerServices.Nullable(new byte[] { 1, 0 })] _ sink) { sink.Run(this); } } }