// 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 Join(this IAsyncEnumerable outer, IAsyncEnumerable inner, Func outerKeySelector, Func innerKeySelector, Func resultSelector, IEqualityComparer comparer) { if (outer == null) { throw new ArgumentNullException(nameof(outer)); } if (inner == null) { throw new ArgumentNullException(nameof(inner)); } if (outerKeySelector == null) { throw new ArgumentNullException(nameof(outerKeySelector)); } if (innerKeySelector == null) { throw new ArgumentNullException(nameof(innerKeySelector)); } if (resultSelector == null) { throw new ArgumentNullException(nameof(resultSelector)); } if (comparer == null) { throw new ArgumentNullException(nameof(comparer)); } return new JoinAsyncIterator(outer, inner, outerKeySelector, innerKeySelector, resultSelector, comparer); } public static IAsyncEnumerable Join(this IAsyncEnumerable outer, IAsyncEnumerable inner, Func outerKeySelector, Func innerKeySelector, Func resultSelector) { if (outer == null) { throw new ArgumentNullException(nameof(outer)); } if (inner == null) { throw new ArgumentNullException(nameof(inner)); } if (outerKeySelector == null) { throw new ArgumentNullException(nameof(outerKeySelector)); } if (innerKeySelector == null) { throw new ArgumentNullException(nameof(innerKeySelector)); } if (resultSelector == null) { throw new ArgumentNullException(nameof(resultSelector)); } return new JoinAsyncIterator(outer, inner, outerKeySelector, innerKeySelector, resultSelector, EqualityComparer.Default); } internal sealed class JoinAsyncIterator : AsyncIterator { private readonly IAsyncEnumerable outer; private readonly IAsyncEnumerable inner; private readonly Func outerKeySelector; private readonly Func innerKeySelector; private readonly Func resultSelector; private readonly IEqualityComparer comparer; private IAsyncEnumerator outerEnumerator; public JoinAsyncIterator(IAsyncEnumerable outer, IAsyncEnumerable inner, Func outerKeySelector, Func innerKeySelector, Func resultSelector, IEqualityComparer comparer) { Debug.Assert(outer != null); Debug.Assert(inner != null); Debug.Assert(outerKeySelector != null); Debug.Assert(innerKeySelector != null); Debug.Assert(resultSelector != null); Debug.Assert(comparer != null); this.outer = outer; this.inner = inner; this.outerKeySelector = outerKeySelector; this.innerKeySelector = innerKeySelector; this.resultSelector = resultSelector; this.comparer = comparer; } public override AsyncIterator Clone() { return new JoinAsyncIterator(outer, inner, outerKeySelector, innerKeySelector, resultSelector, comparer); } public override void Dispose() { if (outerEnumerator != null) { outerEnumerator.Dispose(); outerEnumerator = null; } base.Dispose(); } // State machine vars private Internal.Lookup lookup; private int count; private TInner[] elements; private int index; private TOuter item; private int mode; private const int State_If = 1; private const int State_DoLoop = 2; private const int State_For = 3; private const int State_While = 4; protected override async Task MoveNextCore(CancellationToken cancellationToken) { switch (state) { case AsyncIteratorState.Allocated: outerEnumerator = outer.GetEnumerator(); mode = State_If; state = AsyncIteratorState.Iterating; goto case AsyncIteratorState.Iterating; case AsyncIteratorState.Iterating: switch (mode) { case State_If: if (await outerEnumerator.MoveNext(cancellationToken) .ConfigureAwait(false)) { lookup = await Internal.Lookup.CreateForJoinAsync(inner, innerKeySelector, comparer, cancellationToken).ConfigureAwait(false); if (lookup.Count != 0) { mode = State_DoLoop; goto case State_DoLoop; } } break; case State_DoLoop: item = outerEnumerator.Current; var g = lookup.GetGrouping(outerKeySelector(item), create: false); if (g != null) { count = g._count; elements = g._elements; index = 0; mode = State_For; goto case State_For; } // advance to while mode = State_While; goto case State_While; case State_For: current = resultSelector(item, elements[index]); index++; if (index == count) { mode = State_While; } return true; case State_While: var hasNext = await outerEnumerator.MoveNext(cancellationToken).ConfigureAwait(false); if (hasNext) { goto case State_DoLoop; } break; } Dispose(); break; } return false; } } } }