Programmation réactive en Go : bibliothèque RxGo (Reactive Extensions for Go)

Programmation réactive en Go (Golang) avec la bibliothèque RxGo

La programmation réactive est un paradigme orienté vers les flux de données asynchrones et la propagation des changements. Dans l'écosystème Go (Golang), l'une des implémentations les plus puissantes de ce paradigme est la bibliothèque RxGo (Reactive Extensions for Go). Elle permet aux développeurs de travailler avec des séquences d'événements en utilisant un style déclaratif, ce qui est particulièrement utile pour le traitement des entrées utilisateur, des requêtes réseau, des messages provenant de canaux et de toute autre source de données asynchrone.

Qu'est-ce que RxGo ?

RxGo est un port de la bibliothèque populaire ReactiveX (Reactive Extensions) vers le langage Go. L'idée principale est de créer un Observable (flux observable) qui émet des éléments, et un Observer (observateur) qui s'abonne à ce flux et réagit aux données entrantes, aux erreurs ou à la fin du flux.

La bibliothèque fournit un riche ensemble d'opérateurs pour la transformation, le filtrage, la combinaison et la gestion des flux. L'avantage clé est que le code devient concis, lisible et facile à tester, en particulier dans les scénarios complexes avec plusieurs sources d'événements.

Installation de RxGo

Pour commencer, installez le package via go get :

go get -u github.com/reactivex/rxgo/v2

Assurez-vous d'utiliser la version v2, car elle contient l'API actuelle et la prise en charge des contextes Go.

Concepts de base et exemples

1. Création d'un Observable

La manière la plus simple de créer un flux est d'utiliser rxgo.Just ou rxgo.FromChannel. Considérons un exemple avec Just :

package main\n\nimport (\n    "context"\n    "fmt"\n    "github.com/reactivex/rxgo/v2"\n)\n\nfunc main() {\n    observable := rxgo.Just(1, 2, 3, 4, 5)()\n    ch := observable.Observe()\n    for item := range ch {\n        if item.Error() {\n            fmt.Println("Erreur :", item.Err)\n        } else {\n            fmt.Println("Reçu :", item.V)\n        }\n    }\n}

Ce code crée un flux de cinq nombres. La méthode Observe() retourne un canal par lequel les éléments arrivent. Important : dans RxGo, on utilise le backpressure (contre-pression) — si le consommateur est lent, les données sont mises en mémoire tampon.

2. Opérateurs de transformation

Les opérateurs sont le cœur de la programmation réactive. Par exemple, Map transforme chaque élément :

package main\n\nimport (\n    "context"\n    "fmt"\n    "github.com/reactivex/rxgo/v2"\n)\n\nfunc main() {\n    observable := rxgo.Just(1, 2, 3)().\n        Map(func(ctx context.Context, i interface{}) (interface{}, error) {\n            num := i.(int)\n            return num * 10, nil\n        }, rxgo.WithPool(2)) // traitement parallèle\n\n    for item := range observable.Observe() {\n        fmt.Println(item.V)\n    }\n}

Sortie : 10, 20, 30. L'opérateur Map applique une fonction à chaque élément. Le paramètre rxgo.WithPool(2) active le traitement parallèle avec deux goroutines.

3. Filtrage et fusion de flux

Avec Filter, on peut filtrer les éléments indésirables, et Merge fusionne plusieurs flux :

package main\n\nimport (\n    "fmt"\n    "github.com/reactivex/rxgo/v2"\n)\n\nfunc main() {\n    // Filtrage des nombres pairs\n    even := rxgo.Just(1, 2, 3, 4, 5)().\n        Filter(func(i interface{}) bool {\n            return i.(int)%2 == 0\n        })\n\n    // Deuxième flux\n    odd := rxgo.Just(10, 20, 30)()\n\n    // Fusion\n    merged := rxgo.Merge(even, odd)\n\n    for item := range merged.Observe() {\n        fmt.Println(item.V) // 2, 4, 10, 20, 30\n    }\n}

4. Travail avec les canaux et les contextes

RxGo s'intègre parfaitement avec les canaux natifs de Go. Créons un Observable à partir d'un canal et ajoutons