// 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.Collections.Generic; using System.Threading.Tasks; namespace System.Linq { public static partial class AsyncEnumerableEx { /// /// Concatenates the second async-enumerable sequence to the first async-enumerable sequence upon successful or exceptional termination of the first. /// /// The type of the elements in the source sequences. /// First async-enumerable sequence whose exception (if any) is caught. /// Second async-enumerable sequence used to produce results after the first sequence terminates. /// An async-enumerable sequence that concatenates the first and second sequence, even if the first sequence terminates exceptionally. /// or is null. public static IAsyncEnumerable OnErrorResumeNext(this IAsyncEnumerable first, IAsyncEnumerable second) { if (first == null) throw Error.ArgumentNull(nameof(first)); if (second == null) throw Error.ArgumentNull(nameof(second)); return OnErrorResumeNextCore(new[] { first, second }); } /// /// Concatenates all of the specified async-enumerable sequences, even if the previous async-enumerable sequence terminated exceptionally. /// /// The type of the elements in the source sequences. /// Observable sequences to concatenate. /// An async-enumerable sequence that concatenates the source sequences, even if a sequence terminates exceptionally. /// is null. public static IAsyncEnumerable OnErrorResumeNext(params IAsyncEnumerable[] sources) { if (sources == null) throw Error.ArgumentNull(nameof(sources)); return OnErrorResumeNextCore(sources); } /// /// Concatenates all async-enumerable sequences in the given enumerable sequence, even if the previous async-enumerable sequence terminated exceptionally. /// /// The type of the elements in the source sequences. /// Observable sequences to concatenate. /// An async-enumerable sequence that concatenates the source sequences, even if a sequence terminates exceptionally. /// is null. public static IAsyncEnumerable OnErrorResumeNext(this IEnumerable> sources) { if (sources == null) throw Error.ArgumentNull(nameof(sources)); return OnErrorResumeNextCore(sources); } private static IAsyncEnumerable OnErrorResumeNextCore(IEnumerable> sources) { return new OnErrorResumeNextAsyncIterator(sources); } private sealed class OnErrorResumeNextAsyncIterator : AsyncIterator { private readonly IEnumerable> _sources; private IAsyncEnumerator? _enumerator; private IEnumerator>? _sourcesEnumerator; public OnErrorResumeNextAsyncIterator(IEnumerable> sources) { _sources = sources; } public override AsyncIteratorBase Clone() { return new OnErrorResumeNextAsyncIterator(_sources); } public override async ValueTask DisposeAsync() { if (_sourcesEnumerator != null) { _sourcesEnumerator.Dispose(); _sourcesEnumerator = null; } if (_enumerator != null) { await _enumerator.DisposeAsync().ConfigureAwait(false); _enumerator = null; } await base.DisposeAsync().ConfigureAwait(false); } protected override async ValueTask MoveNextCore() { switch (_state) { case AsyncIteratorState.Allocated: _sourcesEnumerator = _sources.GetEnumerator(); _state = AsyncIteratorState.Iterating; goto case AsyncIteratorState.Iterating; case AsyncIteratorState.Iterating: while (true) { if (_enumerator == null) { if (!_sourcesEnumerator!.MoveNext()) { break; // while -- done, nothing else to do } _enumerator = _sourcesEnumerator.Current.GetAsyncEnumerator(_cancellationToken); } try { if (await _enumerator.MoveNextAsync().ConfigureAwait(false)) { _current = _enumerator.Current; return true; } } catch { // Ignore } // Done with the current one, go to the next await _enumerator.DisposeAsync().ConfigureAwait(false); _enumerator = null; } break; // case } await DisposeAsync().ConfigureAwait(false); return false; } } } }