reactive/Rx.NET/Source/System.Reactive.Linq/Reactive/Linq/QueryLanguage.Aggregates.cs

// 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;
using System.Linq;
using System.Reactive.Disposables;

namespace System.Reactive.Linq
{
#if !NO_PERF
    using ObservableImpl;
#endif

    internal partial class QueryLanguage
    {
        #region + Aggregate +

        public virtual IObservable<TAccumulate> Aggregate<TSource, TAccumulate>(IObservable<TSource> source, TAccumulate seed, Func<TAccumulate, TSource, TAccumulate> accumulator)
        {
#if !NO_PERF
            return new Aggregate<TSource, TAccumulate, TAccumulate>(source, seed, accumulator, Stubs<TAccumulate>.I);
#else
            return source.Scan(seed, accumulator).StartWith(seed).Final();
#endif
        }

        public virtual IObservable<TResult> Aggregate<TSource, TAccumulate, TResult>(IObservable<TSource> source, TAccumulate seed, Func<TAccumulate, TSource, TAccumulate> accumulator, Func<TAccumulate, TResult> resultSelector)
        {
#if !NO_PERF
            return new Aggregate<TSource, TAccumulate, TResult>(source, seed, accumulator, resultSelector);
#else
            return Aggregate(source, seed, accumulator).Select(resultSelector);
#endif
        }

        public virtual IObservable<TSource> Aggregate<TSource>(IObservable<TSource> source, Func<TSource, TSource, TSource> accumulator)
        {
#if !NO_PERF
            return new Aggregate<TSource>(source, accumulator);
#else
            return source.Scan(accumulator).Final();
#endif
        }

        public virtual IObservable<double> Average<TSource>(IObservable<TSource> source, Func<TSource, double> selector)
        {
            return Average(Select(source, selector));
        }

        public virtual IObservable<float> Average<TSource>(IObservable<TSource> source, Func<TSource, float> selector)
        {
            return Average(Select(source, selector));
        }

        public virtual IObservable<decimal> Average<TSource>(IObservable<TSource> source, Func<TSource, decimal> selector)
        {
            return Average(Select(source, selector));
        }

        public virtual IObservable<double> Average<TSource>(IObservable<TSource> source, Func<TSource, int> selector)
        {
            return Average(Select(source, selector));
        }

        public virtual IObservable<double> Average<TSource>(IObservable<TSource> source, Func<TSource, long> selector)
        {
            return Average(Select(source, selector));
        }

        public virtual IObservable<double?> Average<TSource>(IObservable<TSource> source, Func<TSource, double?> selector)
        {
            return Average(Select(source, selector));
        }

        public virtual IObservable<float?> Average<TSource>(IObservable<TSource> source, Func<TSource, float?> selector)
        {
            return Average(Select(source, selector));
        }

        public virtual IObservable<decimal?> Average<TSource>(IObservable<TSource> source, Func<TSource, decimal?> selector)
        {
            return Average(Select(source, selector));
        }

        public virtual IObservable<double?> Average<TSource>(IObservable<TSource> source, Func<TSource, int?> selector)
        {
            return Average(Select(source, selector));
        }

        public virtual IObservable<double?> Average<TSource>(IObservable<TSource> source, Func<TSource, long?> selector)
        {
            return Average(Select(source, selector));
        }

        #endregion

        #region + All +

        public virtual IObservable<bool> All<TSource>(IObservable<TSource> source, Func<TSource, bool> predicate)
        {
#if !NO_PERF
            return new All<TSource>(source, predicate);
#else
            return source.Where(v => !(predicate(v))).Any().Select(b => !b);
#endif
        }

        #endregion

        #region + Any +

        public virtual IObservable<bool> Any<TSource>(IObservable<TSource> source)
        {
#if !NO_PERF
            return new Any<TSource>(source);
#else
            return new AnonymousObservable<bool>(observer => source.Subscribe(
                _ =>
                {
                    observer.OnNext(true);
                    observer.OnCompleted();
                },
                observer.OnError,
                () =>
                {
                    observer.OnNext(false);
                    observer.OnCompleted();
                }));
#endif
        }

        public virtual IObservable<bool> Any<TSource>(IObservable<TSource> source, Func<TSource, bool> predicate)
        {
#if !NO_PERF
            return new Any<TSource>(source, predicate);
#else
            return source.Where(predicate).Any();
#endif
        }

        #endregion

        #region + Average +

        public virtual IObservable<double> Average(IObservable<double> source)
        {
#if !NO_PERF
            return new AverageDouble(source);
#else
            return source.Scan(new { sum = 0.0, count = 0L },
                               (prev, cur) => new { sum = prev.sum + cur, count = checked(prev.count + 1) })
                .Final()
                .Select(s => s.sum / (double)s.count);
#endif
        }

        public virtual IObservable<float> Average(IObservable<float> source)
        {
#if !NO_PERF
            return new AverageSingle(source);
#else
            return source.Scan(new { sum = 0F, count = 0L }, // NOTE: Uses a different accumulator type (float), *not* conform LINQ to Objects.
                               (prev, cur) => new { sum = prev.sum + cur, count = checked(prev.count + 1) })
                .Final()
                .Select(s => s.sum / (float)s.count);
#endif
        }

        public virtual IObservable<decimal> Average(IObservable<decimal> source)
        {
#if !NO_PERF
            return new AverageDecimal(source);
#else
            return source.Scan(new { sum = 0M, count = 0L },
                               (prev, cur) => new { sum = prev.sum + cur, count = checked(prev.count + 1) })
                .Final()
                .Select(s => s.sum / (decimal)s.count);
#endif
        }

        public virtual IObservable<double> Average(IObservable<int> source)
        {
#if !NO_PERF
            return new AverageInt32(source);
#else
            return source.Scan(new { sum = 0L, count = 0L },
                               (prev, cur) => new { sum = checked(prev.sum + cur), count = checked(prev.count + 1) })
                .Final()
                .Select(s => (double)s.sum / (double)s.count);
#endif
        }

