Programación reactiva en PHP: biblioteca RxPHP para flujos asíncronos

Programación reactiva en PHP: guía práctica de RxPHP

La programación reactiva es un paradigma orientado al trabajo con flujos de datos asíncronos. En PHP, un lenguaje tradicionalmente síncrono, este concepto se implementa a través de la biblioteca RxPHP (ReactiveX para PHP). Permite manejar eventos, solicitudes HTTP, flujos de archivos y acciones de usuario en un estilo único y declarativo.

¿Qué es RxPHP?

RxPHP es un puerto de la popular biblioteca ReactiveX. Componentes principales:

• Observable — fuente de datos que "emite" eventos (onNext, onError, onCompleted).
• Observer — suscriptor que reacciona a los eventos.
• Schedulers — gestionan el contexto de ejecución (síncrono, asíncrono, en un nuevo hilo).
• Operadores — funciones para transformar, filtrar y combinar flujos (map, filter, merge, zip).

Instalación y configuración

Para trabajar con RxPHP use Composer:

composer require reactivex/rxphp

Ejemplo básico: creación de Observable y suscripción:

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

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

$observer = new class implements ObserverInterface {    public function onNext($value) {        echo "Recibido: $value";    }    public function onError(\\Throwable $error) {        echo "Error: " . $error->getMessage() . "";    }    public function onCompleted() {        echo "¡Completado!";    }};

$observable->subscribe($observer);// Salida:// Recibido: 1// Recibido: 2// Recibido: 3// Recibido: 4// Recibido: 5// ¡Completado!?>

Operadores principales de RxPHP

Los operadores permiten transformar datos sin modificar el Observable original:

1. map — transformación de cada elemento

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

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

2. filter — selección por condición

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

// Resultado: 2, 4

3. merge — combinación de flujos

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

$merged = Observable::merge($stream1, $stream2);// 1, 2, 3, 4 (el orden puede depender del Scheduler)

Creación de Observables personalizados

Se pueden crear Observables desde cualquier fuente: bases de datos, sockets o eventos personalizados.

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

// Emulación de lectura asíncrona de archivofunction readFileAsync(string $filename): Observable {    return Observable::create(function($observer) use ($filename) {        $handle = fopen($filename, 'r');        if (!$handle) {            $observer->onError(new \\RuntimeException("No se pudo abrir el archivo"));            return;        }                while (($line = fgets($handle)) !== false) {            $observer->onNext(trim($line));        }        fclose($handle);        $observer->onCompleted();    });}

// Uso con EventLoop (requiere 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 "Error encontrado: $line";    });

$loop->run();?>

Manejo de errores y finalización

En el estilo reactivo, los errores se manejan de manera