Flux de données asynchrones

1. Introduction

Vous connaissez déjà async et await. Ils fonctionnent bien pour des actions asynchrones "isolées", comme le téléchargement d'un seul fichier. Mais que faire si les données arrivent sous forme de flux, ou si une ressource nécessite un nettoyage asynchrone ?

Problème avec les "collections" asynchrones : Imaginez qu'il faille récupérer des millions d'enregistrements depuis une base de données. Si la méthode retourne Task<List<T>>, vous attendez que toutes les données soient chargées en mémoire. Ce n'est pas efficace et ça crée des latences. Un IEnumerable<T> synchrone ne convient pas non plus si chaque élément doit être obtenu de façon asynchrone.

Problème avec la libération asynchrone des ressources : IDisposable et using gèrent bien le nettoyage synchrone. Mais que faire si la fermeture d'une connexion réseau ou l'écriture des buffers sur disque est elle‑même asynchrone ? On ne peut pas utiliser await dans un Dispose() synchrone, ce qui peut bloquer le thread ou conduire à un nettoyage incorrect.

Pour résoudre ces problèmes, IAsyncEnumerable<T> et IAsyncDisposable ont été introduits.

2. Flux de données asynchrones

IAsyncEnumerable<T> est l'analogue asynchrone de IEnumerable<T>. Il permet de produire les éléments de la séquence de façon asynchrone, un par un, sans attendre que toutes les données soient prêtes.

Quand l'utiliser ?

• Lecture asynchrone de gros fichiers ligne par ligne : par exemple des logs de plusieurs gigaoctets.
• Streaming de données depuis le réseau ou la base de données : résultats d'une requête API qui arrivent par morceaux.
• Implémentation d'API serveurs en streaming : par exemple gRPC Streaming.
• Dans tout scénario où les données sont générées ou reçues de façon asynchrone et doivent être traitées progressivement.

Comment ça marche ?

1. IAsyncEnumerable<T> : interface avec la méthode GetAsyncEnumerator(CancellationToken cancellationToken). Le token d'annulation est important !
2. IAsyncEnumerator<T> : interface avec ValueTask<bool> MoveNextAsync() (passer à l'élément suivant) et Current (l'élément courant). Il hérite aussi de IAsyncDisposable.
3. await foreach : permet d'itérer facilement sur IAsyncEnumerable<T>. Le compilateur appelle automatiquement MoveNextAsync() et consulte Current. Important : await foreach garantit l'appel de DisposeAsync() pour l'itérateur après la fin de l'itération, même en cas d'erreur.

Créer un IAsyncEnumerable<T> avec async yield return

Vous pouvez utiliser yield return dans une méthode async qui retourne IAsyncEnumerable<T>. Cela permet de créer des générateurs asynchrones. Votre méthode peut utiliser await pour suspendre la génération, attendre une opération asynchrone, puis reprendre.

Exemple : Générateur asynchrone simple

async IAsyncEnumerable<int> GenerateNumbersAsync()
{
    for (int i = 0; i < 3; i++)
    {
        Console.WriteLine($"Génère : {i}");
        await Task.Delay(100); // Simulation de travail asynchrone
        yield return i; 
    }
}

// Utilisation:
async Task ConsumeAsyncNumbers()
{
    await foreach (var number in GenerateNumbersAsync())
    {
        Console.WriteLine($"Reçu : {number}");
    }
}
// Appelez : await ConsumeAsyncNumbers();

Exemple : Lecture de fichier ligne par ligne de façon asynchrone

async IAsyncEnumerable<string> ReadFileLinesAsync(string filePath)
{
    using var reader = new StreamReader(filePath); // 'using' ici (StreamReader implémente IAsyncDisposable)
    string? line;
    while ((line = await reader.ReadLineAsync()) != null) 
    {
        yield return line;
    }
}

