Programação Reativa em PHP com RxPHP

Programação reativa é um paradigma orientado a fluxos de dados assíncronos e propagação de mudanças. Em PHP, esse paradigma é implementado através da biblioteca RxPHP (ReactiveX para PHP), que permite trabalhar com eventos, requisições assíncronas e sequências de dados como fluxos observáveis (Observables).

O que é RxPHP?

RxPHP é a versão portada para PHP do popular framework ReactiveX (Rx). Ela fornece ferramentas para composição de programas assíncronos e orientados a eventos usando sequências observáveis e operadores de consulta semelhantes ao LINQ.

Conceitos principais do RxPHP:

Observable — fonte de dados que pode emitir valores, erros ou um sinal de conclusão.
Observer — assinante que reage aos eventos do Observable.
Subscription — conexão entre Observable e Observer, permitindo cancelar a assinatura.
Operators — funções para transformar, filtrar e combinar fluxos (map, filter, merge, zip, etc.).
Scheduler — gerenciamento de execução (síncrona, assíncrona, com atraso).

Instalação do RxPHP

A instalação é feita via Composer:

composer require reactivex/rxphp

Exemplo 1: Criando um Observable simples

Vamos criar um fluxo que emite três números sequencialmente e finaliza:

<?phprequire 'vendor/autoload.php';

use Rx\\Observable;

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

$disposable = $observable->subscribe(    function ($value) {        echo "Recebido: $value";    },    function (\\Throwable $error) {        echo "Erro: " . $error->getMessage() . "";    },    function () {        echo "Concluído!";    });

Resultado da execução:

Recebido: 10Recebido: 20Recebido: 30Concluído!

Exemplo 2: Usando operadores map e filter

Transformando um fluxo de números: filtrando os pares e multiplicando 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";});

Saída:

Resultado: 20Resultado: 40

Exemplo 3: Trabalhando com HttpClient assíncrono

RxPHP é excelente para trabalhar com requisições HTTP assíncronas. Usando Observable::fromPromise com a 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);    });

Exemplo 4: Combinando fluxos (merge e zip)

Frequentemente é necessário combinar vários fluxos em um. Vamos ver dois operadores:

merge — combina fluxos, intercalando seus elementos conforme chegam:

<?phpuse Rx\\Observable;

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

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