// 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 Case : Producer, IEvaluatableObservable { private readonly Func _selector; private readonly IDictionary> _sources; private readonly IObservable _defaultSource; public Case(Func selector, IDictionary> sources, IObservable defaultSource) { _selector = selector; _sources = sources; _defaultSource = defaultSource; } public IObservable Eval() { var res = default(IObservable); if (_sources.TryGetValue(_selector(), out res)) return res; return _defaultSource; } 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 Case _parent; public _(Case 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