123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662 |
- // 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 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
- }
|