using System.Reactive.Disposables; namespace System.Reactive.Linq.ObservableImpl { internal sealed class TakeUntil<TSource, TOther> : Producer<TSource, TakeUntil<TSource, TOther>._> { internal sealed class _ : IdentitySink<TSource> { private sealed class OtherObserver : IObserver<TOther> { private readonly _ _parent; public OtherObserver(_ parent) { _parent = parent; } public void OnCompleted() { Disposable.TryDispose(ref _parent._otherDisposable); } public void OnError(Exception error) { HalfSerializer.ForwardOnError<TSource>((ISink<TSource>)_parent, error, ref _parent._halfSerializer, ref _parent._error); } public void OnNext(TOther value) { HalfSerializer.ForwardOnCompleted<TSource>((ISink<TSource>)_parent, ref _parent._halfSerializer, ref _parent._error); } } private IDisposable _otherDisposable; private int _halfSerializer; private Exception _error; public _(IObserver<TSource> observer) : base(observer) { } public void Run(TakeUntil<TSource, TOther> parent) { Disposable.SetSingle(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) { 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() { HalfSerializer.ForwardOnCompleted<TSource>((ISink<TSource>)this, ref _halfSerializer, ref _error); } } private readonly IObservable<TSource> _source; private readonly IObservable<TOther> _other; public TakeUntil(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); } } }