// 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; namespace System.Reactive.Linq.ObservableImpl { class Min : Producer { private readonly IObservable _source; private readonly IComparer _comparer; public Min(IObservable source, IComparer comparer) { _source = source; _comparer = comparer; } protected override IDisposable Run(IObserver observer, IDisposable cancel, Action setSink) { // LINQ to Objects makes this distinction in order to make [Min|Max] of an empty collection of reference type objects equal to null. if (default(TSource) == null) { var sink = new _(this, observer, cancel); setSink(sink); return _source.SubscribeSafe(sink); } else { var sink = new Delta(this, observer, cancel); setSink(sink); return _source.SubscribeSafe(sink); } } class Delta : Sink, IObserver { private readonly Min _parent; private bool _hasValue; private TSource _lastValue; public Delta(Min parent, IObserver observer, IDisposable cancel) : base(observer, cancel) { _parent = parent; _hasValue = false; _lastValue = default(TSource); } public void OnNext(TSource value) { if (_hasValue) { var comparison = 0; try { comparison = _parent._comparer.Compare(value, _lastValue); } catch (Exception ex) { base._observer.OnError(ex); base.Dispose(); return; } if (comparison < 0) { _lastValue = value; } } else { _hasValue = true; _lastValue = value; } } public void OnError(Exception error) { base._observer.OnError(error); base.Dispose(); } public void OnCompleted() { if (!_hasValue) { base._observer.OnError(new InvalidOperationException(Strings_Linq.NO_ELEMENTS)); } else { base._observer.OnNext(_lastValue); base._observer.OnCompleted(); } base.Dispose(); } } class _ : Sink, IObserver { private readonly Min _parent; private TSource _lastValue; public _(Min parent, IObserver observer, IDisposable cancel) : base(observer, cancel) { _parent = parent; _lastValue = default(TSource); } public void OnNext(TSource value) { if (value != null) { if (_lastValue == null) { _lastValue = value; } else { var comparison = 0; try { comparison = _parent._comparer.Compare(value, _lastValue); } catch (Exception ex) { base._observer.OnError(ex); base.Dispose(); return; } if (comparison < 0) { _lastValue = value; } } } } public void OnError(Exception error) { base._observer.OnError(error); base.Dispose(); } public void OnCompleted() { base._observer.OnNext(_lastValue); base._observer.OnCompleted(); base.Dispose(); } } } class MinDouble : Producer { private readonly IObservable _source; public MinDouble(IObservable source) { _source = source; } protected override IDisposable Run(IObserver observer, IDisposable cancel, Action setSink) { var sink = new _(observer, cancel); setSink(sink); return _source.SubscribeSafe(sink); } class _ : Sink, IObserver { private bool _hasValue; private double _lastValue; public _(IObserver observer, IDisposable cancel) : base(observer, cancel) { _hasValue = false; _lastValue = default(double); } public void OnNext(double value) { if (_hasValue) { if (value < _lastValue || double.IsNaN(value)) { _lastValue = value; } } else { _lastValue = value; _hasValue = true; } } public void OnError(Exception error) { base._observer.OnError(error); base.Dispose(); } public void OnCompleted() { if (!_hasValue) { base._observer.OnError(new InvalidOperationException(Strings_Linq.NO_ELEMENTS)); } else { base._observer.OnNext(_lastValue); base._observer.OnCompleted(); } base.Dispose(); } } } class MinSingle : Producer { private readonly IObservable _source; public MinSingle(IObservable source) { _source = source; } protected override IDisposable Run(IObserver observer, IDisposable cancel, Action setSink) { var sink = new _(observer, cancel); setSink(sink); return _source.SubscribeSafe(sink); } class _ : Sink, IObserver { private bool _hasValue; private float _lastValue; public _(IObserver observer, IDisposable cancel) : base(observer, cancel) { _hasValue = false; _lastValue = default(float); } public void OnNext(float value) { if (_hasValue) { if (value < _lastValue || float.IsNaN(value)) { _lastValue = value; } } else { _lastValue = value; _hasValue = true; } } public void OnError(Exception error) { base._observer.OnError(error); base.Dispose(); } public void OnCompleted() { if (!_hasValue) { base._observer.OnError(new InvalidOperationException(Strings_Linq.NO_ELEMENTS)); } else { base._observer.OnNext(_lastValue); base._observer.OnCompleted(); } base.Dispose(); } } } class MinDecimal : Producer { private readonly IObservable _source; public MinDecimal(IObservable source) { _source = source; } protected override IDisposable Run(IObserver observer, IDisposable cancel, Action setSink) { var sink = new _(observer, cancel); setSink(sink); return _source.SubscribeSafe(sink); } class _ : Sink, IObserver { private bool _hasValue; private decimal _lastValue; public _(IObserver observer, IDisposable cancel) : base(observer, cancel) { _hasValue = false; _lastValue = default(decimal); } public void OnNext(decimal value) { if (_hasValue) { if (value < _lastValue) { _lastValue = value; } } else { _lastValue = value; _hasValue = true; } } public void OnError(Exception error) { base._observer.OnError(error); base.Dispose(); } public void OnCompleted() { if (!_hasValue) { base._observer.OnError(new InvalidOperationException(Strings_Linq.NO_ELEMENTS)); } else { base._observer.OnNext(_lastValue); base._observer.OnCompleted(); } base.Dispose(); } } } class MinInt32 : Producer { private readonly IObservable _source; public MinInt32(IObservable source) { _source = source; } protected override IDisposable Run(IObserver observer, IDisposable cancel, Action setSink) { var sink = new _(observer, cancel); setSink(sink); return _source.SubscribeSafe(sink); } class _ : Sink, IObserver { private bool _hasValue; private int _lastValue; public _(IObserver observer, IDisposable cancel) : base(observer, cancel) { _hasValue = false; _lastValue = default(int); } public void OnNext(int value) { if (_hasValue) { if (value < _lastValue) { _lastValue = value; } } else { _lastValue = value; _hasValue = true; } } public void OnError(Exception error) { base._observer.OnError(error); base.Dispose(); } public void OnCompleted() { if (!_hasValue) { base._observer.OnError(new InvalidOperationException(Strings_Linq.NO_ELEMENTS)); } else { base._observer.OnNext(_lastValue); base._observer.OnCompleted(); } base.Dispose(); } } } class MinInt64 : Producer { private readonly IObservable _source; public MinInt64(IObservable source) { _source = source; } protected override IDisposable Run(IObserver observer, IDisposable cancel, Action setSink) { var sink = new _(observer, cancel); setSink(sink); return _source.SubscribeSafe(sink); } class _ : Sink, IObserver { private bool _hasValue; private long _lastValue; public _(IObserver observer, IDisposable cancel) : base(observer, cancel) { _hasValue = false; _lastValue = default(long); } public void OnNext(long value) { if (_hasValue) { if (value < _lastValue) { _lastValue = value; } } else { _lastValue = value; _hasValue = true; } } public void OnError(Exception error) { base._observer.OnError(error); base.Dispose(); } public void OnCompleted() { if (!_hasValue) { base._observer.OnError(new InvalidOperationException(Strings_Linq.NO_ELEMENTS)); } else { base._observer.OnNext(_lastValue); base._observer.OnCompleted(); } base.Dispose(); } } } class MinDoubleNullable : Producer { private readonly IObservable _source; public MinDoubleNullable(IObservable source) { _source = source; } protected override IDisposable Run(IObserver observer, IDisposable cancel, Action setSink) { var sink = new _(observer, cancel); setSink(sink); return _source.SubscribeSafe(sink); } class _ : Sink, IObserver { private double? _lastValue; public _(IObserver observer, IDisposable cancel) : base(observer, cancel) { _lastValue = default(double?); } public void OnNext(double? value) { if (!value.HasValue) return; if (_lastValue.HasValue) { if (value < _lastValue || double.IsNaN((double)value)) { _lastValue = value; } } else { _lastValue = value; } } public void OnError(Exception error) { base._observer.OnError(error); base.Dispose(); } public void OnCompleted() { base._observer.OnNext(_lastValue); base._observer.OnCompleted(); base.Dispose(); } } } class MinSingleNullable : Producer { private readonly IObservable _source; public MinSingleNullable(IObservable source) { _source = source; } protected override IDisposable Run(IObserver observer, IDisposable cancel, Action setSink) { var sink = new _(observer, cancel); setSink(sink); return _source.SubscribeSafe(sink); } class _ : Sink, IObserver { private float? _lastValue; public _(IObserver observer, IDisposable cancel) : base(observer, cancel) { _lastValue = default(float?); } public void OnNext(float? value) { if (!value.HasValue) return; if (_lastValue.HasValue) { if (value < _lastValue || float.IsNaN((float)value)) { _lastValue = value; } } else { _lastValue = value; } } public void OnError(Exception error) { base._observer.OnError(error); base.Dispose(); } public void OnCompleted() { base._observer.OnNext(_lastValue); base._observer.OnCompleted(); base.Dispose(); } } } class MinDecimalNullable : Producer { private readonly IObservable _source; public MinDecimalNullable(IObservable source) { _source = source; } protected override IDisposable Run(IObserver observer, IDisposable cancel, Action setSink) { var sink = new _(observer, cancel); setSink(sink); return _source.SubscribeSafe(sink); } class _ : Sink, IObserver { private decimal? _lastValue; public _(IObserver observer, IDisposable cancel) : base(observer, cancel) { _lastValue = default(decimal?); } public void OnNext(decimal? value) { if (!value.HasValue) return; if (_lastValue.HasValue) { if (value < _lastValue) { _lastValue = value; } } else { _lastValue = value; } } public void OnError(Exception error) { base._observer.OnError(error); base.Dispose(); } public void OnCompleted() { base._observer.OnNext(_lastValue); base._observer.OnCompleted(); base.Dispose(); } } } class MinInt32Nullable : Producer { private readonly IObservable _source; public MinInt32Nullable(IObservable source) { _source = source; } protected override IDisposable Run(IObserver observer, IDisposable cancel, Action setSink) { var sink = new _(observer, cancel); setSink(sink); return _source.SubscribeSafe(sink); } class _ : Sink, IObserver { private int? _lastValue; public _(IObserver observer, IDisposable cancel) : base(observer, cancel) { _lastValue = default(int?); } public void OnNext(int? value) { if (!value.HasValue) return; if (_lastValue.HasValue) { if (value < _lastValue) { _lastValue = value; } } else { _lastValue = value; } } public void OnError(Exception error) { base._observer.OnError(error); base.Dispose(); } public void OnCompleted() { base._observer.OnNext(_lastValue); base._observer.OnCompleted(); base.Dispose(); } } } class MinInt64Nullable : Producer { private readonly IObservable _source; public MinInt64Nullable(IObservable source) { _source = source; } protected override IDisposable Run(IObserver observer, IDisposable cancel, Action setSink) { var sink = new _(observer, cancel); setSink(sink); return _source.SubscribeSafe(sink); } class _ : Sink, IObserver { private long? _lastValue; public _(IObserver observer, IDisposable cancel) : base(observer, cancel) { _lastValue = default(long?); } public void OnNext(long? value) { if (!value.HasValue) return; if (_lastValue.HasValue) { if (value < _lastValue) { _lastValue = value; } } else { _lastValue = value; } } public void OnError(Exception error) { base._observer.OnError(error); base.Dispose(); } public void OnCompleted() { base._observer.OnNext(_lastValue); base._observer.OnCompleted(); base.Dispose(); } } } } #endif