// 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.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) => GroupJoin(outer, inner, outerKeySelector, innerKeySelector, resultSelector, comparer: null); public static IAsyncEnumerable GroupJoin(this IAsyncEnumerable outer, IAsyncEnumerable inner, Func outerKeySelector, Func innerKeySelector, Func, TResult> resultSelector, IEqualityComparer comparer) { if (outer == null) throw Error.ArgumentNull(nameof(outer)); if (inner == null) throw Error.ArgumentNull(nameof(inner)); if (outerKeySelector == null) throw Error.ArgumentNull(nameof(outerKeySelector)); if (innerKeySelector == null) throw Error.ArgumentNull(nameof(innerKeySelector)); if (resultSelector == null) throw Error.ArgumentNull(nameof(resultSelector)); #if USE_ASYNC_ITERATOR return Create(Core); async IAsyncEnumerator Core(CancellationToken cancellationToken) { await using (var e = outer.GetAsyncEnumerator(cancellationToken).ConfigureAwait(false)) { if (await e.MoveNextAsync()) { var lookup = await Internal.Lookup.CreateForJoinAsync(inner, innerKeySelector, comparer, cancellationToken).ConfigureAwait(false); do { var item = e.Current; var outerKey = outerKeySelector(item); yield return resultSelector(item, lookup[outerKey].ToAsyncEnumerable()); } while (await e.MoveNextAsync()); } } } #else return new GroupJoinAsyncEnumerable(outer, inner, outerKeySelector, innerKeySelector, resultSelector, comparer); #endif } public static IAsyncEnumerable GroupJoin(this IAsyncEnumerable outer, IAsyncEnumerable inner, Func> outerKeySelector, Func> innerKeySelector, Func, ValueTask> resultSelector) => GroupJoin(outer, inner, outerKeySelector, innerKeySelector, resultSelector, comparer: null); public static IAsyncEnumerable GroupJoin(this IAsyncEnumerable outer, IAsyncEnumerable inner, Func> outerKeySelector, Func> innerKeySelector, Func, ValueTask> resultSelector, IEqualityComparer comparer) { if (outer == null) throw Error.ArgumentNull(nameof(outer)); if (inner == null) throw Error.ArgumentNull(nameof(inner)); if (outerKeySelector == null) throw Error.ArgumentNull(nameof(outerKeySelector)); if (innerKeySelector == null) throw Error.ArgumentNull(nameof(innerKeySelector)); if (resultSelector == null) throw Error.ArgumentNull(nameof(resultSelector)); #if USE_ASYNC_ITERATOR return Create(Core); async IAsyncEnumerator Core(CancellationToken cancellationToken) { await using (var e = outer.GetAsyncEnumerator(cancellationToken).ConfigureAwait(false)) { if (await e.MoveNextAsync()) { var lookup = await Internal.LookupWithTask.CreateForJoinAsync(inner, innerKeySelector, comparer, cancellationToken).ConfigureAwait(false); do { var item = e.Current; var outerKey = await outerKeySelector(item).ConfigureAwait(false); yield return await resultSelector(item, lookup[outerKey].ToAsyncEnumerable()).ConfigureAwait(false); } while (await e.MoveNextAsync()); } } } #else return new GroupJoinAsyncEnumerableWithTask(outer, inner, outerKeySelector, innerKeySelector, resultSelector, comparer); #endif } #if !NO_DEEP_CANCELLATION public static IAsyncEnumerable GroupJoin(this IAsyncEnumerable outer, IAsyncEnumerable inner, Func> outerKeySelector, Func> innerKeySelector, Func, CancellationToken, ValueTask> resultSelector) => GroupJoin(outer, inner, outerKeySelector, innerKeySelector, resultSelector, comparer: null); public static IAsyncEnumerable GroupJoin(this IAsyncEnumerable outer, IAsyncEnumerable inner, Func> outerKeySelector, Func> innerKeySelector, Func, CancellationToken, ValueTask> resultSelector, IEqualityComparer comparer) { if (outer == null) throw Error.ArgumentNull(nameof(outer)); if (inner == null) throw Error.ArgumentNull(nameof(inner)); if (outerKeySelector == null) throw Error.ArgumentNull(nameof(outerKeySelector)); if (innerKeySelector == null) throw Error.ArgumentNull(nameof(innerKeySelector)); if (resultSelector == null) throw Error.ArgumentNull(nameof(resultSelector)); #if USE_ASYNC_ITERATOR return Create(Core); async IAsyncEnumerator Core(CancellationToken cancellationToken) { await using (var e = outer.GetAsyncEnumerator(cancellationToken).ConfigureAwait(false)) { if (await e.MoveNextAsync()) { var lookup = await Internal.LookupWithTask.CreateForJoinAsync(inner, innerKeySelector, comparer, cancellationToken).ConfigureAwait(false); do { var item = e.Current; var outerKey = await outerKeySelector(item, cancellationToken).ConfigureAwait(false); yield return await resultSelector(item, lookup[outerKey].ToAsyncEnumerable(), cancellationToken).ConfigureAwait(false); } while (await e.MoveNextAsync()); } } } #else return new GroupJoinAsyncEnumerableWithTaskAndCancellation(outer, inner, outerKeySelector, innerKeySelector, resultSelector, comparer); #endif } #endif #if !USE_ASYNC_ITERATOR 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 GetAsyncEnumerator(CancellationToken cancellationToken) { cancellationToken.ThrowIfCancellationRequested(); // NB: [LDM-2018-11-28] Equivalent to async iterator behavior. return new GroupJoinAsyncEnumerator( _outer.GetAsyncEnumerator(cancellationToken), _inner, _outerKeySelector, _innerKeySelector, _resultSelector, _comparer, cancellationToken); } 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 readonly CancellationToken _cancellationToken; private Internal.Lookup _lookup; public GroupJoinAsyncEnumerator( IAsyncEnumerator outer, IAsyncEnumerable inner, Func outerKeySelector, Func innerKeySelector, Func, TResult> resultSelector, IEqualityComparer comparer, CancellationToken cancellationToken) { _outer = outer; _inner = inner; _outerKeySelector = outerKeySelector; _innerKeySelector = innerKeySelector; _resultSelector = resultSelector; _comparer = comparer; _cancellationToken = cancellationToken; } public async ValueTask MoveNextAsync() { // nothing to do if (!await _outer.MoveNextAsync().ConfigureAwait(false)) { return false; } if (_lookup == null) { _lookup = await Internal.Lookup.CreateForJoinAsync(_inner, _innerKeySelector, _comparer, _cancellationToken).ConfigureAwait(false); } var item = _outer.Current; var outerKey = _outerKeySelector(item); var inner = _lookup[outerKey].ToAsyncEnumerable(); Current = _resultSelector(item, inner); return true; } public TResult Current { get; private set; } public ValueTask DisposeAsync() => _outer.DisposeAsync(); } } private sealed class GroupJoinAsyncEnumerableWithTask : IAsyncEnumerable { private readonly IEqualityComparer _comparer; private readonly IAsyncEnumerable _inner; private readonly Func> _innerKeySelector; private readonly IAsyncEnumerable _outer; private readonly Func> _outerKeySelector; private readonly Func, ValueTask> _resultSelector; public GroupJoinAsyncEnumerableWithTask( IAsyncEnumerable outer, IAsyncEnumerable inner, Func> outerKeySelector, Func> innerKeySelector, Func, ValueTask> resultSelector, IEqualityComparer comparer) { _outer = outer; _inner = inner; _outerKeySelector = outerKeySelector; _innerKeySelector = innerKeySelector; _resultSelector = resultSelector; _comparer = comparer; } public IAsyncEnumerator GetAsyncEnumerator(CancellationToken cancellationToken) { cancellationToken.ThrowIfCancellationRequested(); // NB: [LDM-2018-11-28] Equivalent to async iterator behavior. return new GroupJoinAsyncEnumeratorWithTask( _outer.GetAsyncEnumerator(cancellationToken), _inner, _outerKeySelector, _innerKeySelector, _resultSelector, _comparer, cancellationToken); } private sealed class GroupJoinAsyncEnumeratorWithTask : IAsyncEnumerator { private readonly IEqualityComparer _comparer; private readonly IAsyncEnumerable _inner; private readonly Func> _innerKeySelector; private readonly IAsyncEnumerator _outer; private readonly Func> _outerKeySelector; private readonly Func, ValueTask> _resultSelector; private readonly CancellationToken _cancellationToken; private Internal.LookupWithTask _lookup; public GroupJoinAsyncEnumeratorWithTask( IAsyncEnumerator outer, IAsyncEnumerable inner, Func> outerKeySelector, Func> innerKeySelector, Func, ValueTask> resultSelector, IEqualityComparer comparer, CancellationToken cancellationToken) { _outer = outer; _inner = inner; _outerKeySelector = outerKeySelector; _innerKeySelector = innerKeySelector; _resultSelector = resultSelector; _comparer = comparer; _cancellationToken = cancellationToken; } public async ValueTask MoveNextAsync() { // nothing to do if (!await _outer.MoveNextAsync().ConfigureAwait(false)) { return false; } if (_lookup == null) { _lookup = await Internal.LookupWithTask.CreateForJoinAsync(_inner, _innerKeySelector, _comparer, _cancellationToken).ConfigureAwait(false); } var item = _outer.Current; var outerKey = await _outerKeySelector(item).ConfigureAwait(false); var inner = _lookup[outerKey].ToAsyncEnumerable(); Current = await _resultSelector(item, inner).ConfigureAwait(false); return true; } public TResult Current { get; private set; } public ValueTask DisposeAsync() => _outer.DisposeAsync(); } } #if !NO_DEEP_CANCELLATION private sealed class GroupJoinAsyncEnumerableWithTaskAndCancellation : IAsyncEnumerable { private readonly IEqualityComparer _comparer; private readonly IAsyncEnumerable _inner; private readonly Func> _innerKeySelector; private readonly IAsyncEnumerable _outer; private readonly Func> _outerKeySelector; private readonly Func, CancellationToken, ValueTask> _resultSelector; public GroupJoinAsyncEnumerableWithTaskAndCancellation( IAsyncEnumerable outer, IAsyncEnumerable inner, Func> outerKeySelector, Func> innerKeySelector, Func, CancellationToken, ValueTask> resultSelector, IEqualityComparer comparer) { _outer = outer; _inner = inner; _outerKeySelector = outerKeySelector; _innerKeySelector = innerKeySelector; _resultSelector = resultSelector; _comparer = comparer; } public IAsyncEnumerator GetAsyncEnumerator(CancellationToken cancellationToken) { cancellationToken.ThrowIfCancellationRequested(); // NB: [LDM-2018-11-28] Equivalent to async iterator behavior. return new GroupJoinAsyncEnumeratorWithTask( _outer.GetAsyncEnumerator(cancellationToken), _inner, _outerKeySelector, _innerKeySelector, _resultSelector, _comparer, cancellationToken); } private sealed class GroupJoinAsyncEnumeratorWithTask : IAsyncEnumerator { private readonly IEqualityComparer _comparer; private readonly IAsyncEnumerable _inner; private readonly Func> _innerKeySelector; private readonly IAsyncEnumerator _outer; private readonly Func> _outerKeySelector; private readonly Func, CancellationToken, ValueTask> _resultSelector; private readonly CancellationToken _cancellationToken; private Internal.LookupWithTask _lookup; public GroupJoinAsyncEnumeratorWithTask( IAsyncEnumerator outer, IAsyncEnumerable inner, Func> outerKeySelector, Func> innerKeySelector, Func, CancellationToken, ValueTask> resultSelector, IEqualityComparer comparer, CancellationToken cancellationToken) { _outer = outer; _inner = inner; _outerKeySelector = outerKeySelector; _innerKeySelector = innerKeySelector; _resultSelector = resultSelector; _comparer = comparer; _cancellationToken = cancellationToken; } public async ValueTask MoveNextAsync() { // nothing to do if (!await _outer.MoveNextAsync().ConfigureAwait(false)) { return false; } if (_lookup == null) { _lookup = await Internal.LookupWithTask.CreateForJoinAsync(_inner, _innerKeySelector, _comparer, _cancellationToken).ConfigureAwait(false); } var item = _outer.Current; var outerKey = await _outerKeySelector(item, _cancellationToken).ConfigureAwait(false); var inner = _lookup[outerKey].ToAsyncEnumerable(); Current = await _resultSelector(item, inner, _cancellationToken).ConfigureAwait(false); return true; } public TResult Current { get; private set; } public ValueTask DisposeAsync() => _outer.DisposeAsync(); } } #endif #endif } }