Пользовательские асинхронные перечисляемые объекты с IAsyncEnumerable в C#

// Старый подход - ждём весь результат
public async Task<IEnumerable<int>> GetNumbersAsync()
{
    var result = new List<int>();
    for (int i = 1; i <= 5; i++)
    {
        await Task.Delay(1000); // Имитация асинхронной работы
        result.Add(i);
    }
    return result;
}
 
// Использование
var numbers = await GetNumbersAsync(); // Ждем 5 секунд!
foreach (var number in numbers) // Только потом начинаем обработку
{
    Console.WriteLine(number);
}

// Новый подход с IAsyncEnumerable
public async IAsyncEnumerable<int> GetNumbersAsync()
{
    for (int i = 1; i <= 5; i++)
    {
        await Task.Delay(1000); // Имитация асинхронной работы
        yield return i; // Возвращаем элемент, как только он готов
    }
}
 
// Использование
await foreach (var number in GetNumbersAsync())
{
    Console.WriteLine(number); // Получаем число каждую секунду
}

// Неоправданное использование
public async IAsyncEnumerable<int> GetSmallArrayAsync()
{
    var array = new[] { 1, 2, 3, 4, 5 };
    foreach (var item in array)
    {
        // Нет реальной асинхронной работы
        yield return item;
    }
}
 
// Оправданное использование
public async IAsyncEnumerable<string> ReadLargeLogFileAsync(string path)
{
    using var reader = new StreamReader(path);
    while (!reader.EndOfStream)
    {
        // Реальная асинхронная операция чтения файла
        yield return await reader.ReadLineAsync();
    }
}

[HttpGet]
public async IAsyncEnumerable<WeatherForecast> GetWeatherForecastStreamAsync()
{
    for (int i = 1; i <= 5; i++)
    {
        await Task.Delay(1000);
        yield return new WeatherForecast { Date = DateTime.Now.AddDays(i) };
    }
}

// Вот как это выглядит в реальности
public interface IAsyncEnumerable<out T>
{
    IAsyncEnumerator<T> GetAsyncEnumerator(CancellationToken cancellationToken = default);
}
 
public interface IAsyncEnumerator<out T> : IAsyncDisposable
{
    T Current { get; }
    ValueTask<bool> MoveNextAsync();
}

public async IAsyncEnumerable<string> GetDataFromMultipleSourcesAsync()
{
var result1 = await FetchFromApiAsync();
yield return result1;
    
var result2 = await FetchFromDatabaseAsync();
yield return result2;
    
// Можно даже комбинировать с параллельным выполнением
var tasks = new[] 
{
    FetchFromCacheAsync(),
    FetchFromFileAsync()
};
foreach (var result in await Task.WhenAll(tasks))
{
    yield return result;
}
}

public async IAsyncEnumerable<int> GenerateNumbersAsync(
    [EnumeratorCancellation] CancellationToken token = default)
{
for (int i = 0; token.IsCancellationRequested == false; i++)
{
    await Task.Delay(100, token);
    yield return i;
}
}
 
// Использование с отменой
async Task ConsumeWithCancellationAsync()
{
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(2));
try
{
    await foreach (var number in GenerateNumbersAsync(cts.Token))
    {
        Console.WriteLine(number);
    }
}
catch (OperationCanceledException)
{
    Console.WriteLine("Операция была отменена");
}
}

using System.Reactive.Linq;
 
IAsyncEnumerable<int> asyncNumbers = GenerateNumbersAsync();
IObservable<int> observableNumbers = asyncNumbers.ToObservable();
 
// И обратно
IAsyncEnumerable<int> backToAsyncEnum = observableNumbers.ToAsyncEnumerable();

public record DataPoint(DateTime Timestamp, double Value);
 
public async IAsyncEnumerable<DataPoint> GetDataPointsAsync()
{
while (true)
{
    var value = await SensorReadAsync();
    yield return new DataPoint(DateTime.Now, value);
    await Task.Delay(1000);
}
}
 
await foreach (var point in GetDataPointsAsync())
{
if (point is DataPoint { Value: > 100 })
{
    Console.WriteLine($"High value detected: {point.Value}");
}
}

public interface IAsyncEnumerable<out T>
{
    IAsyncEnumerator<T> GetAsyncEnumerator(CancellationToken cancellationToken = default);
}
 
public interface IAsyncEnumerator<out T> : IAsyncDisposable
{
    T Current { get; }
    ValueTask<bool> MoveNextAsync();
}

IAsyncEnumerator<T> enumerator = sequence.GetAsyncEnumerator();
try
{
    while (await enumerator.MoveNextAsync())
    {
        T item = enumerator.Current;
        // Ваш код обработки элемента
    }
}
finally
{
    if (enumerator != null)
    {
        await enumerator.DisposeAsync();
    }
}