        public virtual IObservable<double> Average(IObservable<long> source)
        {
#if !NO_PERF
            return new AverageInt64(source);
#else
            return source.Scan(new { sum = 0L, count = 0L },
                               (prev, cur) => new { sum = checked(prev.sum + cur), count = checked(prev.count + 1) })
                .Final()
                .Select(s => (double)s.sum / (double)s.count);
#endif
        }

        public virtual IObservable<double?> Average(IObservable<double?> source)
        {
#if !NO_PERF
            return new AverageDoubleNullable(source);
#else
            return source.Aggregate(new { sum = new double?(0.0), count = 0L },
                               (prev, cur) => cur != null ? new { sum = prev.sum + cur.GetValueOrDefault(), count = checked(prev.count + 1) } : prev)                
                               .Select(s => s.count == 0 ? default(double?) : (double?)s.sum / (double)s.count);
#endif
        }

        public virtual IObservable<float?> Average(IObservable<float?> source)
        {
#if !NO_PERF
            return new AverageSingleNullable(source);
#else
            return source.Aggregate(new { sum = new float?(0f), count = 0L }, // NOTE: Uses a different accumulator type (float), *not* conform LINQ to Objects.
                               (prev, cur) => cur != null ? new { sum = prev.sum + cur.GetValueOrDefault(), count = checked(prev.count + 1) } : prev)
                               .Select(s => s.count == 0 ? default(float?) : (float?)s.sum / (float)s.count);
#endif
        }

        public virtual IObservable<decimal?> Average(IObservable<decimal?> source)
        {
#if !NO_PERF
            return new AverageDecimalNullable(source);
#else
            return source.Aggregate(new { sum = new decimal?(0M), count = 0L },
                               (prev, cur) => cur != null ? new { sum = prev.sum + cur.GetValueOrDefault(), count = checked(prev.count + 1) } : prev)
                               .Select(s => s.count == 0 ? default(decimal?) : (decimal?)s.sum / (decimal)s.count);
#endif
        }

        public virtual IObservable<double?> Average(IObservable<int?> source)
        {
#if !NO_PERF
            return new AverageInt32Nullable(source);
#else
            return source.Aggregate(new { sum = new long?(0), count = 0L },
                               (prev, cur) => cur != null ? new { sum = checked(prev.sum + cur.GetValueOrDefault()), count = checked(prev.count + 1) } : prev)
                               .Select(s => s.count == 0 ? default(double?) : (double?)s.sum / s.count);
#endif
        }

        public virtual IObservable<double?> Average(IObservable<long?> source)
        {
#if !NO_PERF
            return new AverageInt64Nullable(source);
#else
            return source.Aggregate(new { sum = new long?(0), count = 0L },
                               (prev, cur) => cur != null ? new { sum = checked(prev.sum + cur.GetValueOrDefault()), count = checked(prev.count + 1) } : prev)
                               .Select(s => s.count == 0 ? default(double?): (double?)s.sum / s.count);
#endif
        }

        #endregion

        #region + Contains +

        public virtual IObservable<bool> Contains<TSource>(IObservable<TSource> source, TSource value)
        {
#if !NO_PERF
            return new Contains<TSource>(source, value, EqualityComparer<TSource>.Default);
#else
            return Contains_<TSource>(source, value, EqualityComparer<TSource>.Default);
#endif
        }

        public virtual IObservable<bool> Contains<TSource>(IObservable<TSource> source, TSource value, IEqualityComparer<TSource> comparer)
        {
#if !NO_PERF
            return new Contains<TSource>(source, value, comparer);
#else
            return Contains_<TSource>(source, value, comparer);
#endif
        }

#if NO_PERF
        private static IObservable<bool> Contains_<TSource>(IObservable<TSource> source, TSource value, IEqualityComparer<TSource> comparer)
        {
            return source.Where(v => comparer.Equals(v, value)).Any();
        }
#endif

        #endregion

        #region + Count +

        public virtual IObservable<int> Count<TSource>(IObservable<TSource> source)
        {
#if !NO_PERF
            return new Count<TSource>(source);
#else
            return source.Aggregate(0, (count, _) => checked(count + 1));
#endif
        }

        public virtual IObservable<int> Count<TSource>(IObservable<TSource> source, Func<TSource, bool> predicate)
        {
#if !NO_PERF
            return new Count<TSource>(source, predicate);
#else
            return source.Where(predicate).Aggregate(0, (count, _) => checked(count + 1));
#endif
        }

        #endregion

        #region + ElementAt +

        public virtual IObservable<TSource> ElementAt<TSource>(IObservable<TSource> source, int index)
        {
#if !NO_PERF
            return new ElementAt<TSource>(source, index, true);
#else
            return new AnonymousObservable<TSource>(observer =>
            {
                int i = index;
                return source.Subscribe(
                    x =>
                    {
                        if (i == 0)
                        {
                            observer.OnNext(x);
                            observer.OnCompleted();
                        }

                        i--;
                    },
                    observer.OnError,
                    () => observer.OnError(new ArgumentOutOfRangeException("index"))
                );
            });
#endif
        }

        #endregion

        #region + ElementAtOrDefault +

        public virtual IObservable<TSource> ElementAtOrDefault<TSource>(IObservable<TSource> source, int index)
        {
#if !NO_PERF
            return new ElementAt<TSource>(source, index, false);
#else
            return new AnonymousObservable<TSource>(observer =>
            {
                int i = index;
                return source.Subscribe(
                    x =>
                    {
                        if (i == 0)
                        {
                            observer.OnNext(x);
                            observer.OnCompleted();
                        }

                        i--;
                    },
                    observer.OnError,
                    () =>
                    {
                        observer.OnNext(default(TSource));
                        observer.OnCompleted();
                    }
                );
            });
#endif
        }

        #endregion

        #region + FirstAsync +

        public virtual IObservable<TSource> FirstAsync<TSource>(IObservable<TSource> source)
        {
#if !NO_PERF
            return new FirstAsync<TSource>(source, null, true);
#else
            return FirstOrDefaultAsync_(source, true);
#endif
        }

