Sample<TSource, TSample>
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 Sample<[System.Runtime.CompilerServices.Nullable(2)] TSource, [System.Runtime.CompilerServices.Nullable(2)] TSample> : Producer<TSource, Sample<TSource, TSample>._>
{
[System.Runtime.CompilerServices.Nullable(new byte[] {
0,
1
})]
internal sealed class _ : IdentitySink<TSource>
{
[System.Runtime.CompilerServices.Nullable(0)]
private sealed class SampleObserver : IObserver<TSample>
{
[System.Runtime.CompilerServices.Nullable(new byte[] {
1,
0,
0
})]
private readonly _ _parent;
public SampleObserver([System.Runtime.CompilerServices.Nullable(new byte[] {
1,
0,
0
})] _ parent)
{
_parent = parent;
}
public void OnNext(TSample value)
{
lock (_parent._gate) {
if (_parent._hasValue) {
_parent._hasValue = false;
_parent.ForwardOnNext(_parent._value);
}
if (_parent._sourceAtEnd)
_parent.ForwardOnCompleted();
}
}
public void OnError(Exception error)
{
lock (_parent._gate) {
_parent.ForwardOnError(error);
}
}
public void OnCompleted()
{
lock (_parent._gate) {
_parent._samplerAtEnd = true;
if (_parent._hasValue) {
_parent._hasValue = false;
_parent.ForwardOnNext(_parent._value);
}
if (_parent._sourceAtEnd)
_parent.ForwardOnCompleted();
else
_parent._samplerDisposable.Dispose();
}
}
}
private readonly object _gate = new object();
private SingleAssignmentDisposableValue _sourceDisposable;
private SingleAssignmentDisposableValue _samplerDisposable;
private bool _hasValue;
[System.Runtime.CompilerServices.Nullable(2)]
private TSource _value;
private bool _sourceAtEnd;
private bool _samplerAtEnd;
public _(IObserver<TSource> observer)
: base(observer)
{
}
public void Run(Sample<TSource, TSample> parent)
{
_sourceDisposable.Disposable = ObservableExtensions.SubscribeSafe<TSource>(parent._source, (IObserver<TSource>)this);
_samplerDisposable.Disposable = ObservableExtensions.SubscribeSafe<TSample>(parent._sampler, (IObserver<TSample>)new SampleObserver(this));
}
protected override void Dispose(bool disposing)
{
if (disposing) {
_sourceDisposable.Dispose();
_samplerDisposable.Dispose();
}
base.Dispose(disposing);
}
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) {
_sourceAtEnd = true;
if (_samplerAtEnd)
ForwardOnCompleted();
else
_sourceDisposable.Dispose();
}
}
}
private readonly IObservable<TSource> _source;
private readonly IObservable<TSample> _sampler;
public Sample(IObservable<TSource> source, IObservable<TSample> sampler)
{
_source = source;
_sampler = sampler;
}
[return: System.Runtime.CompilerServices.Nullable(new byte[] {
1,
0,
0
})]
protected override _ CreateSink(IObserver<TSource> observer)
{
return new _(observer);
}
protected override void Run([System.Runtime.CompilerServices.Nullable(new byte[] {
1,
0,
0
})] _ sink)
{
sink.Run(this);
}
}
}