public async IAsyncEnumerable<int> GenerateNumbersAsync()
{
    for (int i = 0; i < 10; i++)
    {
        await Task.Delay(100); // Имитируем асинхронную работу
        yield return i;
    }
}

public IAsyncEnumerable<int> GenerateNumbersAsync()
{
    return new GenerateNumbersAsyncEnumerable();
}
 
private sealed class GenerateNumbersAsyncEnumerable : IAsyncEnumerable<int>
{
    public IAsyncEnumerator<int> GetAsyncEnumerator(CancellationToken token = default)
    {
        return new GenerateNumbersAsyncEnumerator();
    }
    
    private sealed class GenerateNumbersAsyncEnumerator : IAsyncEnumerator<int>, IAsyncStateMachine
    {
        // Состояние итератора
        private int _state;
        private int _i;
        private TaskAwaiter _awaiter;
        
        public int Current { get; private set; }
        
        public ValueTask<bool> MoveNextAsync()
        {
            // Сложная машина состояний...
            // Двигаемся по методу, останавливаясь на await
            // и yield return
        }
        
        public ValueTask DisposeAsync()
        {
            return default;
        }
    }
}

// Подход с Task<IEnumerable<T>>
public async Task<IEnumerable<T>> GetDataAsync() { ... }
 
// Подход с IObservable<T> из Rx.NET
public IObservable<T> GetDataStream() { ... }
 
// Подход с Channels из System.Threading.Channels
ChannelReader<T> reader = channel.Reader;

public interface IAsyncDisposable
{
ValueTask DisposeAsync();
}

await using var connection = await GetConnectionAsync();
// Используем соединение
// По выходу из блока будет вызван DisposeAsync()

public async IAsyncEnumerable<string> ReadLinesAsync(string filePath)
{
using var reader = new StreamReader(filePath);
while (!reader.EndOfStream)
{
yield return await reader.ReadLineAsync();
}
}

[AsyncIteratorStateMachine(typeof(ReadLinesAsyncStateMachine))]
public IAsyncEnumerable<string> ReadLinesAsync(string filePath)
{
ReadLinesAsyncStateMachine stateMachine = new ReadLinesAsyncStateMachine();
stateMachine._filePath = filePath;
stateMachine._builder = AsyncIteratorMethodBuilder<string>.Create();
stateMachine._state = -1;
var iterator = new AsyncIterator<string>(stateMachine.MoveNext, stateMachine.DisposeAsync);
return iterator;
}
 
private struct ReadLinesAsyncStateMachine : IAsyncStateMachine
{
public string _filePath;
public AsyncIteratorMethodBuilder<string> _builder;
public int _state;
private StreamReader _reader;
private ConfiguredTaskAwaitable<string>.ConfiguredTaskAwaiter _awaiter;
 
public void MoveNext()
{
    string yielded = null;
    try
    {
        bool finished = false;
        switch (_state)
        {
            case -1: // Инициализация
                _reader = new StreamReader(_filePath);
                _state = 0;
                goto case 0;
            
            case 0: // Проверка EndOfStream
                if (_reader.EndOfStream)
                {
                    finished = true;
                    break;
                }
                
                // Ожидаем чтения строки
                _awaiter = _reader.ReadLineAsync().ConfigureAwait(false).GetAwaiter();
                if (!_awaiter.IsCompleted)
                {
                    _state = 1;
                    _builder.AwaitOnCompleted(ref _awaiter, ref this);
                    return;
                }
                goto case 1;
                
            case 1: // Продолжение после await ReadLineAsync
                yielded = _awaiter.GetResult();
                _state = 0;
                _builder.Yield(yielded);
                return;
        }
        
        if (_reader != null)
        {
            _reader.Dispose();
            _reader = null;
        }
        
        _builder.Complete();
    }
    catch (Exception ex)
    {
        _state = -2;
        if (_reader != null)
        {
            _reader.Dispose();
            _reader = null;
        }
        _builder.SetException(ex);
    }
}
 
public ValueTask DisposeAsync()
{
    if (_reader != null)
    {
        _reader.Dispose();
        _reader = null;
    }
    return default;
}
}

using System.Linq;
using System.Linq.Async; // Требуется NuGet-пакет System.Linq.Async
 
var numbers = asyncSequence
.Where(x => x > 0)
.Select(x => x * 2)
.Skip(10)
.Take(5);
 
await foreach (var number in numbers)
{
Console.WriteLine(number);
}

// Порядок элементов может меняться в зависимости от скорости асинхронных операций
var results = asyncSource.SelectAwait(async item => await ProcessAsync(item));
 
