Rx.NET + Channels + BackgroundService avec les flux asynchrones en .NET

Dans l’article précédent « Rx.NET dans une API Web : modéliser les flux plutôt que les appels« , nous avons posé une base importante : « arrêter de penser en appels, commencer à penser en flux ».

Sur le papier, Rx.NET coche toutes les cases. On compose, on transforme, on enchaîne. C’est lisible et élégant.

Mais dès qu’on sort du labo et qu’on arrive sur une vraie API, le décor change. Les événements n’arrivent pas proprement, ils arrivent en rafale. L’ensemble des traitements prennent du temps et en plus les dépendances externes ralentissent tout.

Et c’est là que le flux Rx.NET commence va montrer ses limites.

Le flux parfait… dans un monde imparfait

Un IObservable fonctionne très bien tant que le rythme est maîtrisé.

var stream = Observable.FromEventPattern<OrderCreatedEvent>(handler);
stream
    .Select(e => Process(e))
    .Subscribe();

Ce genre de pipeline est agréable à lire. Il repose sur l’hypothèse implicite que le système est capable d’absorber tout ce qu’on lui envoie.

Dans la réalité, ce n’est jamais vrai. Les événements s’accumulent, les traitements prennent du retard, et tu n’as aucun moyen simple de ralentir la source.

C’est au niveau du débit que les problèmes commencent.

Remettre une notion essentielle : la pression

Ce que Rx.NET ne force pas, les Channel<T> le rendent explicite.

On arrête de consommer directement un flux. On introduit un point de passage contrôlé.

var channel = Channel.CreateBounded<Order>(100);

Ce simple changement modifie complètement le comportement du système. On bufferise, on rejette la demande quand le système est saturé.

Côté API, ça devient presque trivial :

app.MapPost("/orders", async (Order order) =>
{
    await channel.Writer.WriteAsync(order);
});

À partir de là, l’API n’est plus responsable du traitement. Elle se contente d’alimenter le pipeline. C’est un changement subtil, mais fondamental.

Faire tourner le système en continu

Le BackgroundService vient donner une réalité à ce pipeline.

On ne déclenche plus le traitement au moment de la requête et on laisse un worker consommer à son rythme.

public class OrderWorker : BackgroundService
{
    private readonly Channel<Order> _channel;
    public OrderWorker(Channel<Order> channel)
    {
        _channel = channel;
    }
    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        await foreach (var order in _channel.Reader.ReadAllAsync(stoppingToken))
        {
            await Process(order);
        }
    }
}

Ce qui est intéressant ici, c’est le découplage.

L’API pousse. Le worker consomme. Entre les deux, un buffer absorbe les variations.

On commence à avoir quelque chose de stable.

Réintroduire Rx.NET au bon endroit

À ce stade, on pourrait se contenter du Channel et du BackgroundService car ça fonctionne déjà très bien.

Mais on perd la capacité à composer intelligemment les flux.

C’est là que Rx.NET revient avec un rôle différent.

On ne l’utilise plus pour recevoir les événements. On l’utilise pour les transformer.

var observable = channel.Reader
    .ReadAllAsync()
    .ToObservable();
observable
    .Buffer(TimeSpan.FromSeconds(5))
    .Where(batch => batch.Any())
    .Subscribe(async batch =>
    {
        await ProcessBatch(batch);
    });

On garde la robustesse du Channel, et on retrouve la puissance de composition de Rx.

Une architecture qui tient dans le temps

Quand on prend un peu de recul, on voit apparaître un pipeline assez clair.

C’est un flux structuré, avec des responsabilités bien définies.

Un exemple concret

Prenons quelque chose de simple, dans la continuité de l’article précédent (« Rx.NET dans une API Web : modéliser les flux plutôt que les appels« ).

public record EmployeeEvent(string Type, int EmployeeId);

L’API ne fait qu’émettre :

await channel.Writer.WriteAsync(new EmployeeEvent("CREATED", id));

Derrière, le pipeline devient intéressant :

channel.Reader
    .ReadAllAsync()
    .ToObservable()
    .GroupBy(e => e.EmployeeId)
    .Subscribe(group =>
    {
        group
            .Throttle(TimeSpan.FromSeconds(1))
            .Subscribe(e => HandleEmployeeEvent(e));
    });

On commence à raisonner en termes de flux métier. On regroupe, on filtre, on contrôle le bruit.

On ne fait plus du traitement brut, on fait fait de la modélisation.

Ce qui change vraiment

On change la manière de penser le système avec un pattern qui rend le code plus propre.

On gagne en résilience, en absorbant la montée en charge on ne casse plus l’application.

Conclusion

Rx.NET reste un outil extrêmement puissant, mais utilisé seul, il montre vite ses limites dès qu’on entre dans un contexte réel.

Les Channels apportent la régulation qui manque. Le BackgroundService donne un cadre d’exécution stable.

Et au milieu, Rx.NET retrouve toute sa valeur, là où il est le plus pertinent : dans la transformation du flux.

Ce trio fonctionne parce qu’il respecte une réalité ou un flux, n’est pas juste des données qui passent. C’est aussi un système qui encaisse, et qui s’adapte.

Pour aller plus loin.

ReactiveX

Télécharger le sample sur GitHub
https://github.com/xraboteu/rx-channels-worker-pipeline