// Licensed to the .NET Foundation under one or more agreements. // The .NET Foundation licenses this file to you under the MIT License. // See the LICENSE file in the project root for more information. using System.Threading.Tasks; namespace System.Reactive.Linq { public partial class AsyncObservable { public static IAsyncObservable LastOrDefaultAsync(this IAsyncObservable source) => LastOrDefault(source); public static IAsyncObservable LastOrDefaultAsync(this IAsyncObservable source, Func predicate) => LastOrDefault(source, predicate); public static IAsyncObservable LastOrDefaultAsync(this IAsyncObservable source, Func> predicate) => LastOrDefault(source, predicate); public static IAsyncObservable LastOrDefault(this IAsyncObservable source) { if (source == null) throw new ArgumentNullException(nameof(source)); return Create( source, (source, observer) => source.SubscribeSafeAsync(AsyncObserver.LastOrDefault(observer))); } public static IAsyncObservable LastOrDefault(this IAsyncObservable source, Func predicate) { if (source == null) throw new ArgumentNullException(nameof(source)); if (predicate == null) throw new ArgumentNullException(nameof(predicate)); return CreateAsyncObservable.From( source, predicate, static (source, predicate, observer) => source.SubscribeSafeAsync(AsyncObserver.LastOrDefault(observer, predicate))); } public static IAsyncObservable LastOrDefault(this IAsyncObservable source, Func> predicate) { if (source == null) throw new ArgumentNullException(nameof(source)); if (predicate == null) throw new ArgumentNullException(nameof(predicate)); return CreateAsyncObservable.From( source, predicate, static (source, predicate, observer) => source.SubscribeSafeAsync(AsyncObserver.LastOrDefault(observer, predicate))); } } public partial class AsyncObserver { public static IAsyncObserver LastOrDefault(IAsyncObserver observer) { if (observer == null) throw new ArgumentNullException(nameof(observer)); var hasValue = false; var lastValue = default(TSource); return Create( x => { hasValue = true; lastValue = x; return default; }, observer.OnErrorAsync, async () => { await observer.OnNextAsync(hasValue ? lastValue : default).ConfigureAwait(false); await observer.OnCompletedAsync().ConfigureAwait(false); } ); } public static IAsyncObserver LastOrDefault(IAsyncObserver observer, Func predicate) { if (observer == null) throw new ArgumentNullException(nameof(observer)); if (predicate == null) throw new ArgumentNullException(nameof(predicate)); return Where(LastOrDefault(observer), predicate); } public static IAsyncObserver LastOrDefault(IAsyncObserver observer, Func> predicate) { if (observer == null) throw new ArgumentNullException(nameof(observer)); if (predicate == null) throw new ArgumentNullException(nameof(predicate)); return Where(LastOrDefault(observer), predicate); } } }