// Licensed to the .NET Foundation under one or more agreements. // The .NET Foundation licenses this file to you under the Apache 2.0 License. // See the LICENSE file in the project root for more information. using System.Collections.Generic; namespace System.Reactive.Linq.ObservableImpl { internal sealed class Max : Producer._> { private readonly IObservable _source; private readonly IComparer _comparer; public Max(IObservable source, IComparer comparer) { _source = source; _comparer = comparer; } protected override _ CreateSink(IObserver observer) => default(TSource) == null ? (_)new Null(_comparer, observer) : new NonNull(_comparer, observer); protected override void Run(_ sink) => sink.Run(_source); internal abstract class _ : IdentitySink { protected readonly IComparer _comparer; public _(IComparer comparer, IObserver observer) : base(observer) { _comparer = comparer; } } private sealed class NonNull : _ { private bool _hasValue; private TSource _lastValue; public NonNull(IComparer comparer, IObserver observer) : base(comparer, observer) { _hasValue = false; _lastValue = default(TSource); } public override void OnNext(TSource value) { if (_hasValue) { var comparison = 0; try { comparison = _comparer.Compare(value, _lastValue); } catch (Exception ex) { ForwardOnError(ex); return; } if (comparison > 0) { _lastValue = value; } } else { _hasValue = true; _lastValue = value; } } public override void OnCompleted() { if (!_hasValue) { ForwardOnError(new InvalidOperationException(Strings_Linq.NO_ELEMENTS)); } else { ForwardOnNext(_lastValue); ForwardOnCompleted(); } } } private sealed class Null : _ { private TSource _lastValue; public Null(IComparer comparer, IObserver observer) : base(comparer, observer) { _lastValue = default(TSource); } public override void OnNext(TSource value) { if (value != null) { if (_lastValue == null) { _lastValue = value; } else { var comparison = 0; try { comparison = _comparer.Compare(value, _lastValue); } catch (Exception ex) { ForwardOnError(ex); return; } if (comparison > 0) { _lastValue = value; } } } } public override void OnError(Exception error) { ForwardOnError(error); } public override void OnCompleted() { ForwardOnNext(_lastValue); ForwardOnCompleted(); } } } internal sealed class MaxDouble : Producer { private readonly IObservable _source; public MaxDouble(IObservable source) { _source = source; } protected override _ CreateSink(IObserver observer) => new _(observer); protected override void Run(_ sink) => sink.Run(_source); internal sealed class _ : IdentitySink { private bool _hasValue; private double _lastValue; public _(IObserver observer) : base(observer) { _hasValue = false; _lastValue = default(double); } public override void OnNext(double value) { if (_hasValue) { if (value > _lastValue || double.IsNaN(value)) { _lastValue = value; } } else { _lastValue = value; _hasValue = true; } } public override void OnCompleted() { if (!_hasValue) { ForwardOnError(new InvalidOperationException(Strings_Linq.NO_ELEMENTS)); } else { ForwardOnNext(_lastValue); ForwardOnCompleted(); } } } } internal sealed class MaxSingle : Producer { private readonly IObservable _source; public MaxSingle(IObservable source) { _source = source; } protected override _ CreateSink(IObserver observer) => new _(observer); protected override void Run(_ sink) => sink.Run(_source); internal sealed class _ : IdentitySink { private bool _hasValue; private float _lastValue; public _(IObserver observer) : base(observer) { _hasValue = false; _lastValue = default(float); } public override void OnNext(float value) { if (_hasValue) { if (value > _lastValue || float.IsNaN(value)) { _lastValue = value; } } else { _lastValue = value; _hasValue = true; } } public override void OnCompleted() { if (!_hasValue) { ForwardOnError(new InvalidOperationException(Strings_Linq.NO_ELEMENTS)); } else { ForwardOnNext(_lastValue); ForwardOnCompleted(); } } } } internal sealed class MaxDecimal : Producer { private readonly IObservable _source; public MaxDecimal(IObservable source) { _source = source; } protected override _ CreateSink(IObserver observer) => new _(observer); protected override void Run(_ sink) => sink.Run(_source); internal sealed class _ : IdentitySink { private bool _hasValue; private decimal _lastValue; public _(IObserver observer) : base(observer) { _hasValue = false; _lastValue = default(decimal); } public override void OnNext(decimal value) { if (_hasValue) { if (value > _lastValue) { _lastValue = value; } } else { _lastValue = value; _hasValue = true; } } public override void OnCompleted() { if (!_hasValue) { ForwardOnError(new InvalidOperationException(Strings_Linq.NO_ELEMENTS)); } else { ForwardOnNext(_lastValue); ForwardOnCompleted(); } } } } internal sealed class MaxInt32 : Producer { private readonly IObservable _source; public MaxInt32(IObservable source) { _source = source; } protected override _ CreateSink(IObserver observer) => new _(observer); protected override void Run(_ sink) => sink.Run(_source); internal sealed class _ : IdentitySink { private bool _hasValue; private int _lastValue; public _(IObserver observer) : base(observer) { _hasValue = false; _lastValue = default(int); } public override void OnNext(int value) { if (_hasValue) { if (value > _lastValue) { _lastValue = value; } } else { _lastValue = value; _hasValue = true; } } public override void OnCompleted() { if (!_hasValue) { ForwardOnError(new InvalidOperationException(Strings_Linq.NO_ELEMENTS)); } else { ForwardOnNext(_lastValue); ForwardOnCompleted(); } } } } internal sealed class MaxInt64 : Producer { private readonly IObservable _source; public MaxInt64(IObservable source) { _source = source; } protected override _ CreateSink(IObserver observer) => new _(observer); protected override void Run(_ sink) => sink.Run(_source); internal sealed class _ : IdentitySink { private bool _hasValue; private long _lastValue; public _(IObserver observer) : base(observer) { _hasValue = false; _lastValue = default(long); } public override void OnNext(long value) { if (_hasValue) { if (value > _lastValue) { _lastValue = value; } } else { _lastValue = value; _hasValue = true; } } public override void OnCompleted() { if (!_hasValue) { ForwardOnError(new InvalidOperationException(Strings_Linq.NO_ELEMENTS)); } else { ForwardOnNext(_lastValue); ForwardOnCompleted(); } } } } internal sealed class MaxDoubleNullable : Producer { private readonly IObservable _source; public MaxDoubleNullable(IObservable source) { _source = source; } protected override _ CreateSink(IObserver observer) => new _(observer); protected override void Run(_ sink) => sink.Run(_source); internal sealed class _ : IdentitySink { private double? _lastValue; public _(IObserver observer) : base(observer) { _lastValue = default(double?); } public override void OnNext(double? value) { if (!value.HasValue) return; if (_lastValue.HasValue) { if (value > _lastValue || double.IsNaN((double)value)) { _lastValue = value; } } else { _lastValue = value; } } public override void OnCompleted() { ForwardOnNext(_lastValue); ForwardOnCompleted(); } } } internal sealed class MaxSingleNullable : Producer { private readonly IObservable _source; public MaxSingleNullable(IObservable source) { _source = source; } protected override _ CreateSink(IObserver observer) => new _(observer); protected override void Run(_ sink) => sink.Run(_source); internal sealed class _ : IdentitySink { private float? _lastValue; public _(IObserver observer) : base(observer) { _lastValue = default(float?); } public override void OnNext(float? value) { if (!value.HasValue) return; if (_lastValue.HasValue) { if (value > _lastValue || float.IsNaN((float)value)) { _lastValue = value; } } else { _lastValue = value; } } public override void OnCompleted() { ForwardOnNext(_lastValue); ForwardOnCompleted(); } } } internal sealed class MaxDecimalNullable : Producer { private readonly IObservable _source; public MaxDecimalNullable(IObservable source) { _source = source; } protected override _ CreateSink(IObserver observer) => new _(observer); protected override void Run(_ sink) => sink.Run(_source); internal sealed class _ : IdentitySink { private decimal? _lastValue; public _(IObserver observer) : base(observer) { _lastValue = default(decimal?); } public override void OnNext(decimal? value) { if (!value.HasValue) return; if (_lastValue.HasValue) { if (value > _lastValue) { _lastValue = value; } } else { _lastValue = value; } } public override void OnCompleted() { ForwardOnNext(_lastValue); ForwardOnCompleted(); } } } internal sealed class MaxInt32Nullable : Producer { private readonly IObservable _source; public MaxInt32Nullable(IObservable source) { _source = source; } protected override _ CreateSink(IObserver observer) => new _(observer); protected override void Run(_ sink) => sink.Run(_source); internal sealed class _ : IdentitySink { private int? _lastValue; public _(IObserver observer) : base(observer) { _lastValue = default(int?); } public override void OnNext(int? value) { if (!value.HasValue) return; if (_lastValue.HasValue) { if (value > _lastValue) { _lastValue = value; } } else { _lastValue = value; } } public override void OnCompleted() { ForwardOnNext(_lastValue); ForwardOnCompleted(); } } } internal sealed class MaxInt64Nullable : Producer { private readonly IObservable _source; public MaxInt64Nullable(IObservable source) { _source = source; } protected override _ CreateSink(IObserver observer) => new _(observer); protected override void Run(_ sink) => sink.Run(_source); internal sealed class _ : IdentitySink { private long? _lastValue; public _(IObserver observer) : base(observer) { _lastValue = default(long?); } public override void OnNext(long? value) { if (!value.HasValue) return; if (_lastValue.HasValue) { if (value > _lastValue) { _lastValue = value; } } else { _lastValue = value; } } public override void OnCompleted() { ForwardOnNext(_lastValue); ForwardOnCompleted(); } } } }