// Copyright (c) Microsoft Open Technologies, Inc. All rights reserved. See License.txt in the project root for license information. #if !NO_PERF using System; using System.Collections.Generic; using System.Reactive.Disposables; namespace System.Reactive.Linq.Observαble { class If : Producer, IEvaluatableObservable { private readonly Func _condition; private readonly IObservable _thenSource; private readonly IObservable _elseSource; public If(Func condition, IObservable thenSource, IObservable elseSource) { _condition = condition; _thenSource = thenSource; _elseSource = elseSource; } public IObservable Eval() { return _condition() ? _thenSource : _elseSource; } protected override IDisposable Run(IObserver observer, IDisposable cancel, Action setSink) { var sink = new _(this, observer, cancel); setSink(sink); return sink.Run(); } class _ : Sink, IObserver { private readonly If _parent; public _(If parent, IObserver observer, IDisposable cancel) : base(observer, cancel) { _parent = parent; } public IDisposable Run() { var result = default(IObservable); try { result = _parent.Eval(); } catch (Exception exception) { base._observer.OnError(exception); base.Dispose(); return Disposable.Empty; } return result.SubscribeSafe(this); } public void OnNext(TResult value) { base._observer.OnNext(value); } public void OnError(Exception error) { base._observer.OnError(error); base.Dispose(); } public void OnCompleted() { base._observer.OnCompleted(); base.Dispose(); } } } } #endif