// Copyright (c) Microsoft Open Technologies, Inc. All rights reserved. See License.txt in the project root for license information. #if !NO_PERF using System; namespace System.Reactive.Linq.Observαble { class All : Producer { private readonly IObservable _source; private readonly Func _predicate; public All(IObservable source, Func predicate) { _source = source; _predicate = predicate; } protected override IDisposable Run(IObserver observer, IDisposable cancel, Action setSink) { var sink = new _(this, observer, cancel); setSink(sink); return _source.SubscribeSafe(sink); } class _ : Sink, IObserver { private readonly All _parent; public _(All parent, IObserver observer, IDisposable cancel) : base(observer, cancel) { _parent = parent; } public void OnNext(TSource value) { var res = false; try { res = _parent._predicate(value); } catch (Exception ex) { base._observer.OnError(ex); base.Dispose(); return; } if (!res) { base._observer.OnNext(false); base._observer.OnCompleted(); base.Dispose(); } } public void OnError(Exception error) { base._observer.OnError(error); base.Dispose(); } public void OnCompleted() { base._observer.OnNext(true); base._observer.OnCompleted(); base.Dispose(); } } } } #endif