        public virtual IObservable<TSource> FirstAsync<TSource>(IObservable<TSource> source, Func<TSource, bool> predicate)
        {
#if !NO_PERF
            return new FirstAsync<TSource>(source, predicate, true);
#else
            return source.Where(predicate).FirstAsync();
#endif
        }

        #endregion

        #region + FirstAsyncOrDefaultAsync +

        public virtual IObservable<TSource> FirstOrDefaultAsync<TSource>(IObservable<TSource> source)
        {
#if !NO_PERF
            return new FirstAsync<TSource>(source, null, false);
#else
            return FirstOrDefaultAsync_(source, false);
#endif
        }

        public virtual IObservable<TSource> FirstOrDefaultAsync<TSource>(IObservable<TSource> source, Func<TSource, bool> predicate)
        {
#if !NO_PERF
            return new FirstAsync<TSource>(source, predicate, false);
#else
            return source.Where(predicate).FirstOrDefaultAsync();
#endif
        }

#if NO_PERF
        private static IObservable<TSource> FirstOrDefaultAsync_<TSource>(IObservable<TSource> source, bool throwOnEmpty)
        {
            return new AnonymousObservable<TSource>(observer =>
            {
                return source.Subscribe(
                    x =>
                    {
                        observer.OnNext(x);
                        observer.OnCompleted();
                    },
                    observer.OnError,
                    () =>
                    {
                        if (throwOnEmpty)
                        {
                            observer.OnError(new InvalidOperationException(Strings_Linq.NO_ELEMENTS));
                        }
                        else
                        {
                            observer.OnNext(default(TSource));
                            observer.OnCompleted();
                        }
                    }
                );
            });
        }
#endif

        #endregion

        #region + IsEmpty +

        public virtual IObservable<bool> IsEmpty<TSource>(IObservable<TSource> source)
        {
#if !NO_PERF
            return new IsEmpty<TSource>(source);
#else
            return source.Any().Select(b => !b);
#endif
        }

        #endregion

        #region + LastAsync +

        public virtual IObservable<TSource> LastAsync<TSource>(IObservable<TSource> source)
        {
#if !NO_PERF
            return new LastAsync<TSource>(source, null, true);
#else
            return LastOrDefaultAsync_(source, true);
#endif
        }

        public virtual IObservable<TSource> LastAsync<TSource>(IObservable<TSource> source, Func<TSource, bool> predicate)
        {
#if !NO_PERF
            return new LastAsync<TSource>(source, predicate, true);
#else
            return source.Where(predicate).LastAsync();
#endif
        }

        #endregion

        #region + LastOrDefaultAsync +

        public virtual IObservable<TSource> LastOrDefaultAsync<TSource>(IObservable<TSource> source)
        {
#if !NO_PERF
            return new LastAsync<TSource>(source, null, false);
#else
            return LastOrDefaultAsync_(source, false);
#endif
        }

        public virtual IObservable<TSource> LastOrDefaultAsync<TSource>(IObservable<TSource> source, Func<TSource, bool> predicate)
        {
#if !NO_PERF
            return new LastAsync<TSource>(source, predicate, false);
#else
            return source.Where(predicate).LastOrDefaultAsync();
#endif
        }

#if NO_PERF
        private static IObservable<TSource> LastOrDefaultAsync_<TSource>(IObservable<TSource> source, bool throwOnEmpty)
        {
            return new AnonymousObservable<TSource>(observer =>
            {
                var value = default(TSource);
                var seenValue = false;

                return source.Subscribe(
                    x =>
                    {
                        value = x;
                        seenValue = true;
                    },
                    observer.OnError,
                    () =>
                    {
                        if (throwOnEmpty && !seenValue)
                        {
                            observer.OnError(new InvalidOperationException(Strings_Linq.NO_ELEMENTS));
                        }
                        else
                        {
                            observer.OnNext(value);
                            observer.OnCompleted();
                        }
                    }
                );
            });
        }
#endif

        #endregion

        #region + LongCount +

        public virtual IObservable<long> LongCount<TSource>(IObservable<TSource> source)
        {
#if !NO_PERF
            return new LongCount<TSource>(source);
#else
            return source.Aggregate(0L, (count, _) => checked(count + 1));
#endif
        }

        public virtual IObservable<long> LongCount<TSource>(IObservable<TSource> source, Func<TSource, bool> predicate)
        {
#if !NO_PERF
            return new LongCount<TSource>(source, predicate);
#else
            return source.Where(predicate).Aggregate(0L, (count, _) => checked(count + 1));
#endif
        }

        #endregion

        #region + Max +

        public virtual IObservable<TSource> Max<TSource>(IObservable<TSource> source)
        {
#if !NO_PERF
            // BREAKING CHANGE v2 > v1.x - Behavior for reference types
            return new Max<TSource>(source, Comparer<TSource>.Default);
#else
            return MaxBy(source, x => x).Select(x => x.First());
#endif
        }

        public virtual IObservable<TSource> Max<TSource>(IObservable<TSource> source, IComparer<TSource> comparer)
        {
#if !NO_PERF
            // BREAKING CHANGE v2 > v1.x - Behavior for reference types
            return new Max<TSource>(source, comparer);
#else
            return MaxBy(source, x => x, comparer).Select(x => x.First());
#endif
        }

        public virtual IObservable<double> Max(IObservable<double> source)
        {
#if !NO_PERF
            return new MaxDouble(source);
#else
            return source.Scan(double.MinValue, Math.Max).Final();
#endif
        }

        public virtual IObservable<float> Max(IObservable<float> source)
        {
#if !NO_PERF
            return new MaxSingle(source);
#else
            return source.Scan(float.MinValue, Math.Max).Final();
#endif
        }

        public virtual IObservable<decimal> Max(IObservable<decimal> source)
        {
#if !NO_PERF
            return new MaxDecimal(source);
#else
            return source.Scan(decimal.MinValue, Math.Max).Final();
#endif
        }

