// 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.Diagnostics; using System.Threading; using System.Threading.Tasks; namespace System.Linq { public static partial class AsyncEnumerable { public static IAsyncEnumerable Zip(this IAsyncEnumerable first, IAsyncEnumerable second, Func selector) { if (first == null) throw new ArgumentNullException(nameof(first)); if (second == null) throw new ArgumentNullException(nameof(second)); if (selector == null) throw new ArgumentNullException(nameof(selector)); return new ZipAsyncIterator(first, second, selector); } public static IAsyncEnumerable Zip(this IAsyncEnumerable first, IAsyncEnumerable second, Func> selector) { if (first == null) throw new ArgumentNullException(nameof(first)); if (second == null) throw new ArgumentNullException(nameof(second)); if (selector == null) throw new ArgumentNullException(nameof(selector)); return new ZipAsyncIteratorWithTask(first, second, selector); } private sealed class ZipAsyncIterator : AsyncIterator { private readonly IAsyncEnumerable first; private readonly IAsyncEnumerable second; private readonly Func selector; private IAsyncEnumerator firstEnumerator; private IAsyncEnumerator secondEnumerator; public ZipAsyncIterator(IAsyncEnumerable first, IAsyncEnumerable second, Func selector) { Debug.Assert(first != null); Debug.Assert(second != null); Debug.Assert(selector != null); this.first = first; this.second = second; this.selector = selector; } public override AsyncIterator Clone() { return new ZipAsyncIterator(first, second, selector); } public override async ValueTask DisposeAsync() { if (firstEnumerator != null) { await firstEnumerator.DisposeAsync().ConfigureAwait(false); firstEnumerator = null; } if (secondEnumerator != null) { await secondEnumerator.DisposeAsync().ConfigureAwait(false); secondEnumerator = null; } await base.DisposeAsync().ConfigureAwait(false); } protected override async ValueTask MoveNextCore(CancellationToken cancellationToken) { switch (state) { case AsyncIteratorState.Allocated: firstEnumerator = first.GetAsyncEnumerator(cancellationToken); secondEnumerator = second.GetAsyncEnumerator(cancellationToken); state = AsyncIteratorState.Iterating; goto case AsyncIteratorState.Iterating; case AsyncIteratorState.Iterating: // We kick these off and join so they can potentially run in parallel var ft = firstEnumerator.MoveNextAsync(); var st = secondEnumerator.MoveNextAsync(); await Task.WhenAll(ft.AsTask(), st.AsTask()).ConfigureAwait(false); if (ft.Result && st.Result) { current = selector(firstEnumerator.Current, secondEnumerator.Current); return true; } await DisposeAsync().ConfigureAwait(false); break; } return false; } } private sealed class ZipAsyncIteratorWithTask : AsyncIterator { private readonly IAsyncEnumerable first; private readonly IAsyncEnumerable second; private readonly Func> selector; private IAsyncEnumerator firstEnumerator; private IAsyncEnumerator secondEnumerator; public ZipAsyncIteratorWithTask(IAsyncEnumerable first, IAsyncEnumerable second, Func> selector) { Debug.Assert(first != null); Debug.Assert(second != null); Debug.Assert(selector != null); this.first = first; this.second = second; this.selector = selector; } public override AsyncIterator Clone() { return new ZipAsyncIteratorWithTask(first, second, selector); } public override async ValueTask DisposeAsync() { if (firstEnumerator != null) { await firstEnumerator.DisposeAsync().ConfigureAwait(false); firstEnumerator = null; } if (secondEnumerator != null) { await secondEnumerator.DisposeAsync().ConfigureAwait(false); secondEnumerator = null; } await base.DisposeAsync().ConfigureAwait(false); } protected override async ValueTask MoveNextCore(CancellationToken cancellationToken) { switch (state) { case AsyncIteratorState.Allocated: firstEnumerator = first.GetAsyncEnumerator(cancellationToken); secondEnumerator = second.GetAsyncEnumerator(cancellationToken); state = AsyncIteratorState.Iterating; goto case AsyncIteratorState.Iterating; case AsyncIteratorState.Iterating: // We kick these off and join so they can potentially run in parallel var ft = firstEnumerator.MoveNextAsync(); var st = secondEnumerator.MoveNextAsync(); await Task.WhenAll(ft.AsTask(), st.AsTask()).ConfigureAwait(false); if (ft.Result && st.Result) { current = await selector(firstEnumerator.Current, secondEnumerator.Current).ConfigureAwait(false); return true; } await DisposeAsync().ConfigureAwait(false); break; } return false; } } } }