ToArrayAsync() 抛出“源 IQueryable 未实现 IAsyncEnumerable”

Question

我在 ASP.NET Core 上有一个 MVC 项目，我的问题与 IQueryable 和异步有关。我在IQueryable<T>写了如下搜索方法：

private IQueryable<InternalOrderInfo> WhereSearchTokens(IQueryable<InternalOrderInfo> query, SearchToken[] searchTokens)
{
    if (searchTokens.Length == 0)
    {
        return query;
    }
    var results = new List<InternalOrderInfo>();
    foreach (var searchToken in searchTokens)
    {
        //search logic, intermediate results are being added to `results` using `AddRange()`
    }

    return results.Count != 0 ? results.Distinct().AsQueryable() : query;
}

我在方法ExecuteAsync()中调用它：

public async Task<GetAllInternalOrderInfoResponse> ExecuteAsync(GetAllInternalOrderInfoRequest request)
{
    //rest of the code
    if (searchTokens != null && searchTokens.Any())
    {
        allInternalOrderInfo = WhereSearchTokens(allInternalOrderInfo, searchTokens);
    }
    var orders = await allInternalOrderInfo.Skip(offset).Take(limit).ToArrayAsync();
    //rest of the code
}

当我对此进行测试时，我在调用 ToArrayAsync() 的地方得到一个 InvalidOperationException

源 IQueryable 未实现 IAsyncEnumerable。只有实现 IAsyncEnumerable 的源才能用于实体框架异步操作。

我已将ToArrayAsync() 更改为ToListAsync()，但没有任何改变。我已经搜索了这个问题一段时间，但解决的问题主要与DbContext 和实体创建有关。该项目没有安装EntityFramework，由于应用程序架构，最好不要安装。希望有人对我的情况有什么想法。

Answer 1

如果您不打算更改设计 - 您有多种选择：

1) 将AsQueryable 更改为另一个返回IQueryable 的方法，该方法也实现了IDbAsyncEnumerable。例如你可以扩展EnumerableQuery（由AsQueryable返回）：

public class AsyncEnumerableQuery<T> : EnumerableQuery<T>, IDbAsyncEnumerable<T> {
    public AsyncEnumerableQuery(IEnumerable<T> enumerable) : base(enumerable) {
    }

    public AsyncEnumerableQuery(Expression expression) : base(expression) {
    }

    public IDbAsyncEnumerator<T> GetAsyncEnumerator() {
        return new InMemoryDbAsyncEnumerator<T>(((IEnumerable<T>) this).GetEnumerator());
    }

    IDbAsyncEnumerator IDbAsyncEnumerable.GetAsyncEnumerator() {
        return GetAsyncEnumerator();
    }

    private class InMemoryDbAsyncEnumerator<T> : IDbAsyncEnumerator<T> {
        private readonly IEnumerator<T> _enumerator;

        public InMemoryDbAsyncEnumerator(IEnumerator<T> enumerator) {
            _enumerator = enumerator;
        }

        public void Dispose() {
        }

        public Task<bool> MoveNextAsync(CancellationToken cancellationToken) {
            return Task.FromResult(_enumerator.MoveNext());
        }

        public T Current => _enumerator.Current;

        object IDbAsyncEnumerator.Current => Current;
    }
}

然后你改变

results.Distinct().AsQueryable()

到

new AsyncEnumerableQuery<InternalOrderInfo>(results.Distinct())

之后，ToArrayAsync 将不再抛出异常（显然您可以创建自己的扩展方法，如 AsQueryable）。

2) 更改ToArrayAsync 部分：

public static class EfExtensions {
    public static Task<TSource[]> ToArrayAsyncSafe<TSource>(this IQueryable<TSource> source) {
        if (source == null)
            throw new ArgumentNullException(nameof(source));
        if (!(source is IDbAsyncEnumerable<TSource>))
            return Task.FromResult(source.ToArray());
        return source.ToArrayAsync();
    }
}

并使用ToArrayAsyncSafe 而不是ToArrayAsync，如果IQueryable 不是IDbAsyncEnumerable，它将回退到同步枚举。在您的情况下，这只发生在查询实际上是内存列表而不是查询时，因此异步执行无论如何都没有意义。