        public virtual IObservable<int> Max(IObservable<int> source)
        {
#if !NO_PERF
            return new MaxInt32(source);
#else
            return source.Scan(int.MinValue, Math.Max).Final();
#endif
        }

        public virtual IObservable<long> Max(IObservable<long> source)
        {
#if !NO_PERF
            return new MaxInt64(source);
#else
            return source.Scan(long.MinValue, Math.Max).Final();
#endif
        }

        public virtual IObservable<double?> Max(IObservable<double?> source)
        {
#if !NO_PERF
            return new MaxDoubleNullable(source);
#else
            return source.Aggregate(new double?(), NullableMax);
#endif
        }

        public virtual IObservable<float?> Max(IObservable<float?> source)
        {
#if !NO_PERF
            return new MaxSingleNullable(source);
#else
            return source.Aggregate(new float?(), NullableMax);
#endif
        }

        public virtual IObservable<decimal?> Max(IObservable<decimal?> source)
        {
#if !NO_PERF
            return new MaxDecimalNullable(source);
#else
            return source.Aggregate(new decimal?(), NullableMax);
#endif
        }

        public virtual IObservable<int?> Max(IObservable<int?> source)
        {
#if !NO_PERF
            return new MaxInt32Nullable(source);
#else
            return source.Aggregate(new int?(), NullableMax);
#endif
        }

        public virtual IObservable<long?> Max(IObservable<long?> source)
        {
#if !NO_PERF
            return new MaxInt64Nullable(source);
#else
            return source.Aggregate(new long?(), NullableMax);
#endif
        }

        public virtual IObservable<TResult> Max<TSource, TResult>(IObservable<TSource> source, Func<TSource, TResult> selector)
        {
            return Max(Select(source, selector));
        }

        public virtual IObservable<TResult> Max<TSource, TResult>(IObservable<TSource> source, Func<TSource, TResult> selector, IComparer<TResult> comparer)
        {
            return Max(Select(source, selector), comparer);
        }

        public virtual IObservable<double> Max<TSource>(IObservable<TSource> source, Func<TSource, double> selector)
        {
            return Max(Select(source, selector));
        }

        public virtual IObservable<float> Max<TSource>(IObservable<TSource> source, Func<TSource, float> selector)
        {
            return Max(Select(source, selector));
        }

        public virtual IObservable<decimal> Max<TSource>(IObservable<TSource> source, Func<TSource, decimal> selector)
        {
            return Max(Select(source, selector));
        }

        public virtual IObservable<int> Max<TSource>(IObservable<TSource> source, Func<TSource, int> selector)
        {
            return Max(Select(source, selector));
        }

        public virtual IObservable<long> Max<TSource>(IObservable<TSource> source, Func<TSource, long> selector)
        {
            return Max(Select(source, selector));
        }

        public virtual IObservable<double?> Max<TSource>(IObservable<TSource> source, Func<TSource, double?> selector)
        {
            return Max(Select(source, selector));
        }

        public virtual IObservable<float?> Max<TSource>(IObservable<TSource> source, Func<TSource, float?> selector)
        {
            return Max(Select(source, selector));
        }

        public virtual IObservable<decimal?> Max<TSource>(IObservable<TSource> source, Func<TSource, decimal?> selector)
        {
            return Max(Select(source, selector));
        }

        public virtual IObservable<int?> Max<TSource>(IObservable<TSource> source, Func<TSource, int?> selector)
        {
            return Max(Select(source, selector));
        }

        public virtual IObservable<long?> Max<TSource>(IObservable<TSource> source, Func<TSource, long?> selector)
        {
            return Max(Select(source, selector));
        }

        #endregion

        #region + MaxBy +

        public virtual IObservable<IList<TSource>> MaxBy<TSource, TKey>(IObservable<TSource> source, Func<TSource, TKey> keySelector)
        {
#if !NO_PERF
            return new MaxBy<TSource, TKey>(source, keySelector, Comparer<TKey>.Default);
#else
            return MaxBy(source, keySelector, Comparer<TKey>.Default);
#endif
        }

        public virtual IObservable<IList<TSource>> MaxBy<TSource, TKey>(IObservable<TSource> source, Func<TSource, TKey> keySelector, IComparer<TKey> comparer)
        {
#if !NO_PERF
            return new MaxBy<TSource, TKey>(source, keySelector, comparer);
#else
            return ExtremaBy(source, keySelector, comparer);
#endif
        }

        #endregion

        #region + Min +

        public virtual IObservable<TSource> Min<TSource>(IObservable<TSource> source)
        {
#if !NO_PERF
            // BREAKING CHANGE v2 > v1.x - Behavior for reference types
            return new Min<TSource>(source, Comparer<TSource>.Default);
#else
            return MinBy(source, x => x).Select(x => x.First());
#endif
        }

        public virtual IObservable<TSource> Min<TSource>(IObservable<TSource> source, IComparer<TSource> comparer)
        {
#if !NO_PERF
            // BREAKING CHANGE v2 > v1.x - Behavior for reference types
            return new Min<TSource>(source, comparer);
#else
            return MinBy(source, x => x, comparer).Select(x => x.First());
#endif
        }

        public virtual IObservable<double> Min(IObservable<double> source)
        {
#if !NO_PERF
            return new MinDouble(source);
#else
            return source.Scan(double.MaxValue, Math.Min).Final();
#endif
        }

        public virtual IObservable<float> Min(IObservable<float> source)
        {
#if !NO_PERF
            return new MinSingle(source);
#else
            return source.Scan(float.MaxValue, Math.Min).Final();
#endif
        }

        public virtual IObservable<decimal> Min(IObservable<decimal> source)
        {
#if !NO_PERF
            return new MinDecimal(source);
#else
            return source.Scan(decimal.MaxValue, Math.Min).Final();
#endif
        }