// Гарантируем порядок элементов
var orderedResults = asyncSource.SelectAwaitWithCancellation(
(item, token) => ProcessAsync(item, token)
).WithCompletionOrder(OperationCompletionOrder.SequentialExecution);

// Подсчет элементов, удовлетворяющих условию
int count = await asyncSequence.CountAsync(x => x % 2 == 0);
 
// Вычисление среднего значения
double average = await asyncSequence.AverageAsync();
 
// Преобразование в список
List<int> list = await asyncSequence.ToListAsync();

public async Task<List<T>> GetFilteredItemsAsync<T>(
    IAsyncEnumerable<T> source, Func<T, bool> predicate)
{
return await source
.Where(predicate)
.ToListAsync();
}

await asyncSequence.WhenAll(async item =>
{
// Параллельная обработка каждого элемента
await ProcessItemAsync(item);
});

// Конвертация Task<IEnumerable<T>> в IAsyncEnumerable<T>
Task<IEnumerable<int>> task = GetNumbersAsync();
IAsyncEnumerable<int> asyncNumbers = task.ToAsyncEnumerable();
 
// Конвертация IObservable<T> в IAsyncEnumerable<T>
IObservable<int> observable = GetObservableNumbers();
IAsyncEnumerable<int> fromObservable = observable.ToAsyncEnumerable();

using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5));
await foreach (var item in asyncSequence.WithCancellation(cts.Token))
{
// Обработка прервется через 5 секунд
}

// Избегаем возврата в контекст UI-потока
await foreach (var item in asyncSequence.ConfigureAwait(false))
{
// Обработка будет происходить в потоке из пула
}

   IEnumerator<T> enumerator = collection.GetEnumerator();

   IAsyncEnumerator<T> asyncEnumerator = asyncCollection.GetAsyncEnumerator(token);

   while (enumerator.MoveNext())
   {
      var item = enumerator.Current;
      // Обработка
   }

   while (await asyncEnumerator.MoveNextAsync())
   {
      var item = asyncEnumerator.Current;
      // Обработка
   }

   enumerator.Dispose();

   await asyncEnumerator.DisposeAsync();

// Синхронная итерация
foreach (var item in collection)
{
    ProcessItem(item);
}
 
// Асинхронная итерация
await foreach (var item in asyncCollection)
{
    await ProcessItemAsync(item);
}

// Более эффективный вариант для небольших синхронных коллекций
public static IAsyncEnumerable<T> ToAsyncEnumerable<T>(this IEnumerable<T> source)
{
var list = source.ToList(); // Материализуем один раз
return YieldElements();
 
async IAsyncEnumerable<T> YieldElements()
{
    foreach (var item in list)
    {
        yield return item;
    }
}
}

public ValueTask<bool> MoveNextAsync()
{
// Если у нас есть кэшированные данные, мы можем вернуть результат синхронно
if (_cache.TryGetNext(out var next))
{
    Current = next;
    return new ValueTask<bool>(true); // Нет выделения памяти
}
 
// В противном случае выполняем асинхронную операцию
return DoAsyncMoveAsync();
}
 
private async ValueTask<bool> DoAsyncMoveAsync()
{
var result = await _source.FetchNextAsync();
if (result != null)
{
    Current = result;
    _cache.Store(result);
    return true;
}
return false;
}

public async IAsyncEnumerable<string> ReadLinesFromUrlAsync(
string url, [EnumeratorCancellation] CancellationToken token = default)
{
using var client = new HttpClient();
using var response = await client.GetAsync(url, HttpCompletionOption.ResponseHeadersRead, token);
using var stream = await response.Content.ReadAsStreamAsync();
using var reader = new StreamReader(stream);
 
while (!reader.EndOfStream)
{
token.ThrowIfCancellationRequested();
yield return await reader.ReadLineAsync();
}
}

public class AsyncNumberRange : IAsyncEnumerable<int>
{
private readonly int _start;
private readonly int _count;
private readonly int _delay;
 
public AsyncNumberRange(int start, int count, int delayMs)
{
    _start = start;
    _count = count;
    _delay = delayMs;
}
 
public IAsyncEnumerator<int> GetAsyncEnumerator(CancellationToken token = default)
{
    return new Enumerator(_start, _count, _delay, token);
}
 
private class Enumerator : IAsyncEnumerator<int>
{
    private readonly int _start;
    private readonly int _count;
    private readonly int _delay;
    private readonly CancellationToken _token;
    private int _current;
    private int _position;
    
    public Enumerator(int start, int count, int delay, CancellationToken token)
    {
        _start = start;
        _count = count;
        _delay = delay;
        _token = token;
        _position = -1;
    }
    
