Programmation réactive en PHP avec RxPHP

Programmation réactive en PHP : guide pratique de RxPHP

La programmation réactive est un paradigme orienté vers le travail avec des flux de données asynchrones. En PHP, un langage traditionnellement synchrone, ce concept est implémenté via la bibliothèque RxPHP (ReactiveX pour PHP). Elle permet de traiter les événements, les requêtes HTTP, les flux de fichiers et les actions utilisateur dans un style unique et déclaratif.

Qu'est-ce que RxPHP ?

RxPHP est un port de la bibliothèque populaire ReactiveX. Les composants principaux :

Observable — source de données qui "émet" des événements (onNext, onError, onCompleted).
Observer — abonné qui réagit aux événements.
Schedulers — gèrent le contexte d'exécution (synchrone, asynchrone, sur un nouveau thread).
Opérateurs — fonctions pour transformer, filtrer et combiner les flux (map, filter, merge, zip).

Installation et configuration

Pour travailler avec RxPHP, utilisez Composer :

composer require reactivex/rxphp

Exemple de base : création d'un Observable et abonnement :

<?phpuse Rx\\Observable;use Rx\\ObserverInterface;

$observable = Observable::fromArray([1, 2, 3, 4, 5]);

$observer = new class implements ObserverInterface {    public function onNext($value) {        echo "Reçu : $value";    }    public function onError(\\Throwable $error) {        echo "Erreur : " . $error->getMessage() . "";    }    public function onCompleted() {        echo "Terminé !";    }};

$observable->subscribe($observer);// Sortie :// Reçu : 1// Reçu : 2// Reçu : 3// Reçu : 4// Reçu : 5// Terminé !?>

Opérateurs principaux de RxPHP

Les opérateurs permettent de transformer les données sans modifier l'Observable d'origine :

1. map — transformation de chaque élément

$observable = Observable::fromArray([1, 2, 3])    ->map(function($value) {        return $value * 2;    });

$observable->subscribe(function($value) {    echo $value . "";});// 2, 4, 6

2. filter — sélection selon une condition

$observable = Observable::fromArray([1, 2, 3, 4, 5])    ->filter(function($value) {        return $value % 2 === 0;    });

// Résultat : 2, 4

3. merge — fusion de flux

$stream1 = Observable::fromArray([1, 2]);$stream2 = Observable::fromArray([3, 4]);

$merged = Observable::merge($stream1, $stream2);// 1, 2, 3, 4 (l'ordre peut dépendre du Scheduler)

Création d'Observables personnalisés

Vous pouvez créer des Observables à partir de n'importe quelle source : base de données, sockets ou événements utilisateur.

<?phpuse Rx\\Observable;use Rx\\Scheduler\\EventLoopScheduler;

// Émulation d'une lecture asynchrone de fichierfunction readFileAsync(string $filename): Observable {    return Observable::create(function($observer) use ($filename) {        $handle = fopen($filename, 'r');        if (!$handle) {            $observer->onError(new \\RuntimeException("Impossible d'ouvrir le fichier"));            return;        }                while (($line = fgets($handle)) !== false) {            $observer->onNext(trim($line));        }        fclose($handle);        $observer->onCompleted();    });}

// Utilisation avec EventLoop (nécessite react/event-loop)$loop = \\React\\EventLoop\\Loop::get();$scheduler = new EventLoopScheduler($loop);

readFileAsync('data.log')    ->subscribeOn($scheduler)    ->filter(function($line) {        return strpos($line, 'ERROR') !== false;    })    ->subscribe(function($line) {        echo "Erreur trouvée : $line";    });

$loop->run();?>