        public virtual IObservable<int> Min(IObservable<int> source)
        {
#if !NO_PERF
            return new MinInt32(source);
#else
            return source.Scan(int.MaxValue, Math.Min).Final();
#endif
        }

        public virtual IObservable<long> Min(IObservable<long> source)
        {
#if !NO_PERF
            return new MinInt64(source);
#else
            return source.Scan(long.MaxValue, Math.Min).Final();
#endif
        }

        public virtual IObservable<double?> Min(IObservable<double?> source)
        {
#if !NO_PERF
            return new MinDoubleNullable(source);
#else
            return source.Aggregate(new double?(), NullableMin);
#endif
        }

        public virtual IObservable<float?> Min(IObservable<float?> source)
        {
#if !NO_PERF
            return new MinSingleNullable(source);
#else
            return source.Aggregate(new float?(), NullableMin);
#endif
        }

        public virtual IObservable<decimal?> Min(IObservable<decimal?> source)
        {
#if !NO_PERF
            return new MinDecimalNullable(source);
#else
            return source.Aggregate(new decimal?(), NullableMin);
#endif
        }

        public virtual IObservable<int?> Min(IObservable<int?> source)
        {
#if !NO_PERF
            return new MinInt32Nullable(source);
#else
            return source.Aggregate(new int?(), NullableMin);
#endif
        }

        public virtual IObservable<long?> Min(IObservable<long?> source)
        {
#if !NO_PERF
            return new MinInt64Nullable(source);
#else
            return source.Aggregate(new long?(), NullableMin);
#endif
        }

        public virtual IObservable<TResult> Min<TSource, TResult>(IObservable<TSource> source, Func<TSource, TResult> selector)
        {
            return Min(Select(source, selector));
        }

        public virtual IObservable<TResult> Min<TSource, TResult>(IObservable<TSource> source, Func<TSource, TResult> selector, IComparer<TResult> comparer)
        {
            return Min(Select(source, selector), comparer);
        }

        public virtual IObservable<double> Min<TSource>(IObservable<TSource> source, Func<TSource, double> selector)
        {
            return Min(Select(source, selector));
        }

        public virtual IObservable<float> Min<TSource>(IObservable<TSource> source, Func<TSource, float> selector)
        {
            return Min(Select(source, selector));
        }

        public virtual IObservable<decimal> Min<TSource>(IObservable<TSource> source, Func<TSource, decimal> selector)
        {
            return Min(Select(source, selector));
        }

        public virtual IObservable<int> Min<TSource>(IObservable<TSource> source, Func<TSource, int> selector)
        {
            return Min(Select(source, selector));
        }

        public virtual IObservable<long> Min<TSource>(IObservable<TSource> source, Func<TSource, long> selector)
        {
            return Min(Select(source, selector));
        }

        public virtual IObservable<double?> Min<TSource>(IObservable<TSource> source, Func<TSource, double?> selector)
        {
            return Min(Select(source, selector));
        }

        public virtual IObservable<float?> Min<TSource>(IObservable<TSource> source, Func<TSource, float?> selector)
        {
            return Min(Select(source, selector));
        }

        public virtual IObservable<decimal?> Min<TSource>(IObservable<TSource> source, Func<TSource, decimal?> selector)
        {
            return Min(Select(source, selector));
        }

        public virtual IObservable<int?> Min<TSource>(IObservable<TSource> source, Func<TSource, int?> selector)
        {
            return Min(Select(source, selector));
        }

        public virtual IObservable<long?> Min<TSource>(IObservable<TSource> source, Func<TSource, long?> selector)
        {
            return Min(Select(source, selector));
        }

        #endregion

        #region + MinBy +

        public virtual IObservable<IList<TSource>> MinBy<TSource, TKey>(IObservable<TSource> source, Func<TSource, TKey> keySelector)
        {
#if !NO_PERF
            return new MinBy<TSource, TKey>(source, keySelector, Comparer<TKey>.Default);
#else
            return MinBy(source, keySelector, Comparer<TKey>.Default);
#endif
        }

        public virtual IObservable<IList<TSource>> MinBy<TSource, TKey>(IObservable<TSource> source, Func<TSource, TKey> keySelector, IComparer<TKey> comparer)
        {
#if !NO_PERF
            return new MinBy<TSource, TKey>(source, keySelector, comparer);
#else
            return ExtremaBy(source, keySelector, new AnonymousComparer<TKey>((x, y) => comparer.Compare(x, y) * -1));
#endif
        }

        #endregion

        #region + SequenceEqual +

        public virtual IObservable<bool> SequenceEqual<TSource>(IObservable<TSource> first, IObservable<TSource> second)
        {
#if !NO_PERF
            return new SequenceEqual<TSource>(first, second, EqualityComparer<TSource>.Default);
#else
            return first.SequenceEqual(second, EqualityComparer<TSource>.Default);
#endif
        }