    public int Current => _current;
    
    public async ValueTask<bool> MoveNextAsync()
    {
        _token.ThrowIfCancellationRequested();
        
        if (_position + 1 < _count)
        {
            _position++;
            _current = _start + _position;
            
            if (_delay > 0)
            {
                await Task.Delay(_delay, _token);
            }
            
            return true;
        }
        
        return false;
    }
    
    public ValueTask DisposeAsync()
    {
        // Никаких ресурсов для освобождения
        return ValueTask.CompletedTask;
    }
}
}

public static IAsyncEnumerable<T> GenerateAsync<T>(
Func<int, CancellationToken, ValueTask<T>> generator,
int count)
{
return AsyncEnumerable.CreateEnumerable<T>(token =>
{
    int index = 0;
    
    return AsyncEnumerable.CreateEnumerator(
        moveNext: async () =>
        {
            if (index < count)
            {
                Current = await generator(index++, token);
                return true;
            }
            return false;
        },
        current: default,
        dispose: () => ValueTask.CompletedTask);
});
}
 
// Использование
var numbers = GenerateAsync(
(index, token) => new ValueTask<int>(index * index),
10);

// Внутренний механизм AsyncIteratorMethodBuilder (упрощённо)
public struct AsyncIteratorMethodBuilder<T>
{
    private ManualResetValueTaskSourceCore<bool> _valueTaskSource;
    private Exception _exception;
    private T _current;
    
    public T Current => _current;
 
    public static AsyncIteratorMethodBuilder<T> Create() => new();
    
    public void Yield(T value)
    {
        _current = value;
        _valueTaskSource.SetResult(true);
    }
    
    public void Complete()
    {
        _valueTaskSource.SetResult(false);
    }
    
    public void SetException(Exception exception)
    {
        _exception = exception;
        _valueTaskSource.SetException(exception);
    }
}

// Не забудьте подключить пространство имен
using System.Linq.Async;
 
// Теперь можно использовать знакомые операции
var filteredData = asyncSequence
    .Where(x => x.IsValid)
    .Select(x => x.Value)
    .Take(10);
 
await foreach (var item in filteredData)
{
    ProcessItem(item);
}

// Последовательное выполнение (затратно по времени)
var sequentialResult = await source.SelectAwait(async x => await SlowProcessAsync(x));
 
// Параллельное выполнение (эффективнее)
var parallelResult = await source.SelectAwaitWithCancellation(
    (item, token) => FastProcessAsync(item, token)
);

// Неэффективно: трансформируем все элементы
var inefficient = source.Select(x => ExpensiveTransform(x)).Where(x => x > 0);
 
// Эффективно: трансформируем только подходящие
var efficient = source.Where(x => ShouldProcess(x)).Select(x => ExpensiveTransform(x));

// Не пытайтесь сделать это
var list = new List<int>();
await foreach (var item in asyncSequence)
{
    list.Add(item);
}
 
// Используйте вместо этого
var list = await asyncSequence.ToListAsync();

// С IEnumerable обычно можно делать так
var collection = GetNumbers();
foreach (var n in collection) { /* первый обход */ }
foreach (var n in collection) { /* второй обход */ }
 
// С IAsyncEnumerable будьте осторожны
var asyncCollection = GetNumbersAsync();
await foreach (var n in asyncCollection) { /* первый обход */ }
// Второй обход может завершиться ошибкой или пустой коллекцией

try {
    await foreach (var item in GetUnstableCollection())
    {
        Process(item);
    }
} catch (Exception ex) {
    // После этого перечислитель непригоден
}

public async IAsyncEnumerable<double> GenerateSensorReadingsAsync(
    [EnumeratorCancellation] CancellationToken token = default)
{
    var random = new Random();
    while (!token.IsCancellationRequested)
    {
        // Имитируем получение данных с датчика
        await Task.Delay(500, token);
        yield return random.NextDouble() * 100;
    }
}

public class DatabaseResultStream<T> : IAsyncEnumerable<T>
{
    private readonly string _connectionString;
    private readonly string _query;
    private readonly Func<DbDataReader, T> _mapper;
 
    public DatabaseResultStream(string connectionString, string query, 
                               Func<DbDataReader, T> mapper)
    {
        _connectionString = connectionString;
        _query = query;
        _mapper = mapper;
    }
 
    public IAsyncEnumerator<T> GetAsyncEnumerator(
        CancellationToken token = default)
    {
        return new DatabaseResultEnumerator<T>(
            _connectionString, _query, _mapper, token);
    }
 
    private class DatabaseResultEnumerator<T> : IAsyncEnumerator<T>
    {
        // Реализация перечислителя...
        // (Подробности опущены для краткости)
    }
}

