// Copyright (c) Microsoft Open Technologies, Inc. All rights reserved. See License.txt in the project root for license information. using System.Collections.Generic; using System.Linq; using System.Reactive.Disposables; namespace System.Reactive.Linq { #if !NO_PERF using Observαble; #endif internal partial class QueryLanguage { #region + Aggregate + public virtual IObservable Aggregate(IObservable source, TAccumulate seed, Func accumulator) { #if !NO_PERF return new Aggregate(source, seed, accumulator, Stubs.I); #else return source.Scan(seed, accumulator).StartWith(seed).Final(); #endif } public virtual IObservable Aggregate(IObservable source, TAccumulate seed, Func accumulator, Func resultSelector) { #if !NO_PERF return new Aggregate(source, seed, accumulator, resultSelector); #else return Aggregate(source, seed, accumulator).Select(resultSelector); #endif } public virtual IObservable Aggregate(IObservable source, Func accumulator) { #if !NO_PERF return new Aggregate(source, accumulator); #else return source.Scan(accumulator).Final(); #endif } public virtual IObservable Average(IObservable source, Func selector) { return Average(Select(source, selector)); } public virtual IObservable Average(IObservable source, Func selector) { return Average(Select(source, selector)); } public virtual IObservable Average(IObservable source, Func selector) { return Average(Select(source, selector)); } public virtual IObservable Average(IObservable source, Func selector) { return Average(Select(source, selector)); } public virtual IObservable Average(IObservable source, Func selector) { return Average(Select(source, selector)); } public virtual IObservable Average(IObservable source, Func selector) { return Average(Select(source, selector)); } public virtual IObservable Average(IObservable source, Func selector) { return Average(Select(source, selector)); } public virtual IObservable Average(IObservable source, Func selector) { return Average(Select(source, selector)); } public virtual IObservable Average(IObservable source, Func selector) { return Average(Select(source, selector)); } public virtual IObservable Average(IObservable source, Func selector) { return Average(Select(source, selector)); } #endregion #region + All + public virtual IObservable All(IObservable source, Func predicate) { #if !NO_PERF return new All(source, predicate); #else return source.Where(v => !(predicate(v))).Any().Select(b => !b); #endif } #endregion #region + Any + public virtual IObservable Any(IObservable source) { #if !NO_PERF return new Any(source); #else return new AnonymousObservable(observer => source.Subscribe( _ => { observer.OnNext(true); observer.OnCompleted(); }, observer.OnError, () => { observer.OnNext(false); observer.OnCompleted(); })); #endif } public virtual IObservable Any(IObservable source, Func predicate) { #if !NO_PERF return new Any(source, predicate); #else return source.Where(predicate).Any(); #endif } #endregion #region + Average + public virtual IObservable Average(IObservable 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 Average(IObservable 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 Average(IObservable 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 Average(IObservable 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 Average(IObservable 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 Average(IObservable 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 Average(IObservable 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 Average(IObservable 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 Average(IObservable 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 Average(IObservable 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 Contains(IObservable source, TSource value) { #if !NO_PERF return new Contains(source, value, EqualityComparer.Default); #else return Contains_(source, value, EqualityComparer.Default); #endif } public virtual IObservable Contains(IObservable source, TSource value, IEqualityComparer comparer) { #if !NO_PERF return new Contains(source, value, comparer); #else return Contains_(source, value, comparer); #endif } #if NO_PERF private static IObservable Contains_(IObservable source, TSource value, IEqualityComparer comparer) { return source.Where(v => comparer.Equals(v, value)).Any(); } #endif #endregion #region + Count + public virtual IObservable Count(IObservable source) { #if !NO_PERF return new Count(source); #else return source.Aggregate(0, (count, _) => checked(count + 1)); #endif } public virtual IObservable Count(IObservable source, Func predicate) { #if !NO_PERF return new Count(source, predicate); #else return source.Where(predicate).Aggregate(0, (count, _) => checked(count + 1)); #endif } #endregion #region + ElementAt + public virtual IObservable ElementAt(IObservable source, int index) { #if !NO_PERF return new ElementAt(source, index, true); #else return new AnonymousObservable(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 ElementAtOrDefault(IObservable source, int index) { #if !NO_PERF return new ElementAt(source, index, false); #else return new AnonymousObservable(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 FirstAsync(IObservable source) { #if !NO_PERF return new FirstAsync(source, null, true); #else return FirstOrDefaultAsync_(source, true); #endif } public virtual IObservable FirstAsync(IObservable source, Func predicate) { #if !NO_PERF return new FirstAsync(source, predicate, true); #else return source.Where(predicate).FirstAsync(); #endif } #endregion #region + FirstAsyncOrDefaultAsync + public virtual IObservable FirstOrDefaultAsync(IObservable source) { #if !NO_PERF return new FirstAsync(source, null, false); #else return FirstOrDefaultAsync_(source, false); #endif } public virtual IObservable FirstOrDefaultAsync(IObservable source, Func predicate) { #if !NO_PERF return new FirstAsync(source, predicate, false); #else return source.Where(predicate).FirstOrDefaultAsync(); #endif } #if NO_PERF private static IObservable FirstOrDefaultAsync_(IObservable source, bool throwOnEmpty) { return new AnonymousObservable(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 IsEmpty(IObservable source) { #if !NO_PERF return new IsEmpty(source); #else return source.Any().Select(b => !b); #endif } #endregion #region + LastAsync + public virtual IObservable LastAsync(IObservable source) { #if !NO_PERF return new LastAsync(source, null, true); #else return LastOrDefaultAsync_(source, true); #endif } public virtual IObservable LastAsync(IObservable source, Func predicate) { #if !NO_PERF return new LastAsync(source, predicate, true); #else return source.Where(predicate).LastAsync(); #endif } #endregion #region + LastOrDefaultAsync + public virtual IObservable LastOrDefaultAsync(IObservable source) { #if !NO_PERF return new LastAsync(source, null, false); #else return LastOrDefaultAsync_(source, false); #endif } public virtual IObservable LastOrDefaultAsync(IObservable source, Func predicate) { #if !NO_PERF return new LastAsync(source, predicate, false); #else return source.Where(predicate).LastOrDefaultAsync(); #endif } #if NO_PERF private static IObservable LastOrDefaultAsync_(IObservable source, bool throwOnEmpty) { return new AnonymousObservable(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 LongCount(IObservable source) { #if !NO_PERF return new LongCount(source); #else return source.Aggregate(0L, (count, _) => checked(count + 1)); #endif } public virtual IObservable LongCount(IObservable source, Func predicate) { #if !NO_PERF return new LongCount(source, predicate); #else return source.Where(predicate).Aggregate(0L, (count, _) => checked(count + 1)); #endif } #endregion #region + Max + public virtual IObservable Max(IObservable source) { #if !NO_PERF // BREAKING CHANGE v2 > v1.x - Behavior for reference types return new Max(source, Comparer.Default); #else return MaxBy(source, x => x).Select(x => x.First()); #endif } public virtual IObservable Max(IObservable source, IComparer comparer) { #if !NO_PERF // BREAKING CHANGE v2 > v1.x - Behavior for reference types return new Max(source, comparer); #else return MaxBy(source, x => x, comparer).Select(x => x.First()); #endif } public virtual IObservable Max(IObservable source) { #if !NO_PERF return new MaxDouble(source); #else return source.Scan(double.MinValue, Math.Max).Final(); #endif } public virtual IObservable Max(IObservable source) { #if !NO_PERF return new MaxSingle(source); #else return source.Scan(float.MinValue, Math.Max).Final(); #endif } public virtual IObservable Max(IObservable source) { #if !NO_PERF return new MaxDecimal(source); #else return source.Scan(decimal.MinValue, Math.Max).Final(); #endif } public virtual IObservable Max(IObservable source) { #if !NO_PERF return new MaxInt32(source); #else return source.Scan(int.MinValue, Math.Max).Final(); #endif } public virtual IObservable Max(IObservable source) { #if !NO_PERF return new MaxInt64(source); #else return source.Scan(long.MinValue, Math.Max).Final(); #endif } public virtual IObservable Max(IObservable source) { #if !NO_PERF return new MaxDoubleNullable(source); #else return source.Aggregate(new double?(), NullableMax); #endif } public virtual IObservable Max(IObservable source) { #if !NO_PERF return new MaxSingleNullable(source); #else return source.Aggregate(new float?(), NullableMax); #endif } public virtual IObservable Max(IObservable source) { #if !NO_PERF return new MaxDecimalNullable(source); #else return source.Aggregate(new decimal?(), NullableMax); #endif } public virtual IObservable Max(IObservable source) { #if !NO_PERF return new MaxInt32Nullable(source); #else return source.Aggregate(new int?(), NullableMax); #endif } public virtual IObservable Max(IObservable source) { #if !NO_PERF return new MaxInt64Nullable(source); #else return source.Aggregate(new long?(), NullableMax); #endif } public virtual IObservable Max(IObservable source, Func selector) { return Max(Select(source, selector)); } public virtual IObservable Max(IObservable source, Func selector, IComparer comparer) { return Max(Select(source, selector), comparer); } public virtual IObservable Max(IObservable source, Func selector) { return Max(Select(source, selector)); } public virtual IObservable Max(IObservable source, Func selector) { return Max(Select(source, selector)); } public virtual IObservable Max(IObservable source, Func selector) { return Max(Select(source, selector)); } public virtual IObservable Max(IObservable source, Func selector) { return Max(Select(source, selector)); } public virtual IObservable Max(IObservable source, Func selector) { return Max(Select(source, selector)); } public virtual IObservable Max(IObservable source, Func selector) { return Max(Select(source, selector)); } public virtual IObservable Max(IObservable source, Func selector) { return Max(Select(source, selector)); } public virtual IObservable Max(IObservable source, Func selector) { return Max(Select(source, selector)); } public virtual IObservable Max(IObservable source, Func selector) { return Max(Select(source, selector)); } public virtual IObservable Max(IObservable source, Func selector) { return Max(Select(source, selector)); } #endregion #region + MaxBy + public virtual IObservable> MaxBy(IObservable source, Func keySelector) { #if !NO_PERF return new MaxBy(source, keySelector, Comparer.Default); #else return MaxBy(source, keySelector, Comparer.Default); #endif } public virtual IObservable> MaxBy(IObservable source, Func keySelector, IComparer comparer) { #if !NO_PERF return new MaxBy(source, keySelector, comparer); #else return ExtremaBy(source, keySelector, comparer); #endif } #endregion #region + Min + public virtual IObservable Min(IObservable source) { #if !NO_PERF // BREAKING CHANGE v2 > v1.x - Behavior for reference types return new Min(source, Comparer.Default); #else return MinBy(source, x => x).Select(x => x.First()); #endif } public virtual IObservable Min(IObservable source, IComparer comparer) { #if !NO_PERF // BREAKING CHANGE v2 > v1.x - Behavior for reference types return new Min(source, comparer); #else return MinBy(source, x => x, comparer).Select(x => x.First()); #endif } public virtual IObservable Min(IObservable source) { #if !NO_PERF return new MinDouble(source); #else return source.Scan(double.MaxValue, Math.Min).Final(); #endif } public virtual IObservable Min(IObservable source) { #if !NO_PERF return new MinSingle(source); #else return source.Scan(float.MaxValue, Math.Min).Final(); #endif } public virtual IObservable Min(IObservable source) { #if !NO_PERF return new MinDecimal(source); #else return source.Scan(decimal.MaxValue, Math.Min).Final(); #endif } public virtual IObservable Min(IObservable source) { #if !NO_PERF return new MinInt32(source); #else return source.Scan(int.MaxValue, Math.Min).Final(); #endif } public virtual IObservable Min(IObservable source) { #if !NO_PERF return new MinInt64(source); #else return source.Scan(long.MaxValue, Math.Min).Final(); #endif } public virtual IObservable Min(IObservable source) { #if !NO_PERF return new MinDoubleNullable(source); #else return source.Aggregate(new double?(), NullableMin); #endif } public virtual IObservable Min(IObservable source) { #if !NO_PERF return new MinSingleNullable(source); #else return source.Aggregate(new float?(), NullableMin); #endif } public virtual IObservable Min(IObservable source) { #if !NO_PERF return new MinDecimalNullable(source); #else return source.Aggregate(new decimal?(), NullableMin); #endif } public virtual IObservable Min(IObservable source) { #if !NO_PERF return new MinInt32Nullable(source); #else return source.Aggregate(new int?(), NullableMin); #endif } public virtual IObservable Min(IObservable source) { #if !NO_PERF return new MinInt64Nullable(source); #else return source.Aggregate(new long?(), NullableMin); #endif } public virtual IObservable Min(IObservable source, Func selector) { return Min(Select(source, selector)); } public virtual IObservable Min(IObservable source, Func selector, IComparer comparer) { return Min(Select(source, selector), comparer); } public virtual IObservable Min(IObservable source, Func selector) { return Min(Select(source, selector)); } public virtual IObservable Min(IObservable source, Func selector) { return Min(Select(source, selector)); } public virtual IObservable Min(IObservable source, Func selector) { return Min(Select(source, selector)); } public virtual IObservable Min(IObservable source, Func selector) { return Min(Select(source, selector)); } public virtual IObservable Min(IObservable source, Func selector) { return Min(Select(source, selector)); } public virtual IObservable Min(IObservable source, Func selector) { return Min(Select(source, selector)); } public virtual IObservable Min(IObservable source, Func selector) { return Min(Select(source, selector)); } public virtual IObservable Min(IObservable source, Func selector) { return Min(Select(source, selector)); } public virtual IObservable Min(IObservable source, Func selector) { return Min(Select(source, selector)); } public virtual IObservable Min(IObservable source, Func selector) { return Min(Select(source, selector)); } #endregion #region + MinBy + public virtual IObservable> MinBy(IObservable source, Func keySelector) { #if !NO_PERF return new MinBy(source, keySelector, Comparer.Default); #else return MinBy(source, keySelector, Comparer.Default); #endif } public virtual IObservable> MinBy(IObservable source, Func keySelector, IComparer comparer) { #if !NO_PERF return new MinBy(source, keySelector, comparer); #else return ExtremaBy(source, keySelector, new AnonymousComparer((x, y) => comparer.Compare(x, y) * -1)); #endif } #endregion #region + SequenceEqual + public virtual IObservable SequenceEqual(IObservable first, IObservable second) { #if !NO_PERF return new SequenceEqual(first, second, EqualityComparer.Default); #else return first.SequenceEqual(second, EqualityComparer.Default); #endif } public virtual IObservable SequenceEqual(IObservable first, IObservable second, IEqualityComparer comparer) { #if !NO_PERF return new SequenceEqual(first, second, comparer); #else return new AnonymousObservable(observer => { var gate = new object(); var donel = false; var doner = false; var ql = new Queue(); var qr = new Queue(); 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 SequenceEqual(IObservable first, IEnumerable second) { #if !NO_PERF return new SequenceEqual(first, second, EqualityComparer.Default); #else return SequenceEqual(first, second, EqualityComparer.Default); #endif } public virtual IObservable SequenceEqual(IObservable first, IEnumerable second, IEqualityComparer comparer) { #if !NO_PERF return new SequenceEqual(first, second, comparer); #else return new AnonymousObservable(observer => { var e = default(IEnumerator); 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 SingleAsync(IObservable source) { #if !NO_PERF return new SingleAsync(source, null, true); #else return SingleOrDefaultAsync_(source, true); #endif } public virtual IObservable SingleAsync(IObservable source, Func predicate) { #if !NO_PERF return new SingleAsync(source, predicate, true); #else return source.Where(predicate).SingleAsync(); #endif } #endregion #region + SingleOrDefaultAsync + public virtual IObservable SingleOrDefaultAsync(IObservable source) { #if !NO_PERF return new SingleAsync(source, null, false); #else return SingleOrDefaultAsync_(source, false); #endif } public virtual IObservable SingleOrDefaultAsync(IObservable source, Func predicate) { #if !NO_PERF return new SingleAsync(source, predicate, false); #else return source.Where(predicate).SingleOrDefaultAsync(); #endif } #if NO_PERF private static IObservable SingleOrDefaultAsync_(IObservable source, bool throwOnEmpty) { return new AnonymousObservable(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 Sum(IObservable source) { #if !NO_PERF return new SumDouble(source); #else return source.Aggregate(0.0, (prev, curr) => prev + curr); #endif } public virtual IObservable Sum(IObservable source) { #if !NO_PERF return new SumSingle(source); #else return source.Aggregate(0f, (prev, curr) => prev + curr); #endif } public virtual IObservable Sum(IObservable source) { #if !NO_PERF return new SumDecimal(source); #else return source.Aggregate(0M, (prev, curr) => prev + curr); #endif } public virtual IObservable Sum(IObservable source) { #if !NO_PERF return new SumInt32(source); #else return source.Aggregate(0, (prev, curr) => checked(prev + curr)); #endif } public virtual IObservable Sum(IObservable source) { #if !NO_PERF return new SumInt64(source); #else return source.Aggregate(0L, (prev, curr) => checked(prev + curr)); #endif } public virtual IObservable Sum(IObservable 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 Sum(IObservable 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 Sum(IObservable 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 Sum(IObservable 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 Sum(IObservable 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 Sum(IObservable source, Func selector) { return Sum(Select(source, selector)); } public virtual IObservable Sum(IObservable source, Func selector) { return Sum(Select(source, selector)); } public virtual IObservable Sum(IObservable source, Func selector) { return Sum(Select(source, selector)); } public virtual IObservable Sum(IObservable source, Func selector) { return Sum(Select(source, selector)); } public virtual IObservable Sum(IObservable source, Func selector) { return Sum(Select(source, selector)); } public virtual IObservable Sum(IObservable source, Func selector) { return Sum(Select(source, selector)); } public virtual IObservable Sum(IObservable source, Func selector) { return Sum(Select(source, selector)); } public virtual IObservable Sum(IObservable source, Func selector) { return Sum(Select(source, selector)); } public virtual IObservable Sum(IObservable source, Func selector) { return Sum(Select(source, selector)); } public virtual IObservable Sum(IObservable source, Func selector) { return Sum(Select(source, selector)); } #endregion #region + ToArray + public virtual IObservable ToArray(IObservable source) { #if !NO_PERF return new ToArray(source); #else return source.ToList().Select(xs => xs.ToArray()); #endif } #endregion #region + ToDictionary + public virtual IObservable> ToDictionary(IObservable source, Func keySelector, Func elementSelector, IEqualityComparer comparer) { #if !NO_PERF return new ToDictionary(source, keySelector, elementSelector, comparer); #else return source.Aggregate((IDictionary)new Dictionary(comparer), (dict, x) => { dict.Add(keySelector(x), elementSelector(x)); return dict; }); #endif } public virtual IObservable> ToDictionary(IObservable source, Func keySelector, Func elementSelector) { #if !NO_PERF return new ToDictionary(source, keySelector, elementSelector, EqualityComparer.Default); #else return source.ToDictionary(keySelector, elementSelector, EqualityComparer.Default); #endif } public virtual IObservable> ToDictionary(IObservable source, Func keySelector, IEqualityComparer comparer) { #if !NO_PERF return new ToDictionary(source, keySelector, x => x, comparer); #else return source.ToDictionary(keySelector, x => x, comparer); #endif } public virtual IObservable> ToDictionary(IObservable source, Func keySelector) { #if !NO_PERF return new ToDictionary(source, keySelector, x => x, EqualityComparer.Default); #else return source.ToDictionary(keySelector, x => x, EqualityComparer.Default); #endif } #endregion #region + ToList + public virtual IObservable> ToList(IObservable source) { #if !NO_PERF return new ToList(source); #else return source.Aggregate((IList)new List(), (list, x) => { list.Add(x); return list; }); #endif } #endregion #region + ToLookup + public virtual IObservable> ToLookup(IObservable source, Func keySelector, Func elementSelector, IEqualityComparer comparer) { #if !NO_PERF return new ToLookup(source, keySelector, elementSelector, comparer); #else return source.Aggregate(new Lookup(comparer), (lookup, x) => { lookup.Add(keySelector(x), elementSelector(x)); return lookup; }).Select(xs => (ILookup)xs); #endif } public virtual IObservable> ToLookup(IObservable source, Func keySelector, IEqualityComparer comparer) { #if !NO_PERF return new ToLookup(source, keySelector, x => x, comparer); #else return source.ToLookup(keySelector, x => x, comparer); #endif } public virtual IObservable> ToLookup(IObservable source, Func keySelector, Func elementSelector) { #if !NO_PERF return new ToLookup(source, keySelector, elementSelector, EqualityComparer.Default); #else return source.ToLookup(keySelector, elementSelector, EqualityComparer.Default); #endif } public virtual IObservable> ToLookup(IObservable source, Func keySelector) { #if !NO_PERF return new ToLookup(source, keySelector, x => x, EqualityComparer.Default); #else return source.ToLookup(keySelector, x => x, EqualityComparer.Default); #endif } #endregion #region |> Helpers <| #if NO_PERF private static T? NullableMin(T? x, T? y) where T : struct, IComparable { 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? x, T? y) where T : struct, IComparable { if (!x.HasValue) return y; if (!y.HasValue) return x; if (x.Value.CompareTo(y.Value) >= 0) return x; return y; } private static IObservable> ExtremaBy(IObservable source, Func keySelector, IComparer comparer) { return new AnonymousObservable>(observer => { var hasValue = false; var lastKey = default(TKey); var list = new List(); 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 Final(this IObservable source) { return new AnonymousObservable(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 : IComparer { private readonly Func comparer; /// /// Creates an instance of IComparer by providing a method that compares two objects. /// public AnonymousComparer(Func comparer) { this.comparer = comparer; } /// /// Compares two objects and returns a value indicating whether one is less than, equal to, or greater than the other. /// public int Compare(T x, T y) { return comparer(x, y); } } #endif #endregion }