Programmation réactive en PHP avec RxPHP : guide complet

Programmation réactive en PHP avec RxPHP

\n\n

La programmation réactive est un paradigme orienté vers les flux de données asynchrones et la propagation des changements. En PHP, ce paradigme est implémenté via la bibliothèque RxPHP (ReactiveX pour PHP), qui permet de travailler avec des événements, des requêtes asynchrones et des séquences de données comme des flux observables (Observables).

\n\n

Qu'est-ce que RxPHP ?

\n

RxPHP est la version portée en PHP du célèbre framework ReactiveX (Rx). Il fournit des outils pour composer des programmes asynchrones et orientés événements à l'aide de séquences observables et d'opérateurs de requêtes de type LINQ.

\n\n

Concepts principaux de RxPHP :

\n

\n
• Observable — source de données qui peut émettre des valeurs, des erreurs ou un signal de terminaison.
\n
• Observer — abonné qui réagit aux événements de l'Observable.
\n
• Subscription — lien entre l'Observable et l'Observer, permettant de se désabonner.
\n
• Operators — fonctions pour transformer, filtrer et combiner les flux (map, filter, merge, zip, etc.).
\n
• Scheduler — gestion de l'exécution (synchrone, asynchrone, avec délai).
\n

\n\n

Installation de RxPHP

\n

L'installation se fait via Composer :

\n

composer require reactivex/rxphp

\n\n

Exemple 1 : Création d'un Observable simple

\n

Créons un flux qui émet séquentiellement trois nombres puis se termine :

\n

<?php\nrequire 'vendor/autoload.php';\n\nuse Rx\\Observable;\n\n$observable = Observable::fromArray([10, 20, 30]);\n\n$disposable = $observable->subscribe(\n    function ($value) {\n        echo "Reçu : $value\\n";\n    },\n    function (\\Throwable $error) {\n        echo "Erreur : " . $error->getMessage() . "\\n";\n    },\n    function () {\n        echo "Terminé !\\n";\n    }\n);\n

\n

Résultat de l'exécution :

\n

Reçu : 10\nReçu : 20\nReçu : 30\nTerminé !

\n\n

Exemple 2 : Utilisation des opérateurs map et filter

\n

Transformons un flux de nombres : filtrons les pairs et multiplions-les par 10 :

\n

<?php\nuse Rx\\Observable;\n\n$observable = Observable::fromArray([1, 2, 3, 4, 5])\n    ->filter(function ($value) {\n        return $value % 2 === 0;\n    })\n    ->map(function ($value) {\n        return $value * 10;\n    });\n\n$observable->subscribe(function ($value) {\n    echo "Résultat : $value\\n";\n});\n

\n

Sortie :

\n

Résultat : 20\nRésultat : 40

\n\n

Exemple 3 : Travail asynchrone avec HttpClient

\n

RxPHP est parfait pour travailler avec des requêtes HTTP asynchrones. Utilisons Observable::fromPromise avec la bibliothèque GuzzleHttp :

\n

<?php\nuse Rx\\Observable;\nuse GuzzleHttp\\Client;\nuse GuzzleHttp\\Promise;\n\n$client = new Client();\n$promise1 = $client->getAsync('https://api.example.com/data1');\n$promise2 = $client->getAsync('https://api.example.com/data2');\n\nObservable::fromArray([$promise1, $promise2])\n    ->flatMap(function ($promise) {\n        return Observable::fromPromise($promise);\n    })\n    ->map(function ($response) {\n        return json_decode($response->getBody(), true);\n    })\n    ->subscribe(function ($data) {\n        print_r($data);\n    });\n

\n\n

Exemple 4 : Fusion de flux (merge et zip)

\n

Il est souvent nécessaire de fusionner plusieurs flux en un seul. Examinons deux opérateurs :

\n\n

merge — fusionne les flux en entrelaçant leurs éléments au fur et à mesure de leur arrivée :

\n

<?php\nuse Rx\\Observable;\n\n$stream1 = Observable::fromArray(['A1', 'A2', 'A3']);\n$stream2 = Observable::fromArray(['B1', 'B2']);\n\nObservable::merge([$stream1, $stream2])\n    ->subscribe(function ($value) {\n        echo "$value ";\n    });\n// Sortie : A1 B1 A2 B2 A3\n

\n\n