Как сделать асинхронный IEnumerable?

#c_sharp #tpl

Делаю поиск, по нескольким сразу местам.

На вернем уровне пока что-то типа:

  foreach (var item in Plugins.SelectMany(p => p.Search(query)))
    Items.Add(new ViewModel(item));


А в каждой реализации метода Search:

public override IEnumerable Search(string name)
{
  await some resource
  foreach (var element in networkResult)
    ...
    yield return result;
}


На деле, хочу параллельный доступ ко всем поискам, чтобы каждый элемент появлялся
в UI когда он готов, а не когда закончится всё целиком, как это сейчас работает. Что
именно тут в таски оборачивать - нету хороших идей. Снаружи вроде логичнее выглядит
Task, но реализовывать как правильно - не понимаю.
    

Ответы

Ответ 1

Смотрите, это не так сложно.

Вы устанавливаете Ix, nuget-пакеты System.Interactive и System.Interactive.Async.
У вас появляется интерфейс IAsyncEnumerable и вспомогательные классы.

Пользоваться можно вот так:

static class FileEx
{
    public static IAsyncEnumerable ReadLinesAsync(string path)
    {
        // IAsyncEnumerable обладает только одной функцией - создать энумератор
        return AsyncEnumerable.CreateEnumerable(() =>
        {
            var stream = File.OpenText(path);
            string current = null;
            // создаём энумератор при помощи готовой фабрики
            return AsyncEnumerable.CreateEnumerator(
                // у StreamReader.ReadLineAsync нет перегрузки с ct, пичалько
                moveNext: async ct => (current = await stream.ReadLineAsync()) != null,
                current: () => current,
                dispose: stream.Dispose);
        });
    }
}


Смотрите, что тут происходит. Для начала, одна и та же последовательность может пробегаться
разными кусками кода вперемежку, поэтому состояние текущего обхода мы держим в энумераторе.
В принципе, нам нужно было бы завести отдельный класс для энумератора, и держать в
нём свойства. Но мы пойдём более модным путём, и будем держать данные в замыкании.
Мы открываем StreamReader, заводим переменную для текущей строки.

Асинхронная функция MoveNext итератора получает следующую строку из StreamReader'а,
и проверяет результат на null (null означает конец файла). Функция Current просто выдаёт
текущую строку. А функция Dispose закрывает в конце поток.

Теперь с этим можно работать:

class Program
{
    static async Task Main(string[] args)
    {
        var lines = FileEx.ReadLinesAsync("text.txt");
        using (var en = lines.GetEnumerator())
        {
            while (await en.MoveNext())
                Console.WriteLine(en.Current);
        }
    }
}


Или просто

await FileEx.ReadLinesAsync("text.txt")
            .ForEachAsync(s => Console.WriteLine(s));




В следующей версии C# планируется поддержка асинхронных энумераторов прямо в языке.
С ней наш пример запишется так:

static class FileEx
{
    public static async IAsyncEnumerable ReadLinesAsync(string path)
    {
        using (var stream = File.OpenText(path))
        {
            string current;
            while ((current = await stream.ReadLineAsync()) != null)
                yield return current;
        };
    }
}

class Program
{
    static async Task Main(string[] args)
    {
        foreach await (var s in FileEx.ReadLinesAsync("text.txt"))
            Console.WriteLine(s);
    }
}




Смотрите, для конкретно вашего случая (вот такой код) вам стоит разобрать задачу
на составные части, поскольку асинхронных штук в ней много. 
Потом можно будет связать их вместо.

Начнём с получения страниц и их разбора. Список хостов получить просто, тут не нужна
асинхронность:

var hosts = ConfigStorage.Plugins
    .Where(p => p.GetParser().GetType() == typeof(Parser))
    .Select(p => p.GetSettings().MainUri);


Теперь, нам нужно по хосту получить список HtmlNodeCollection. Это «длинная» задача,
выносим её в таск:

async Task GetHostMangasAsync(string name, Uri host, CookieClient
client)
{
    var searchHost = new Uri(host, "search?q=" + WebUtility.UrlEncode(name));
    var page = await Task.Run(() => Page.GetPage(searchHost, client));
    if (!page.HasContent)
        return null;

    return await Task.Run(() =>
    {
        var document = new HtmlDocument();
        document.LoadHtml(page.Content);
        return document.DocumentNode.SelectNodes("//div[@class='tile col-sm-6']");
    });
}