public static IAsyncEnumerable<T> CreateFrom<T>(
    Func<CancellationToken, IAsyncEnumerator<T>> factory)
{
    return new AnonymousAsyncEnumerable<T>(factory);
}
 
// Использование
var range = CreateFrom<int>(token => 
    new RangeAsyncEnumerator(1, 10, token));

public async IAsyncEnumerable<Order> GetAllOrdersAsync(
    int customerId, [EnumeratorCancellation] CancellationToken token = default)
{
    int page = 1;
    bool hasMore = true;
    
    while (hasMore && !token.IsCancellationRequested)
    {
        var batch = await _apiClient.GetOrdersAsync(customerId, page++, token);
        foreach (var order in batch.Orders)
        {
            yield return order;
        }
        
        hasMore = batch.HasMorePages;
    }
}

public class BufferedAsyncCollection<T> : IAsyncEnumerable<T>
{
    private readonly IAsyncEnumerable<T> _source;
    private readonly int _bufferSize;
    private readonly Channel<T> _channel;
 
    public BufferedAsyncCollection(IAsyncEnumerable<T> source, int bufferSize = 100)
    {
        _source = source;
        _bufferSize = bufferSize;
        _channel = Channel.CreateBounded<T>(new BoundedChannelOptions(bufferSize)
        {
            FullMode = BoundedChannelFullMode.Wait
        });
        
        // Запускаем наполнение буфера в фоне
        _ = FillBufferAsync();
    }
 
    private async Task FillBufferAsync()
    {
        try
        {
            await foreach (var item in _source)
            {
                await _channel.Writer.WriteAsync(item);
            }
            // Сигнализируем, что больше элементов не будет
            _channel.Writer.Complete();
        }
        catch (Exception ex)
        {
            // Пробрасываем ошибку потребителю
            _channel.Writer.Complete(ex);
        }
    }
 
    public async IAsyncEnumerator<T> GetAsyncEnumerator(CancellationToken token = default)
    {
        try
        {
            while (await _channel.Reader.WaitToReadAsync(token))
            {
                while (_channel.Reader.TryRead(out var item))
                {
                    yield return item;
                }
            }
        }
        finally
        {
            // Дополнительная логика очистки, если необходимо
        }
    }
}

public static class AsyncEnumerableExtensions
{
    public static IAsyncEnumerable<T> ThrottleTime<T>(
        this IAsyncEnumerable<T> source, 
        TimeSpan throttleInterval)
    {
        return new ThrottledAsyncEnumerable<T>(source, throttleInterval);
    }
}
 
// Использование
var throttledData = dataStream
    .ThrottleTime(TimeSpan.FromMilliseconds(500))
    .Where(x => x.IsValid);

public static IAsyncEnumerable<T> Cache<T>(this IAsyncEnumerable<T> source)
{
    var cachedList = new List<T>();
    var semaphore = new SemaphoreSlim(1, 1);
    var completed = false;
    var enumerator = source.GetAsyncEnumerator();
 
    return ExecuteAsync();
 
    async IAsyncEnumerable<T> ExecuteAsync([EnumeratorCancellation] CancellationToken token = default)
    {
        var index = 0;
        
        while (true)
        {
            T item;
            
            // Проверяем, есть ли элемент в кэше
            await semaphore.WaitAsync(token);
            try
            {
                if (index < cachedList.Count)
                {
                    item = cachedList[index++];
                    yield return item;
                    continue;
                }
                
                if (completed)
                {
                    break;
                }
            }
            finally
            {
                semaphore.Release();
            }
            
            // Получаем новый элемент из источника
            bool hasNext;
            await semaphore.WaitAsync(token);
            try
            {
                hasNext = await enumerator.MoveNextAsync();
                if (hasNext)
                {
                    item = enumerator.Current;
                    cachedList.Add(item);
                    index++;
                }
                else
                {
                    completed = true;
                    break;
                }
            }
            finally
            {
                semaphore.Release();
            }
            
            yield return item;
        }
    }
}

public static class AsyncEnumerableFactory
{
    public static IAsyncEnumerable<T> Create<T>(
        Func<CancellationToken, IAsyncEnumerator<T>> enumeratorFactory)
    {
        return new AnonymousAsyncEnumerable<T>(enumeratorFactory);
    }
    
    private class AnonymousAsyncEnumerable<T> : IAsyncEnumerable<T>
    {
        private readonly Func<CancellationToken, IAsyncEnumerator<T>> _factory;
        
        public AnonymousAsyncEnumerable(Func<CancellationToken, IAsyncEnumerator<T>> factory)
        {
            _factory = factory;
        }
        