Answer 2

我发现我必须做更多的工作才能让事情顺利进行：

namespace TestDoubles
{
    using Microsoft.EntityFrameworkCore.Query.Internal;
    using System;
    using System.Collections.Generic;
    using System.Linq;
    using System.Linq.Expressions;
    using System.Threading;
    using System.Threading.Tasks;

    public static class AsyncQueryable
    {
        /// <summary>
        /// Returns the input typed as IQueryable that can be queried asynchronously
        /// </summary>
        /// <typeparam name="TEntity">The item type</typeparam>
        /// <param name="source">The input</param>
        public static IQueryable<TEntity> AsAsyncQueryable<TEntity>(this IEnumerable<TEntity> source)
            => new AsyncQueryable<TEntity>(source ?? throw new ArgumentNullException(nameof(source)));
    }

    public class AsyncQueryable<TEntity> : EnumerableQuery<TEntity>, IAsyncEnumerable<TEntity>, IQueryable<TEntity>
    {
        public AsyncQueryable(IEnumerable<TEntity> enumerable) : base(enumerable) { }
        public AsyncQueryable(Expression expression) : base(expression) { }
        public IAsyncEnumerator<TEntity> GetEnumerator() => new AsyncEnumerator(this.AsEnumerable().GetEnumerator());
        public IAsyncEnumerator<TEntity> GetAsyncEnumerator(CancellationToken cancellationToken = default) => new AsyncEnumerator(this.AsEnumerable().GetEnumerator());
        IQueryProvider IQueryable.Provider => new AsyncQueryProvider(this);

        class AsyncEnumerator : IAsyncEnumerator<TEntity>
        {
            private readonly IEnumerator<TEntity> inner;
            public AsyncEnumerator(IEnumerator<TEntity> inner) => this.inner = inner;
            public void Dispose() => inner.Dispose();
            public TEntity Current => inner.Current;
            public ValueTask<bool> MoveNextAsync() => new ValueTask<bool>(inner.MoveNext());
#pragma warning disable CS1998 // Nothing to await
            public async ValueTask DisposeAsync() => inner.Dispose();
#pragma warning restore CS1998
        }

        class AsyncQueryProvider : IAsyncQueryProvider
        {
            private readonly IQueryProvider inner;
            internal AsyncQueryProvider(IQueryProvider inner) => this.inner = inner;
            public IQueryable CreateQuery(Expression expression) => new AsyncQueryable<TEntity>(expression);
            public IQueryable<TElement> CreateQuery<TElement>(Expression expression) => new AsyncQueryable<TElement>(expression);
            public object Execute(Expression expression) => inner.Execute(expression);
            public TResult Execute<TResult>(Expression expression) => inner.Execute<TResult>(expression);
            public IAsyncEnumerable<TResult> ExecuteAsync<TResult>(Expression expression) => new AsyncQueryable<TResult>(expression);
            TResult IAsyncQueryProvider.ExecuteAsync<TResult>(Expression expression, CancellationToken cancellationToken) => Execute<TResult>(expression);
        }
    }
}