Теперь, нам нужно из неасинхронной последовательности host'ов и Task'а, который получает
из каждого хоста HtmlNodeCollection, получить асинхронную последовательность. Такого
метода в Ix из коробки я не нашёл, но его легко сколотить самому. Сделаем его обобщённым,
вдруг ещё понадобится. Код практически ничем не отличается от примера с File.ReadLinesAsync.

static class AsyncEnumerableExtensions
{
    public static IAsyncEnumerable SelectAsync(
        this IEnumerable seq, Func> selector)
    {
        return AsyncEnumerable.CreateEnumerable(() =>
        {
            IEnumerator seqEnum = seq.GetEnumerator();
            R current = default;
            return AsyncEnumerable.CreateEnumerator(
                moveNext: async ct =>
                {
                    if (!seqEnum.MoveNext())
                        return false;
                    current = await selector(seqEnum.Current);
                    return true;
                },
                current: () => current,
                dispose: seqEnum.Dispose);
        });
    }
}


Вооружившись этим, мы можем написать такое:

IAsyncEnumerable GetSearchPages(string name)
{
    var hosts = ConfigStorage.Plugins
        .Where(p => p.GetParser().GetType() == typeof(Parser))
        .Select(p => p.GetSettings().MainUri);
    var client = new CookieClient();
    return hosts.SelectAsync(host => GetHostMangasAsync(name, host, client)))
                .Where(nc => nc != null);
}


Проверка на null нужна, потому что GetHostMangasAsync может вернуть null.

Отлично, переходим дальше. Итак, у нас снова есть неасинхронная коллекция HtmlNodeCollection,
из каждого элемента которой мы может вытащить при помощи асинхронной функции (т. к.
у нас есть обращение к сети) экземпляр IManga. Пишем код:

async Task GetMangaFromNode(Uri host, CookieClient client, HtmlNode manga)
{
    // Это переводчик, идем дальше.
    if (manga.SelectSingleNode(".//i[@class='fa fa-user text-info']") != null)
        return null;

    var image = manga.SelectSingleNode(".//div[@class='img']//a//img");
    var imageUri = image?.Attributes.Single(a => a.Name == "data-original").Value;

    var mangaNode = manga.SelectSingleNode(".//h3//a");
    var mangaUri = mangaNode.Attributes.Single(a => a.Name == "href").Value;
    var mangaName = mangaNode.Attributes.Single(a => a.Name == "title").Value;

    if (!Uri.TryCreate(mangaUri, UriKind.Relative, out Uri test))
        return null;

    var result = Mangas.Create(new Uri(host, mangaUri));
    result.Name = WebUtility.HtmlDecode(mangaName);
    if (imageUri != null)
        result.Cover = await client.DownloadDataAsync(imageUri);
    return result;
}


Нам нужно теперь их соединить. Это несложно. Единственная проблема — в GetMangaFromNode
тоже нужен CookieClient, а у нас от спрятан внутри GetSearchPages. Окей, будем передавать
его снаружи. Затем, у нас из GetSearchPages возвращается только HtmlNodeCollection,
а нужен ещё и host. Модифицируем GetSearchPages: будем возвращать пары из хоста и коллекции
HtmlNode, и принимать на вход CookieClient:

IAsyncEnumerable<(Uri host, HtmlNodeCollection nodes)> GetSearchPages(
        string name, CookieClient client)
{
    var hosts = ConfigStorage.Plugins
        .Where(p => p.GetParser().GetType() == typeof(Parser))
        .Select(p => p.GetSettings().MainUri);
    return
        hosts.SelectAsync(
                async host => (host, nodes: await GetHostMangasAsync(name, host,
client)))
             .Where(pair => pair.nodes != null);
}


Ну и комбинируем. У нас каждая синхронная коллекция HtmlNode при помощи асинхронной
функции даёт коллекцию экземпляров IManga. Это делается снова при помощи нашего SelectAsync:

IAsyncEnumerable GetFromHostAndNodes(
        Uri host, HtmlNodeCollection nodes, CookieClient client) =>
    nodes.SelectAsync(node => GetMangaFromNode(host, client, node));


Теперь можно складывать паззл:

public IAsyncEnumerable Search(string name)
{
    var client = new CookieClient();
    return
        GetSearchPages(name, client)
            .SelectMany(pair => GetFromHostAndNodes(pair.host, pair.nodes, client))
            .Where(m => m != null);
}


Всё!