        public IAsyncEnumerator<T> GetAsyncEnumerator(CancellationToken token = default)
        {
            return _factory(token);
        }
    }
}

public class LazyAsyncEnumerable<T> : IAsyncEnumerable<T>
{
    private readonly Func<Task<IAsyncEnumerable<T>>> _factory;
    
    public LazyAsyncEnumerable(Func<Task<IAsyncEnumerable<T>>> factory)
    {
        _factory = factory;
    }
    
    public async IAsyncEnumerator<T> GetAsyncEnumerator(CancellationToken token = default)
    {
        var source = await _factory();
        await foreach (var item in source.WithCancellation(token))
        {
            yield return item;
        }
    }
}

public class AsyncPaginator<T> : IAsyncEnumerable<T>
{
    private readonly Func<int, CancellationToken, Task<(IEnumerable<T> items, bool hasMore)>> _pageProvider;
    
    public AsyncPaginator(Func<int, CancellationToken, Task<(IEnumerable<T>, bool)>> pageProvider)
    {
        _pageProvider = pageProvider;
    }
    
    public async IAsyncEnumerator<T> GetAsyncEnumerator(CancellationToken token = default)
    {
        int pageNumber = 1;
        bool hasMore = true;
        
        while (hasMore && !token.IsCancellationRequested)
        {
            var (items, more) = await _pageProvider(pageNumber++, token);
            foreach (var item in items)
            {
                yield return item;
            }
            hasMore = more;
        }
    }
}

public class LoggingAsyncEnumerable<T> : IAsyncEnumerable<T>
{
    private readonly IAsyncEnumerable<T> _inner;
    private readonly Action<T> _logger;
    
    public LoggingAsyncEnumerable(IAsyncEnumerable<T> inner, Action<T> logger)
    {
        _inner = inner;
        _logger = logger;
    }
    
    public async IAsyncEnumerator<T> GetAsyncEnumerator(CancellationToken token = default)
    {
        await foreach (var item in _inner.WithCancellation(token))
        {
            _logger(item);
            yield return item;
        }
    }
}

public async IAsyncEnumerable<string> ReadLongLogFileAsync(
    string filePath,
    [EnumeratorCancellation] CancellationToken token = default)
{
    using var reader = new StreamReader(filePath);
    while (!reader.EndOfStream)
    {
        // Проверяем отмену перед каждой потенциально долгой операцией
        token.ThrowIfCancellationRequested();
        
        string line = await reader.ReadLineAsync();
        if (ShouldProcessLine(line))
        {
            // И здесь тоже проверяем, не отменили ли операцию пока мы обрабатывали строку
            token.ThrowIfCancellationRequested();
            yield return line;
        }
    }
}

using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5));
// Этот токен пройдёт в параметр с атрибутом [EnumeratorCancellation]
await foreach (var line in ReadLongLogFileAsync("huge.log").WithCancellation(cts.Token))
{
    Console.WriteLine(line);
}

public IAsyncEnumerator<T> GetAsyncEnumerator(CancellationToken token = default)
{
    return new MyCustomEnumerator<T>(_source, token);
}

public async IAsyncEnumerable<DataItem> FetchFromMultipleSourcesAsync(
    [EnumeratorCancellation] CancellationToken externalToken = default)
{
    using var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(externalToken);
    
    // Устанавливаем таймаут в 30 секунд для каждой операции
    using var timeoutCts = new CancellationTokenSource(TimeSpan.FromSeconds(30));
    using var combinedCts = CancellationTokenSource.CreateLinkedTokenSource(
        externalToken, timeoutCts.Token);
    
    var token = combinedCts.Token;
    
    try
    {
        // Теперь операция будет отменена либо по внешнему токену, либо по таймауту
        var result = await FetchDataAsync(token);
        yield return result;
    }
    catch (OperationCanceledException) when (timeoutCts.IsCancellationRequested)
    {
        yield return new DataItem { IsTimeout = true };
    }
}

public async Task ProcessGigabytesOfDataAsync(string hugeDatabaseTableName)
{
    await foreach (var chunk in GetDataChunksAsync(hugeDatabaseTableName, chunkSize: 1000))
    {
        var transformed = chunk
            .AsParallel()
            .Select(Transform)
            .ToList();
            
        await SaveResultsAsync(transformed);
        
        // Освобождаем память сразу после использования чанка
        GC.Collect();
    }
}