这使我能够编写这样的测试：

    [TestCase("", 3, 5)]
    [TestCase("100", 2, 4)]
    public async Task GetOrderStatusCounts_ReturnsCorrectNumberOfRecords(string query, int expectedCount, int expectedStatusProductionCount)
    {
        // omitted CreateOrder helper function

        const int productionStatus = 6;
        const int firstOtherStatus = 5;
        const int otherOtherStatus = 7;

        var items = new[]
        {
            CreateOrder(1, "100000", firstOtherStatus, 1),
            CreateOrder(2, "100000", firstOtherStatus, 4),
            CreateOrder(3, "100000", productionStatus, 4),
            CreateOrder(4, "100001", productionStatus, 4),
            CreateOrder(5, "100100", productionStatus, 4),
            CreateOrder(6, "200000", otherOtherStatus, 4),
            CreateOrder(7, "200001", productionStatus, 4),
            CreateOrder(8, "200100", productionStatus, 4)
        }.AsAsyncQueryable(); // this is where the magic happens

        var mocker = new AutoMocker();

        // IRepository implementation is also generic and calls DBCntext
        // for easier testing
        mocker.GetMock<IRepository<Order>>() 
            .Setup(m => m.BaseQuery()
            .Returns(items); 
            // the base query is extended in the system under test.
            // that's the behavior I'm testing here

        var sut = mocker.CreateInstance<OrderService>();

        var counts = await sut.GetOrderStatusCountsAsync(4, query);

        counts.Should().HaveCount(expectedCount);
        counts[OrderStatus.Production].Should().Be(expectedStatusProductionCount);
    }

Answer 3

对于 EF Core：

public static class QueryableExtensions
{
    public static IQueryable<T> AsAsyncQueryable<T>(this IEnumerable<T> input)
    {
        return new NotInDbSet<T>( input );
    }

}

public class NotInDbSet< T > : IQueryable<T>, IAsyncEnumerable< T >, IEnumerable< T >, IEnumerable
{
    private readonly List< T > _innerCollection;
    public NotInDbSet( IEnumerable< T > innerCollection )
    {
        _innerCollection = innerCollection.ToList();
    }


    public IAsyncEnumerator< T > GetAsyncEnumerator( CancellationToken cancellationToken = new CancellationToken() )
    {
        return new AsyncEnumerator( GetEnumerator() );
    }

    public IEnumerator< T > GetEnumerator()
    {
        return _innerCollection.GetEnumerator();
    }

    IEnumerator IEnumerable.GetEnumerator()
    {
        return GetEnumerator();
    }

    public class AsyncEnumerator : IAsyncEnumerator< T >
    {
        private readonly IEnumerator< T > _enumerator;
        public AsyncEnumerator( IEnumerator< T > enumerator )
        {
            _enumerator = enumerator;
        }

        public ValueTask DisposeAsync()
        {
            return new ValueTask();
        }

        public ValueTask< bool > MoveNextAsync()
        {
            return new ValueTask< bool >( _enumerator.MoveNext() );
        }

        public T Current => _enumerator.Current;
    }

    public Type ElementType => typeof( T );
    public Expression Expression => Expression.Empty();
    public IQueryProvider Provider => new EnumerableQuery<T>( Expression );
}

Answer 4

对于 EFCore

晚了一点，但对于希望解决此类问题的其他人来说，一种可能的解决方案是更改代码以通过这种方式使用 Task.FromResult() 方法：

var result= await allInternalOrderInfo.Skip(offset).Take(limit);
var orders = await Task.FromResult(result.ToArray());

Answer 5

AsQueryable() 不会将result 列表转换为实体框架IQueryable。并且正如错误所述，与ToArrayAsync() 一起使用的IQueryable 应该实现IAsyncEnumerable，这不是AsQueryable 将返回的内容。

您可以阅读更多关于 AsQueryable 在枚举 here 上的用法。

Answer 6

我写了一个 ICollection 扩展 AsAsyncQueryable 我在我的测试中使用

using System;
using System.Collections;
using System.Collections.Generic;
using System.Linq;
using System.Linq.Expressions;
using System.Threading;
using System.Threading.Tasks;

namespace Whatevaaaaaaaa
{
    public static class ICollectionExtensions
    {
        public static IQueryable<T> AsAsyncQueryable<T>(this ICollection<T> source) =>
            new AsyncQueryable<T>(source.AsQueryable());
    }

    internal class AsyncQueryable<T> : IAsyncEnumerable<T>, IQueryable<T>
    {
        private IQueryable<T> Source;

        public AsyncQueryable(IQueryable<T> source)
        {
            Source = source;
        }

        public Type ElementType => typeof(T);

        public Expression Expression => Source.Expression;

        public IQueryProvider Provider => new AsyncQueryProvider<T>(Source.Provider);

        public IAsyncEnumerator<T> GetAsyncEnumerator(CancellationToken cancellationToken = default)
        {
            return new AsyncEnumeratorWrapper<T>(Source.GetEnumerator());
        }

        public IEnumerator<T> GetEnumerator() => Source.GetEnumerator();
        IEnumerator IEnumerable.GetEnumerator() => GetEnumerator();
    }

    internal class AsyncQueryProvider<T> : IQueryProvider
    {
        private readonly IQueryProvider Source;

        public AsyncQueryProvider(IQueryProvider source)
        {
            Source = source;
        }

        public IQueryable CreateQuery(Expression expression) =>
            Source.CreateQuery(expression);

        public IQueryable<TElement> CreateQuery<TElement>(Expression expression) =>
            new AsyncQueryable<TElement>(Source.CreateQuery<TElement>(expression));

        public object Execute(Expression expression) => Execute<T>(expression);

        public TResult Execute<TResult>(Expression expression) =>
            Source.Execute<TResult>(expression);
    }



    internal class AsyncEnumeratorWrapper<T> : IAsyncEnumerator<T>
    {
        private readonly IEnumerator<T> Source;

        public AsyncEnumeratorWrapper(IEnumerator<T> source)
        {
            Source = source;
        }

        public T Current => Source.Current;

        public ValueTask DisposeAsync()
        {
            return new ValueTask(Task.CompletedTask);
        }

        public ValueTask<bool> MoveNextAsync()
        {
            return new ValueTask<bool>(Source.MoveNext());
        }
    }
}

Answer 7

正如@Titian Cernicova-Dragomir 所述，异常意味着List<InternalOrderInfo> 没有实现IAsyncEnumerable

但这是一个逻辑/设计错误。如果您的方法与IQueryable 一起使用并返回IQueryable，则它应该与IQueryable 一起使用，而不是与IEnumarable 一起使用，假设集合在app 的内存中。您确实需要阅读更多关于IQueryable 和IEnumarable 之间的区别以及您应该从该方法返回的内容。一个好的开始是阅读答案here和here

因此，由于您已经在 WhereSearchTokens 方法中甚至之前从 db 中获取了结果，因此没有理由对 db 进行异步请求，这将由 ToArrayAsync 完成并返回 IQueryable。

您有两种选择：

1) 如果您的InternalOrderInfo 集合在WhereSearchTokens 之前从数据库提取到内存中，则使您的所有操作都处于同步模式，即调用ToArray 而不是ToArrayAsync，并返回IEnumerable 而不是Taks<IQueryable>来自WhereSearchTokens 和ExecuteAsync。

