// 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; using System.Collections.Generic; using System.Linq; using System.Threading; using System.Threading.Tasks; namespace System.Linq { public static partial class AsyncEnumerable { public static IAsyncEnumerable GroupJoin(this IAsyncEnumerable outer, IAsyncEnumerable inner, Func outerKeySelector, Func innerKeySelector, Func, TResult> 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 GroupJoinAsyncEnumerable(outer, inner, outerKeySelector, innerKeySelector, resultSelector, comparer); } public static IAsyncEnumerable GroupJoin(this IAsyncEnumerable outer, IAsyncEnumerable inner, Func outerKeySelector, Func innerKeySelector, Func, TResult> 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 outer.GroupJoin(inner, outerKeySelector, innerKeySelector, resultSelector, EqualityComparer.Default); } private sealed class AsyncEnumerableAdapter : IAsyncEnumerable { private readonly IEnumerable _source; public AsyncEnumerableAdapter(IEnumerable source) { _source = source; } public IAsyncEnumerator GetEnumerator() => new AsyncEnumeratorAdapter(_source.GetEnumerator()); private sealed class AsyncEnumeratorAdapter : IAsyncEnumerator { private readonly IEnumerator _enumerator; public AsyncEnumeratorAdapter(IEnumerator enumerator) { _enumerator = enumerator; } public Task MoveNext(CancellationToken cancellationToken) { cancellationToken.ThrowIfCancellationRequested(); #if HAS_AWAIT return Task.FromResult(_enumerator.MoveNext()); #else return TaskEx.FromResult(_enumerator.MoveNext()); #endif } public T Current => _enumerator.Current; public void Dispose() => _enumerator.Dispose(); } } private sealed class GroupJoinAsyncEnumerable : IAsyncEnumerable { private readonly IEqualityComparer _comparer; private readonly IAsyncEnumerable _inner; private readonly Func _innerKeySelector; private readonly IAsyncEnumerable _outer; private readonly Func _outerKeySelector; private readonly Func, TResult> _resultSelector; public GroupJoinAsyncEnumerable( IAsyncEnumerable outer, IAsyncEnumerable inner, Func outerKeySelector, Func innerKeySelector, Func, TResult> resultSelector, IEqualityComparer comparer) { _outer = outer; _inner = inner; _outerKeySelector = outerKeySelector; _innerKeySelector = innerKeySelector; _resultSelector = resultSelector; _comparer = comparer; } public IAsyncEnumerator GetEnumerator() => new GroupJoinAsyncEnumerator( _outer.GetEnumerator(), _inner, _outerKeySelector, _innerKeySelector, _resultSelector, _comparer); private sealed class GroupJoinAsyncEnumerator : IAsyncEnumerator { private readonly IEqualityComparer _comparer; private readonly IAsyncEnumerable _inner; private readonly Func _innerKeySelector; private readonly IAsyncEnumerator _outer; private readonly Func _outerKeySelector; private readonly Func, TResult> _resultSelector; private Internal.Lookup _lookup; public GroupJoinAsyncEnumerator( IAsyncEnumerator outer, IAsyncEnumerable inner, Func outerKeySelector, Func innerKeySelector, Func, TResult> resultSelector, IEqualityComparer comparer) { _outer = outer; _inner = inner; _outerKeySelector = outerKeySelector; _innerKeySelector = innerKeySelector; _resultSelector = resultSelector; _comparer = comparer; } public async Task MoveNext(CancellationToken cancellationToken) { // nothing to do if (!await _outer.MoveNext(cancellationToken) .ConfigureAwait(false)) { return false; } if (_lookup == null) { _lookup = await Internal.Lookup.CreateForJoinAsync(_inner, _innerKeySelector, _comparer, cancellationToken) .ConfigureAwait(false); } var item = _outer.Current; Current = _resultSelector(item, new AsyncEnumerableAdapter(_lookup[_outerKeySelector(item)])); return true; } public TResult Current { get; private set; } public void Dispose() { _outer.Dispose(); } } } } }