public async IAsyncEnumerable<ReadOnlyMemory<byte>> ReadFileChunksAsync(
    string filePath, int chunkSize = 81920, 
    [EnumeratorCancellation] CancellationToken token = default)
{
    using var fileStream = new FileStream(
        filePath, FileMode.Open, FileAccess.Read, FileShare.Read, 
        bufferSize: 4096, useAsync: true);
    
    var buffer = new byte[chunkSize];
    int bytesRead;
    
    while ((bytesRead = await fileStream.ReadAsync(buffer, 0, buffer.Length, token)) > 0)
    {
        yield return new ReadOnlyMemory<byte>(buffer, 0, bytesRead);
    }
}

public static async IAsyncEnumerable<IEnumerable<T>> BatchAsync<T>(
    this IAsyncEnumerable<T> source, int batchSize,
    [EnumeratorCancellation] CancellationToken token = default)
{
    List<T> batch = new(batchSize);
    
    await foreach (var item in source.WithCancellation(token))
    {
        batch.Add(item);
        
        if (batch.Count >= batchSize)
        {
            yield return batch;
            batch = new List<T>(batchSize);
        }
    }
    
    if (batch.Count > 0)
    {
        yield return batch;
    }
}

public static async IAsyncEnumerable<T> MergeAsync<T>(
    this IEnumerable<IAsyncEnumerable<T>> sources,
    [EnumeratorCancellation] CancellationToken token = default)
{
    var tasks = sources.Select(source => ProcessSourceAsync(source, token)).ToList();
    var channels = new Channel<T>[tasks.Count];
    
    // Настройка каналов и обработчиков
    // ...
 
    while (tasks.Any())
    {
        var completedTask = await Task.WhenAny(tasks);
        int index = tasks.IndexOf(completedTask);
        
        // Обработка результатов из завершившейся задачи
        // ...
        
        yield return await completedTask;
    }
}

public static IAsyncEnumerable<T> WithRetry<T>(
    this IAsyncEnumerable<T> source, int maxRetries = 3)
{
    return ExecuteWithRetry();
    
    async IAsyncEnumerable<T> ExecuteWithRetry(
        [EnumeratorCancellation] CancellationToken token = default)
    {
        var enumerator = source.GetAsyncEnumerator(token);
        var currentTry = 0;
        
        while (true)
        {
            try
            {
                if (!await enumerator.MoveNextAsync())
                    break;
                    
                yield return enumerator.Current;
                currentTry = 0; // Сбрасываем счетчик после успешной операции
            }
            catch (Exception) when (++currentTry <= maxRetries)
            {
                // Экспоненциальная задержка между попытками
                await Task.Delay(TimeSpan.FromSeconds(Math.Pow(2, currentTry)), token);
            }
        }
    }
}

public async IAsyncEnumerable<Product> GetAllProductsAsync(
    [EnumeratorCancellation] CancellationToken token = default)
{
    int page = 1;
    const int pageSize = 100;
    bool hasMore = true;
    
    while (hasMore && !token.IsCancellationRequested)
    {
        var response = await _httpClient.GetFromJsonAsync<PagedResponse<Product>>(
            $"api/products?page={page}&size={pageSize}", token);
            
        foreach (var product in response.Items)
        {
            yield return product;
        }
        
        hasMore = response.HasNextPage;
        page++;
    }
}

public async IAsyncEnumerable<Customer> GetLargeCustomerListAsync(
    [EnumeratorCancellation] CancellationToken token = default)
{
    await using var context = new MyDbContext();
    
    // AsAsyncEnumerable автоматически обработает пагинацию на уровне БД
    await foreach (var customer in context.Customers
        .Where(c => c.IsActive)
        .AsAsyncEnumerable()
        .WithCancellation(token))
    {
        yield return customer;
    }
}

public async IAsyncEnumerable<LogEntry> ParseLogFileAsync(
    string filePath, 
    [EnumeratorCancellation] CancellationToken token = default)
{
    using var fileStream = new FileStream(
        filePath, FileMode.Open, FileAccess.Read, FileShare.Read, 
        bufferSize: 4096, useAsync: true);
    using var reader = new StreamReader(fileStream);
    
    while (!reader.EndOfStream)
    {
        token.ThrowIfCancellationRequested();
        var line = await reader.ReadLineAsync();
        
        if (LogEntry.TryParse(line, out var entry))
        {
            yield return entry;
        }
    }
}

// Неправильно: утечка ресурсов
IAsyncEnumerator<Data> enumerator = dataSource.GetAsyncEnumerator();
while (await enumerator.MoveNextAsync())
{
    // Работаем с данными
}
// Забыли вызвать await enumerator.DisposeAsync()!
 
// Правильно: использование await using
await using var enumerator = dataSource.GetAsyncEnumerator();
while (await enumerator.MoveNextAsync())
{
    // Работаем с данными
}

// Избыточно: асинхронность не используется
public async IAsyncEnumerable<int> GetNumbersAsync()
{
    for (int i = 0; i < 1000; i++)
    {
        // Нет реальных асинхронных операций
        yield return i;
    }
}
 
