Programação Reativa em PHP com RxPHP

Programação Reativa em PHP: Guia Prático de RxPHP

Programação reativa é um paradigma focado no trabalho com fluxos de dados assíncronos. Em PHP, uma linguagem tradicionalmente síncrona, este conceito é implementado através da biblioteca RxPHP (ReactiveX para PHP). Ela permite processar eventos, requisições HTTP, fluxos de arquivos e ações do usuário em um estilo único e declarativo.

O que é RxPHP?

RxPHP é um port da popular biblioteca ReactiveX. Os componentes principais são:

Observable — fonte de dados que "emite" eventos (onNext, onError, onCompleted).
Observer — assinante que reage aos eventos.
Schedulers — gerenciam o contexto de execução (síncrono, assíncrono, em uma nova thread).
Operadores — funções para transformação, filtragem e combinação de fluxos (map, filter, merge, zip).

Instalação e Configuração

Para trabalhar com RxPHP, use o Composer:

composer require reactivex/rxphp

Exemplo básico: criar um Observable e assinar:

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

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

$observer = new class implements ObserverInterface {    public function onNext($value) {        echo "Recebido: $value";    }    public function onError(\\Throwable $error) {        echo "Erro: " . $error->getMessage() . "";    }    public function onCompleted() {        echo "Concluído!";    }};

$observable->subscribe($observer);// Saída:// Recebido: 1// Recebido: 2// Recebido: 3// Recebido: 4// Recebido: 5// Concluído!?>

Operadores Principais do RxPHP

Os operadores permitem transformar dados sem modificar o Observable original:

1. map — transformação 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 — seleção por condição

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

// Resultado: 2, 4

3. merge — combinação de fluxos

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

$merged = Observable::merge($stream1, $stream2);// 1, 2, 3, 4 (a ordem pode depender do Scheduler)

Criação de Observables Personalizados

Você pode criar Observables a partir de qualquer fonte: banco de dados, sockets ou eventos personalizados.

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

// Emulação de leitura assíncrona de arquivofunction readFileAsync(string $filename): Observable {    return Observable::create(function($observer) use ($filename) {        $handle = fopen($filename, 'r');        if (!$handle) {            $observer->onError(new \\RuntimeException("Não foi possível abrir o arquivo"));            return;        }                while (($line = fgets($handle)) !== false) {            $observer->onNext(trim($line));        }        fclose($handle);        $observer->onCompleted();    });}

// Uso com EventLoop (requer 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 "Erro encontrado: $line";    });

$loop->run();?>