using System.Reactive.Disposables; namespace System.Reactive.Linq.ObservableImpl { internal sealed class SkipUntil<TSource, TOther> : Producer<TSource, SkipUntil<TSource, TOther>._> { internal sealed class _ : IdentitySink<TSource> { private sealed class OtherObserver : IObserver<TOther>, IDisposable { private readonly _ _parent; public OtherObserver(_ parent) { _parent = parent; } public void Dispose() { if (!Disposable.GetIsDisposed(ref _parent._otherDisposable)) Disposable.TryDispose(ref _parent._otherDisposable); } public void OnCompleted() { Dispose(); } public void OnError(Exception error) { HalfSerializer.ForwardOnError<TSource>((ISink<TSource>)_parent, error, ref _parent._halfSerializer, ref _parent._error); } public void OnNext(TOther value) { _parent.OtherComplete(); Dispose(); } } private IDisposable _otherDisposable; private bool _forward; private int _halfSerializer; private Exception _error; public _(IObserver<TSource> observer) : base(observer) { } public void Run(SkipUntil<TSource, TOther> parent) { Disposable.TrySetSingle(ref _otherDisposable, parent._other.Subscribe(new OtherObserver(this))); Run(parent._source); } protected override void Dispose(bool disposing) { if (disposing && !Disposable.GetIsDisposed(ref _otherDisposable)) Disposable.TryDispose(ref _otherDisposable); base.Dispose(disposing); } public override void OnNext(TSource value) { if (_forward) HalfSerializer.ForwardOnNext<TSource>((ISink<TSource>)this, value, ref _halfSerializer, ref _error); } public override void OnError(Exception ex) { HalfSerializer.ForwardOnError<TSource>((ISink<TSource>)this, ex, ref _halfSerializer, ref _error); } public override void OnCompleted() { if (_forward) HalfSerializer.ForwardOnCompleted<TSource>((ISink<TSource>)this, ref _halfSerializer, ref _error); else DisposeUpstream(); } private void OtherComplete() { _forward = true; } } private readonly IObservable<TSource> _source; private readonly IObservable<TOther> _other; public SkipUntil(IObservable<TSource> source, IObservable<TOther> other) { _source = source; _other = other; } protected override _ CreateSink(IObserver<TSource> observer) { return new _(observer); } protected override void Run(_ sink) { sink.Run(this); } } }