2) 如果您的 InternalOrderInfo 集合在 WhereSearchTokens 中获取，并且您想要对 db 执行异步请求，您需要仅在 //search logic, intermediate results are being added to results using AddRange() 的某处调用异步 EF API 并再次返回 Taks<IEnumerable> 而不是 @ 987654346@来自WhereSearchTokens

Answer 8

最好使用 IAsyncEnumerable 和 IQueriable 实现集合，而不是创建自己的 ToListAsync 扩展。

你不能在库中应用你的扩展

Fore EF Core 5 及更高版本检查此implementation 和tests。

短版：

public sealed class FixedQuery<T> : IAsyncEnumerable<T>, IQueryable<T>
{
    public static readonly IQueryable<T> Empty = Create(ArraySegment<T>.Empty);

    public static IQueryable<T> Create(params T[] items)
    {
        return Create((IEnumerable<T>)items);
    }
    public static IQueryable<T> Create(IEnumerable<T> items)
    {
        return new FixedQuery<T>(items ?? ArraySegment<T>.Empty).AsQueryable();
    }

    private readonly IQueryable<T> _items;

    private FixedQuery(IEnumerable<T> items)
    {
        _items = (items ?? throw new ArgumentNullException(nameof(items))).AsQueryable();
    }

    #pragma warning disable CS1998
    public async IAsyncEnumerator<T> GetAsyncEnumerator(CancellationToken cancellationToken = default)
    #pragma warning restore CS1998
    {
        foreach (var item in _items)
        {
            yield return item;
        }
    }

    public IEnumerator<T> GetEnumerator()
    {
        return _items.GetEnumerator();
    }
    IEnumerator IEnumerable.GetEnumerator()
    {
        return GetEnumerator();
    }

    public Type ElementType => _items.ElementType;
    public Expression Expression => _items.Expression;
    public IQueryProvider Provider => _items.Provider;
}