// 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.Threading.Tasks; namespace System.Reactive.Linq { partial class AsyncObservable { public static IAsyncObservable Max(IAsyncObservable source) { if (source == null) throw new ArgumentNullException(nameof(source)); return Create(observer => source.SubscribeSafeAsync(AsyncObserver.Max(observer))); } public static IAsyncObservable Max(IAsyncObservable source, IComparer comparer) { if (source == null) throw new ArgumentNullException(nameof(source)); if (comparer == null) throw new ArgumentNullException(nameof(comparer)); return Create(observer => source.SubscribeSafeAsync(AsyncObserver.Max(observer, comparer))); } } partial class AsyncObserver { public static IAsyncObserver Max(IAsyncObserver observer) { if (observer == null) throw new ArgumentNullException(nameof(observer)); return Max(observer, Comparer.Default); } public static IAsyncObserver Max(IAsyncObserver observer, IComparer comparer) { if (observer == null) throw new ArgumentNullException(nameof(observer)); if (comparer == null) throw new ArgumentNullException(nameof(comparer)); var max = default(TSource); var found = false; return Create( async x => { if (found) { bool isGreater; try { isGreater = comparer.Compare(x, max) > 0; } catch (Exception ex) { await observer.OnErrorAsync(ex).ConfigureAwait(false); return; } if (isGreater) { max = x; } } else { max = x; found = true; } }, observer.OnErrorAsync, async () => { if (!found) { await observer.OnErrorAsync(new InvalidOperationException("The sequence is empty.")).ConfigureAwait(false); } else { await observer.OnNextAsync(max).ConfigureAwait(false); await observer.OnCompletedAsync().ConfigureAwait(false); } } ); } public static IAsyncObserver MaxInt32(IAsyncObserver observer) { if (observer == null) throw new ArgumentNullException(nameof(observer)); var max = 0; var found = false; return Create( x => { if (found) { if (x > max) { max = x; } } else { max = x; found = true; } return Task.CompletedTask; }, observer.OnErrorAsync, async () => { if (!found) { await observer.OnErrorAsync(new InvalidOperationException("The sequence is empty.")).ConfigureAwait(false); } else { await observer.OnNextAsync(max).ConfigureAwait(false); await observer.OnCompletedAsync().ConfigureAwait(false); } } ); } public static IAsyncObserver MaxInt64(IAsyncObserver observer) { if (observer == null) throw new ArgumentNullException(nameof(observer)); var max = 0L; var found = false; return Create( x => { if (found) { if (x > max) { max = x; } } else { max = x; found = true; } return Task.CompletedTask; }, observer.OnErrorAsync, async () => { if (!found) { await observer.OnErrorAsync(new InvalidOperationException("The sequence is empty.")).ConfigureAwait(false); } else { await observer.OnNextAsync(max).ConfigureAwait(false); await observer.OnCompletedAsync().ConfigureAwait(false); } } ); } public static IAsyncObserver MaxSingle(IAsyncObserver observer) { if (observer == null) throw new ArgumentNullException(nameof(observer)); var max = 0.0f; var found = false; return Create( x => { if (found) { if (x > max || double.IsNaN(x)) { max = x; } } else { max = x; found = true; } return Task.CompletedTask; }, observer.OnErrorAsync, async () => { if (!found) { await observer.OnErrorAsync(new InvalidOperationException("The sequence is empty.")).ConfigureAwait(false); } else { await observer.OnNextAsync(max).ConfigureAwait(false); await observer.OnCompletedAsync().ConfigureAwait(false); } } ); } public static IAsyncObserver MaxDouble(IAsyncObserver observer) { if (observer == null) throw new ArgumentNullException(nameof(observer)); var max = 0.0; var found = false; return Create( x => { if (found) { if (x > max || double.IsNaN(x)) { max = x; } } else { max = x; found = true; } return Task.CompletedTask; }, observer.OnErrorAsync, async () => { if (!found) { await observer.OnErrorAsync(new InvalidOperationException("The sequence is empty.")).ConfigureAwait(false); } else { await observer.OnNextAsync(max).ConfigureAwait(false); await observer.OnCompletedAsync().ConfigureAwait(false); } } ); } public static IAsyncObserver MaxDecimal(IAsyncObserver observer) { if (observer == null) throw new ArgumentNullException(nameof(observer)); var max = 0m; var found = false; return Create( x => { if (found) { if (x > max) { max = x; } } else { max = x; found = true; } return Task.CompletedTask; }, observer.OnErrorAsync, async () => { if (!found) { await observer.OnErrorAsync(new InvalidOperationException("The sequence is empty.")).ConfigureAwait(false); } else { await observer.OnNextAsync(max).ConfigureAwait(false); await observer.OnCompletedAsync().ConfigureAwait(false); } } ); } public static IAsyncObserver MaxNullableInt32(IAsyncObserver observer) { if (observer == null) throw new ArgumentNullException(nameof(observer)); var max = default(int?); return Create( x => { if (max == null || x > max) { max = x; } return Task.CompletedTask; }, observer.OnErrorAsync, async () => { await observer.OnNextAsync(max).ConfigureAwait(false); await observer.OnCompletedAsync().ConfigureAwait(false); } ); } public static IAsyncObserver MaxNullableInt64(IAsyncObserver observer) { if (observer == null) throw new ArgumentNullException(nameof(observer)); var max = default(long?); return Create( x => { if (max == null || x > max) { max = x; } return Task.CompletedTask; }, observer.OnErrorAsync, async () => { await observer.OnNextAsync(max).ConfigureAwait(false); await observer.OnCompletedAsync().ConfigureAwait(false); } ); } public static IAsyncObserver MaxNullableSingle(IAsyncObserver observer) { if (observer == null) throw new ArgumentNullException(nameof(observer)); var max = default(float?); return Create( x => { if (x != null && (max == null || x > max || double.IsNaN(x.Value))) { max = x; } return Task.CompletedTask; }, observer.OnErrorAsync, async () => { await observer.OnNextAsync(max).ConfigureAwait(false); await observer.OnCompletedAsync().ConfigureAwait(false); } ); } public static IAsyncObserver MaxNullableDouble(IAsyncObserver observer) { if (observer == null) throw new ArgumentNullException(nameof(observer)); var max = default(double?); return Create( x => { if (x != null && (max == null || x > max || double.IsNaN(x.Value))) { max = x; } return Task.CompletedTask; }, observer.OnErrorAsync, async () => { await observer.OnNextAsync(max).ConfigureAwait(false); await observer.OnCompletedAsync().ConfigureAwait(false); } ); } public static IAsyncObserver MaxNullableDecimal(IAsyncObserver observer) { if (observer == null) throw new ArgumentNullException(nameof(observer)); var max = default(decimal?); return Create( x => { if (max == null || x > max) { max = x; } return Task.CompletedTask; }, observer.OnErrorAsync, async () => { await observer.OnNextAsync(max).ConfigureAwait(false); await observer.OnCompletedAsync().ConfigureAwait(false); } ); } } }