        public virtual IObservable<bool> SequenceEqual<TSource>(IObservable<TSource> first, IObservable<TSource> second, IEqualityComparer<TSource> comparer)
        {
#if !NO_PERF
            return new SequenceEqual<TSource>(first, second, comparer);
#else
            return new AnonymousObservable<bool>(observer =>
            {
                var gate = new object();
                var donel = false;
                var doner = false;
                var ql = new Queue<TSource>();
                var qr = new Queue<TSource>();

                var subscription1 = first.Subscribe(
                    x =>
                    {
                        lock (gate)
                        {
                            if (qr.Count > 0)
                            {
                                var equal = false;
                                var v = qr.Dequeue();
                                try
                                {
                                    equal = comparer.Equals(x, v);
                                }
                                catch (Exception exception)
                                {
                                    observer.OnError(exception);
                                    return;
                                }
                                if (!equal)
                                {
                                    observer.OnNext(false);
                                    observer.OnCompleted();
                                }
                            }
                            else if (doner)
                            {
                                observer.OnNext(false);
                                observer.OnCompleted();
                            }
                            else
                                ql.Enqueue(x);
                        }
                    },
                    observer.OnError,
                    () =>
                    {
                        lock (gate)
                        {
                            donel = true;
                            if (ql.Count == 0)
                            {
                                if (qr.Count > 0)
                                {
                                    observer.OnNext(false);
                                    observer.OnCompleted();
                                }
                                else if (doner)
                                {
                                    observer.OnNext(true);
                                    observer.OnCompleted();
                                }
                            }
                        }
                    });

                var subscription2 = second.Subscribe(
                    x =>
                    {
                        lock (gate)
                        {
                            if (ql.Count > 0)
                            {
                                var equal = false;
                                var v = ql.Dequeue();
                                try
                                {
                                    equal = comparer.Equals(v, x);
                                }
                                catch (Exception exception)
                                {
                                    observer.OnError(exception);
                                    return;
                                }
                                if (!equal)
                                {
                                    observer.OnNext(false);
                                    observer.OnCompleted();
                                }
                            }
                            else if (donel)
                            {
                                observer.OnNext(false);
                                observer.OnCompleted();
                            }
                            else
                                qr.Enqueue(x);
                        }
                    },
                    observer.OnError,
                    () =>
                    {
                        lock (gate)
                        {
                            doner = true;
                            if (qr.Count == 0)
                            {
                                if (ql.Count > 0)
                                {
                                    observer.OnNext(false);
                                    observer.OnCompleted();
                                }
                                else if (donel)
                                {
                                    observer.OnNext(true);
                                    observer.OnCompleted();
                                }
                            }
                        }
                    });

                return new CompositeDisposable(subscription1, subscription2);
            });
#endif
        }

        public virtual IObservable<bool> SequenceEqual<TSource>(IObservable<TSource> first, IEnumerable<TSource> second)
        {
#if !NO_PERF
            return new SequenceEqual<TSource>(first, second, EqualityComparer<TSource>.Default);
#else
            return SequenceEqual<TSource>(first, second, EqualityComparer<TSource>.Default);
#endif
        }

        public virtual IObservable<bool> SequenceEqual<TSource>(IObservable<TSource> first, IEnumerable<TSource> second, IEqualityComparer<TSource> comparer)
        {
#if !NO_PERF
            return new SequenceEqual<TSource>(first, second, comparer);
#else
            return new AnonymousObservable<bool>(observer =>
            {
                var e = default(IEnumerator<TSource>);
                try
                {
                    e = second.GetEnumerator();
                }
                catch (Exception ex)
                {
                    observer.OnError(ex);
                    return Disposable.Empty;
                }

                return new CompositeDisposable(
                    first.Subscribe(
                        value =>
                        {
                            var equal = false;
                            try
                            {
                                var hasNext = e.MoveNext();
                                if (hasNext)
                                {
                                    var current = e.Current;
                                    equal = comparer.Equals(value, current);
                                }
                            }
                            catch (Exception ex)
                            {
                                observer.OnError(ex);
                                return;
                            }

                            if (!equal)
                            {
                                observer.OnNext(false);
                                observer.OnCompleted();
                            }
                        },
                        observer.OnError,
                        () =>
                        {
                            var hasNext = false;

                            try
                            {
                                hasNext = e.MoveNext();
                            }
                            catch (Exception exception)
                            {
                                observer.OnError(exception);
                                return;
                            }

                            observer.OnNext(!hasNext);
                            observer.OnCompleted();
                        }
                    ),
                    e
                );
            });
#endif
        }

        #endregion

        #region + SingleAsync +

        public virtual IObservable<TSource> SingleAsync<TSource>(IObservable<TSource> source)
        {
#if !NO_PERF
            return new SingleAsync<TSource>(source, null, true);
#else
            return SingleOrDefaultAsync_(source, true);
#endif
        }

        public virtual IObservable<TSource> SingleAsync<TSource>(IObservable<TSource> source, Func<TSource, bool> predicate)
        {
#if !NO_PERF
            return new SingleAsync<TSource>(source, predicate, true);
#else
            return source.Where(predicate).SingleAsync();
#endif
        }

        #endregion

        #region + SingleOrDefaultAsync +

        public virtual IObservable<TSource> SingleOrDefaultAsync<TSource>(IObservable<TSource> source)
        {
#if !NO_PERF
            return new SingleAsync<TSource>(source, null, false);
#else
            return SingleOrDefaultAsync_(source, false);
#endif
        }

        public virtual IObservable<TSource> SingleOrDefaultAsync<TSource>(IObservable<TSource> source, Func<TSource, bool> predicate)
        {
#if !NO_PERF
            return new SingleAsync<TSource>(source, predicate, false);
#else
            return source.Where(predicate).SingleOrDefaultAsync();
#endif
        }

#if NO_PERF
        private static IObservable<TSource> SingleOrDefaultAsync_<TSource>(IObservable<TSource> source, bool throwOnEmpty)
        {
            return new AnonymousObservable<TSource>(observer =>
            {
                var value = default(TSource);
                var seenValue = false;

                return source.Subscribe(
                    x =>
                    {
                        if (seenValue)
                        {
                            observer.OnError(new InvalidOperationException(Strings_Linq.MORE_THAN_ONE_ELEMENT));
                        }
                        else
                        {
                            value = x;
                            seenValue = true;
                        }
                    },
                    observer.OnError,
                    () =>
                    {
                        if (throwOnEmpty && !seenValue)
                        {
                            observer.OnError(new InvalidOperationException(Strings_Linq.NO_ELEMENTS));
                        }
                        else
                        {
                            observer.OnNext(value);
                            observer.OnCompleted();
                        }
                    }
                );
            });
        }
#endif

        #endregion

        #region + Sum +

        public virtual IObservable<double> Sum(IObservable<double> source)
        {
#if !NO_PERF
            return new SumDouble(source);
#else
            return source.Aggregate(0.0, (prev, curr) => prev + curr);
#endif
        }

        public virtual IObservable<float> Sum(IObservable<float> source)
        {
#if !NO_PERF
            return new SumSingle(source);
#else
            return source.Aggregate(0f, (prev, curr) => prev + curr);
#endif
        }