// Лучше использовать обычный IEnumerable
public IEnumerable<int> GetNumbers()
{
    for (int i = 0; i < 1000; i++)
    {
        yield return i;
    }
}

try
{
    await foreach (var item in GetDataAsync())
    {
        try
        {
            ProcessItem(item); // Может выбросить исключение
        }
        catch (ItemProcessingException ex)
        {
            // Обрабатываем исключение конкретного элемента
            LogError(ex);
            // Продолжаем обработку следующих элементов
        }
    }
}
catch (Exception ex)
{
    // Обрабатываем общие исключения самого перечисления
    LogFatalError(ex);
}

await foreach (var item in dataStream.ConfigureAwait(false))
{
    // Выполняется в потоке из пула, без возврата в UI-поток
    Process(item);
}

// Опасно: параллельное перечисление
var sequence = GetDataAsync();
 
// Этот код может привести к состоянию гонки
Task.WhenAll(
    Task.Run(async () => await foreach (var item in sequence) { /* ... */ }),
    Task.Run(async () => await foreach (var item in sequence) { /* ... */ })
);

// Потенциальная проблема: неконтролируемое потребление памяти
await foreach (var item in GetMillionsOfItemsAsync())
{
    await SlowProcessingAsync(item); // Если это медленно, память может быстро закончиться
}

public async IAsyncEnumerable<Data> GetDataAsync(
    [EnumeratorCancellation] CancellationToken token = default)
{
    while (ShouldContinue())
    {
        // Забыли проверить token.IsCancellationRequested!
        var item = await FetchNextItemAsync();
        yield return item;
    }
}

// Было:
foreach (var item in GetItems())
{
    ProcessItem(item);
}
 
// Стало (неправильно):
foreach (var item in GetItemsAsync().Result) // Блокирующий вызов!
{
    ProcessItem(item);
}

// Правильно:
await foreach (var item in GetItemsAsync())
{
    await ProcessItemAsync(item);
}

// Плохая идея:
public async IAsyncEnumerable<int> BadMixAsync()
{
    foreach (var item in GetSyncData()) // Синхронный вызов
    {
        yield return await ProcessAsync(item); // Асинхронный вызов
    }
}

// Неправильно:
IAsyncEnumerator<Data> enumerator = source.GetAsyncEnumerator();
while (await enumerator.MoveNextAsync())
{
    // Обработка
} 
// Утечка ресурсов!
 
// Правильно:
await using var enumerator = source.GetAsyncEnumerator();
while (await enumerator.MoveNextAsync())
{
    // Обработка
}

// Потенциальная проблема производительности:
await foreach (var item in GetItemsAsync())
{
    // Каждая итерация возвращается в UI-поток!
}
 
// Более эффективно:
await foreach (var item in GetItemsAsync().ConfigureAwait(false))
{
    // Выполняется в потоке из пула
}

// Запуск анализа приложения с Perfview
// perfview collect -CircularMB 1024 -NoNGenRundown -Merge:true -Zip:true YourApp.exe

// Часто диагностирует ситуацию:
var enumerator = GetItemsAsync().GetAsyncEnumerator();
// Missing: await using var enumerator = ...

!dumpasync -all    # Показывает все асинхронные операции
!dumpheap -stat    # Статистика по выделенным объектам

// Каждая итерация возвращается в UI-поток = тормоза
await foreach (var item in GetItemsAsync())
{
    // Обработка данных, не связанная с UI
    var result = ProcessData(item);
    
    // Обновление интерфейса
    dataGrid.Items.Add(result);
}

// Создаём временную коллекцию
var results = new List<ProcessedItem>();
 
// Обработка в фоновом потоке без возврата в UI
await foreach (var item in GetItemsAsync().ConfigureAwait(false))
{
    // Чистая бизнес-логика без доступа к UI
    var processed = ProcessData(item);
    results.Add(processed);
}
 
// Единожды возвращаемся в UI-поток для обновления интерфейса
await Dispatcher.InvokeAsync(() =>
{
    foreach (var result in results)
    {
        dataGrid.Items.Add(result);
    }
});

var buffer = new List<ProcessedItem>();
await foreach (var item in GetItemsAsync().ConfigureAwait(false))
{
    var processed = ProcessData(item);
    buffer.Add(processed);
    
    // Периодически обновляем UI
    if (buffer.Count >= 100)
    {
        var toDisplay = buffer.ToList();
        buffer.Clear();
        
        await Dispatcher.InvokeAsync(() =>
        {
            foreach (var result in toDisplay)
            {
                dataGrid.Items.Add(result);
            }
        });
    }
}