// Utilisation:
async Task ProcessFileAsync()
{
    await File.WriteAllLinesAsync("data.txt", new[] { "Ligne 1", "Ligne 2", "Ligne 3" });
    await foreach (var line in ReadFileLinesAsync("data.txt")) 
    {
        Console.WriteLine($"Ligne traitée : {line}");
    }
}
// Appelez : await ProcessFileAsync();

Exemple : Générateur asynchrone avec annulation (CancellationToken)

async IAsyncEnumerable<int> GetCancelableSequence(
    [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken token = default)
{
    for (int i = 0; i < 10; i++)
    {
        token.ThrowIfCancellationRequested(); // Vérifie l'annulation
        await Task.Delay(200, token); // Task.Delay supporte aussi l'annulation via token
        yield return i;
    }
}

// Utilisation:
async Task ConsumeAndCancel()
{
    var cts = new CancellationTokenSource(500); // Annulation après 500ms
    try
    {
        await foreach (var num in GetCancelableSequence(cts.Token))
        {
            Console.WriteLine($"Reçu : {num}");
        }
    }
    catch (OperationCanceledException)
    {
        Console.WriteLine("Génération annulée !");
    }
}
// Appelez : await ConsumeAndCancel();

L'attribut [EnumeratorCancellation] permet de passer un CancellationToken au générateur asynchrone. Cela donne la possibilité d'annuler l'itération si le code appelant demande l'annulation via CancellationTokenSource. Sans cet attribut, le token ne sera pas automatiquement transmis à GetAsyncEnumerator.

3. Gestion asynchrone des ressources

Problème du IDisposable synchrone

La méthode Dispose() dans IDisposable est synchrone (void Dispose()). Vous ne pouvez pas utiliser await à l'intérieur. Si la fermeture d'une connexion DB ou le vidage de buffers vers un fichier sont des opérations longues et asynchrones, un Dispose() synchrone bloquera le thread, ce qui est mauvais pour les applications asynchrones.

Solution : IAsyncDisposable

IAsyncDisposable résout ce problème. Il contient une seule méthode : ValueTask DisposeAsync() — une méthode asynchrone pour le nettoyage.

await using

C'est l'analogue asynchrone de using. Il est destiné aux objets qui implémentent IAsyncDisposable.

• await using garantit que DisposeAsync() sera appelé à la fin du bloc où la ressource est déclarée (ou lors de la sortie due à une exception).
• Permet de libérer correctement les ressources de manière asynchrone, évitant les blocages.

Exemple : Implémentation basique de IAsyncDisposable et await using

class MyAsyncResource : IAsyncDisposable
{
    public MyAsyncResource() => Console.WriteLine("Ressource ouverte.");
    
    public async ValueTask DisposeAsync()
    {
        Console.WriteLine("Commence le nettoyage asynchrone...");
        await Task.Delay(200); // Simulation du nettoyage asynchrone
        Console.WriteLine("Nettoyage asynchrone terminé.");
    }
}

// Utilisation await using
async Task UseAndDisposeResource()
{
    await using var resource = new MyAsyncResource(); 
    Console.WriteLine("La ressource est utilisée...");
} // resource.DisposeAsync() est appelé automatiquement ici

// Appelez : await UseAndDisposeResource();

Exemple : Plusieurs blocs await using

async Task UseMultipleResources()
{
    await using var res1 = new MyAsyncResource();
    await using var res2 = new MyAsyncResource();
    Console.WriteLine("J'utilise les deux ressources...");
} // les ressources sont libérées selon LIFO (Last In, First Out) : res2.DisposeAsync() est appelé en premier, puis res1.DisposeAsync().
// Appelez : await UseMultipleResources();

Compatibilité de IAsyncEnumerable<T> avec IAsyncDisposable

Important : IAsyncEnumerator<T> (utilisé par await foreach) hérite lui‑même de IAsyncDisposable. Cela veut dire que si votre générateur asynchrone utilise des ressources (comme StreamReader dans l'exemple ci‑dessus) qui peuvent être nettoyées de façon asynchrone, await foreach s'en occupera. Il appellera DisposeAsync() sur l'itérateur quand l'itération sera terminée.