        public virtual IObservable<decimal> Sum(IObservable<decimal> source)
        {
#if !NO_PERF
            return new SumDecimal(source);
#else
            return source.Aggregate(0M, (prev, curr) => prev + curr);
#endif
        }

        public virtual IObservable<int> Sum(IObservable<int> source)
        {
#if !NO_PERF
            return new SumInt32(source);
#else
            return source.Aggregate(0, (prev, curr) => checked(prev + curr));
#endif
        }

        public virtual IObservable<long> Sum(IObservable<long> source)
        {
#if !NO_PERF
            return new SumInt64(source);
#else
            return source.Aggregate(0L, (prev, curr) => checked(prev + curr));
#endif
        }

        public virtual IObservable<double?> Sum(IObservable<double?> source)
        {
#if !NO_PERF
            return new SumDoubleNullable(source);
#else
            return source.Aggregate(0.0, (prev, curr) => prev + curr.GetValueOrDefault()).Select(x => (double?)x);
#endif
        }

        public virtual IObservable<float?> Sum(IObservable<float?> source)
        {
#if !NO_PERF
            return new SumSingleNullable(source);
#else
            return source.Aggregate(0f, (prev, curr) => prev + curr.GetValueOrDefault()).Select(x => (float?)x);
#endif
        }

        public virtual IObservable<decimal?> Sum(IObservable<decimal?> source)
        {
#if !NO_PERF
            return new SumDecimalNullable(source);
#else
            return source.Aggregate(0M, (prev, curr) => prev + curr.GetValueOrDefault()).Select(x => (decimal?)x);
#endif
        }

        public virtual IObservable<int?> Sum(IObservable<int?> source)
        {
#if !NO_PERF
            return new SumInt32Nullable(source);
#else
            return source.Aggregate(0, (prev, curr) => checked(prev + curr.GetValueOrDefault())).Select(x => (int?)x);
#endif
        }

        public virtual IObservable<long?> Sum(IObservable<long?> source)
        {
#if !NO_PERF
            return new SumInt64Nullable(source);
#else
            return source.Aggregate(0L, (prev, curr) => checked(prev + curr.GetValueOrDefault())).Select(x => (long?)x);
#endif
        }

        public virtual IObservable<double> Sum<TSource>(IObservable<TSource> source, Func<TSource, double> selector)
        {
            return Sum(Select(source, selector));
        }

        public virtual IObservable<float> Sum<TSource>(IObservable<TSource> source, Func<TSource, float> selector)
        {
            return Sum(Select(source, selector));
        }

        public virtual IObservable<decimal> Sum<TSource>(IObservable<TSource> source, Func<TSource, decimal> selector)
        {
            return Sum(Select(source, selector));
        }

        public virtual IObservable<int> Sum<TSource>(IObservable<TSource> source, Func<TSource, int> selector)
        {
            return Sum(Select(source, selector));
        }

        public virtual IObservable<long> Sum<TSource>(IObservable<TSource> source, Func<TSource, long> selector)
        {
            return Sum(Select(source, selector));
        }

        public virtual IObservable<double?> Sum<TSource>(IObservable<TSource> source, Func<TSource, double?> selector)
        {
            return Sum(Select(source, selector));
        }

        public virtual IObservable<float?> Sum<TSource>(IObservable<TSource> source, Func<TSource, float?> selector)
        {
            return Sum(Select(source, selector));
        }

        public virtual IObservable<decimal?> Sum<TSource>(IObservable<TSource> source, Func<TSource, decimal?> selector)
        {
            return Sum(Select(source, selector));
        }

        public virtual IObservable<int?> Sum<TSource>(IObservable<TSource> source, Func<TSource, int?> selector)
        {
            return Sum(Select(source, selector));
        }

        public virtual IObservable<long?> Sum<TSource>(IObservable<TSource> source, Func<TSource, long?> selector)
        {
            return Sum(Select(source, selector));
        }

        #endregion

        #region + ToArray +

        public virtual IObservable<TSource[]> ToArray<TSource>(IObservable<TSource> source)
        {
#if !NO_PERF
            return new ToArray<TSource>(source);
#else
            return source.ToList().Select(xs => xs.ToArray());
#endif
        }

        #endregion

        #region + ToDictionary +

        public virtual IObservable<IDictionary<TKey, TElement>> ToDictionary<TSource, TKey, TElement>(IObservable<TSource> source, Func<TSource, TKey> keySelector, Func<TSource, TElement> elementSelector, IEqualityComparer<TKey> comparer)
        {
#if !NO_PERF
            return new ToDictionary<TSource, TKey, TElement>(source, keySelector, elementSelector, comparer);
#else
            return source.Aggregate((IDictionary<TKey, TElement>)new Dictionary<TKey, TElement>(comparer), (dict, x) =>
                {
                    dict.Add(keySelector(x), elementSelector(x));
                    return dict;
                });
#endif
        }

        public virtual IObservable<IDictionary<TKey, TElement>> ToDictionary<TSource, TKey, TElement>(IObservable<TSource> source, Func<TSource, TKey> keySelector, Func<TSource, TElement> elementSelector)
        {
#if !NO_PERF
            return new ToDictionary<TSource, TKey, TElement>(source, keySelector, elementSelector, EqualityComparer<TKey>.Default);
#else
            return source.ToDictionary(keySelector, elementSelector, EqualityComparer<TKey>.Default);
#endif
        }

        public virtual IObservable<IDictionary<TKey, TSource>> ToDictionary<TSource, TKey>(IObservable<TSource> source, Func<TSource, TKey> keySelector, IEqualityComparer<TKey> comparer)
        {
#if !NO_PERF
            return new ToDictionary<TSource, TKey, TSource>(source, keySelector, x => x, comparer);
#else
            return source.ToDictionary(keySelector, x => x, comparer);
#endif
        }

