Programación Reactiva en PHP con RxPHP

La programación reactiva es un paradigma orientado a flujos de datos asíncronos y la propagación de cambios. En PHP, este paradigma se implementa a través de la biblioteca RxPHP (ReactiveX para PHP), que permite trabajar con eventos, solicitudes asíncronas y secuencias de datos como flujos observables (Observables).

¿Qué es RxPHP?

RxPHP es la versión portada a PHP del popular framework ReactiveX (Rx). Proporciona herramientas para la composición de programas asíncronos y orientados a eventos mediante secuencias observables y operadores de consulta similares a LINQ.

Conceptos principales de RxPHP:

Observable — fuente de datos que puede emitir valores, errores o una señal de finalización.
Observer — suscriptor que reacciona a los eventos del Observable.
Subscription — vínculo entre Observable y Observer que permite cancelar la suscripción.
Operators — funciones para transformar, filtrar y combinar flujos (map, filter, merge, zip, etc.).
Scheduler — control de la ejecución (síncrona, asíncrona, con retardo).

Instalación de RxPHP

La instalación se realiza mediante Composer:

composer require reactivex/rxphp

Ejemplo 1: Creación de un Observable simple

Creemos un flujo que emita tres números secuencialmente y finalice:

<?phprequire 'vendor/autoload.php';

use Rx\\Observable;

$observable = Observable::fromArray([10, 20, 30]);

$disposable = $observable->subscribe(    function ($value) {        echo "Recibido: $value";    },    function (\\Throwable $error) {        echo "Error: " . $error->getMessage() . "";    },    function () {        echo "¡Completado!";    });

Resultado de la ejecución:

Recibido: 10Recibido: 20Recibido: 30¡Completado!

Ejemplo 2: Uso de los operadores map y filter

Transformemos un flujo de números: filtremos los pares y multipliquémoslos por 10:

<?phpuse Rx\\Observable;

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

$observable->subscribe(function ($value) {    echo "Resultado: $value";});

Salida:

Resultado: 20Resultado: 40

Ejemplo 3: Trabajo asíncrono con HttpClient

RxPHP es ideal para trabajar con solicitudes HTTP asíncronas. Usemos Observable::fromPromise con la biblioteca GuzzleHttp:

<?phpuse Rx\\Observable;use GuzzleHttp\\Client;use GuzzleHttp\\Promise;

$client = new Client();$promise1 = $client->getAsync('https://api.example.com/data1');$promise2 = $client->getAsync('https://api.example.com/data2');

Observable::fromArray([$promise1, $promise2])    ->flatMap(function ($promise) {        return Observable::fromPromise($promise);    })    ->map(function ($response) {        return json_decode($response->getBody(), true);    })    ->subscribe(function ($data) {        print_r($data);    });

Ejemplo 4: Combinación de flujos (merge y zip)

A menudo es necesario combinar varios flujos en uno solo. Veamos dos operadores:

merge — combina flujos, intercalando sus elementos a medida que llegan:

<?phpuse Rx\\Observable;

$stream1 = Observable::fromArray(['A1', 'A2', 'A3']);$stream2 = Observable::fromArray(['B1', 'B2']);

Observable::merge([$stream1, $stream2])    ->subscribe(function ($value) {        echo "$value ";    });// Salida: A1 B1 A2 B2 A3