        public virtual IObservable<IDictionary<TKey, TSource>> ToDictionary<TSource, TKey>(IObservable<TSource> source, Func<TSource, TKey> keySelector)
        {
#if !NO_PERF
            return new ToDictionary<TSource, TKey, TSource>(source, keySelector, x => x, EqualityComparer<TKey>.Default);
#else
            return source.ToDictionary(keySelector, x => x, EqualityComparer<TKey>.Default);
#endif
        }

        #endregion

        #region + ToList +

        public virtual IObservable<IList<TSource>> ToList<TSource>(IObservable<TSource> source)
        {
#if !NO_PERF
            return new ToList<TSource>(source);
#else
            return source.Aggregate((IList<TSource>)new List<TSource>(), (list, x) =>
            {
                list.Add(x);
                return list;
            });
#endif
        }

        #endregion

        #region + ToLookup +

        public virtual IObservable<ILookup<TKey, TElement>> ToLookup<TSource, TKey, TElement>(IObservable<TSource> source, Func<TSource, TKey> keySelector, Func<TSource, TElement> elementSelector, IEqualityComparer<TKey> comparer)
        {
#if !NO_PERF
            return new ToLookup<TSource, TKey, TElement>(source, keySelector, elementSelector, comparer);
#else
            return source.Aggregate(new Lookup<TKey, TElement>(comparer), (lookup, x) =>
            {
                lookup.Add(keySelector(x), elementSelector(x));
                return lookup;
            }).Select(xs => (ILookup<TKey, TElement>)xs);
#endif
        }

        public virtual IObservable<ILookup<TKey, TSource>> ToLookup<TSource, TKey>(IObservable<TSource> source, Func<TSource, TKey> keySelector, IEqualityComparer<TKey> comparer)
        {
#if !NO_PERF
            return new ToLookup<TSource, TKey, TSource>(source, keySelector, x => x, comparer);
#else
            return source.ToLookup(keySelector, x => x, comparer);
#endif
        }

        public virtual IObservable<ILookup<TKey, TElement>> ToLookup<TSource, TKey, TElement>(IObservable<TSource> source, Func<TSource, TKey> keySelector, Func<TSource, TElement> elementSelector)
        {
#if !NO_PERF
            return new ToLookup<TSource, TKey, TElement>(source, keySelector, elementSelector, EqualityComparer<TKey>.Default);
#else
            return source.ToLookup(keySelector, elementSelector, EqualityComparer<TKey>.Default);
#endif
        }

        public virtual IObservable<ILookup<TKey, TSource>> ToLookup<TSource, TKey>(IObservable<TSource> source, Func<TSource, TKey> keySelector)
        {
#if !NO_PERF
            return new ToLookup<TSource, TKey, TSource>(source, keySelector, x => x, EqualityComparer<TKey>.Default);
#else
            return source.ToLookup(keySelector, x => x, EqualityComparer<TKey>.Default);
#endif
        }

        #endregion

        #region |> Helpers <|

#if NO_PERF
        private static T? NullableMin<T>(T? x, T? y)
            where T : struct, IComparable<T>
        {
            if (!x.HasValue)
                return y;
            if (!y.HasValue)
                return x;
            if (x.Value.CompareTo(y.Value) <= 0)
                return x;
            return y;
        }

        private static T? NullableMax<T>(T? x, T? y)
            where T : struct, IComparable<T>
        {
            if (!x.HasValue)
                return y;
            if (!y.HasValue)
                return x;
            if (x.Value.CompareTo(y.Value) >= 0)
                return x;
            return y;
        }

        private static IObservable<IList<TSource>> ExtremaBy<TSource, TKey>(IObservable<TSource> source, Func<TSource, TKey> keySelector, IComparer<TKey> comparer)
        {
            return new AnonymousObservable<IList<TSource>>(observer =>
            {
                var hasValue = false;
                var lastKey = default(TKey);
                var list = new List<TSource>();
                return source.Subscribe(
                    x =>
                    {
                        var key = default(TKey);
                        try
                        {
                            key = keySelector(x);
                        }
                        catch (Exception ex)
                        {
                            observer.OnError(ex);
                            return;
                        }
                        var comparison = 0;

                        if (!hasValue)
                        {
                            hasValue = true;
                            lastKey = key;
                        }
                        else
                        {
                            try
                            {
                                comparison = comparer.Compare(key, lastKey);
                            }
                            catch (Exception ex)
                            {
                                observer.OnError(ex);
                                return;
                            }
                        }
                        if (comparison > 0)
                        {
                            lastKey = key;
                            list.Clear();
                        }
                        if (comparison >= 0)
                        {
                            list.Add(x);
                        }
                    },
                    observer.OnError,
                    () =>
                    {
                        observer.OnNext(list);
                        observer.OnCompleted();
                    }
                );
            });
        }
#endif

        #endregion
    }

    #region |> Helper types <|

#if NO_PERF
    static class AggregateExtensions
    {
        public static IObservable<TSource> Final<TSource>(this IObservable<TSource> source)
        {
            return new AnonymousObservable<TSource>(observer =>
            {
                var value = default(TSource);
                var hasValue = false;

                return source.Subscribe(
                    x =>
                    {
                        hasValue = true;
                        value = x;
                    },
                    observer.OnError,
                    () =>
                    {
                        if (!hasValue)
                            observer.OnError(new InvalidOperationException(Strings_Linq.NO_ELEMENTS));
                        else
                        {
                            observer.OnNext(value);
                            observer.OnCompleted();
                        }
                    });
            });
        }
    }

    sealed class AnonymousComparer<T> : IComparer<T>
    {
        private readonly Func<T, T, int> comparer;

        /// <summary>
        /// Creates an instance of IComparer by providing a method that compares two objects.
        /// </summary>
        public AnonymousComparer(Func<T, T, int> comparer)
        {
            this.comparer = comparer;
        }

        /// <summary>
        /// Compares two objects and returns a value indicating whether one is less than, equal to, or greater than the other.
        /// </summary>
        public int Compare(T x, T y)
        {
            return comparer(x, y);
        }
    }
#endif

    #endregion
}