Reactive Programming in PHP with RxPHP

Reactive Programming in PHP: A Practical Guide to RxPHP

Reactive programming is a paradigm focused on working with asynchronous data streams. In PHP, a traditionally synchronous language, this concept is implemented through the RxPHP library (ReactiveX for PHP). It allows handling events, HTTP requests, file streams, and user actions in a unified, declarative style.

What is RxPHP?

RxPHP is a port of the popular ReactiveX library. Main components:

Observable — a data source that "emits" events (onNext, onError, onCompleted).
Observer — a subscriber that reacts to events.
Schedulers — manage the execution context (synchronous, asynchronous, on a new thread).
Operators — functions for transforming, filtering, and combining streams (map, filter, merge, zip).

Installation and Setup

To work with RxPHP, use Composer:

composer require reactivex/rxphp

Basic example: creating an Observable and subscribing:

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

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

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

$observable->subscribe($observer);// Output:// Received: 1// Received: 2// Received: 3// Received: 4// Received: 5// Completed!?>

Basic RxPHP Operators

Operators allow transforming data without modifying the original Observable:

1. map — transform each element

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

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

2. filter — select by condition

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

// Result: 2, 4

3. merge — combine streams

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

$merged = Observable::merge($stream1, $stream2);// 1, 2, 3, 4 (order may depend on Scheduler)

Creating Custom Observables

You can create Observables from any source: databases, sockets, or custom events.

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

// Emulating asynchronous file readingfunction readFileAsync(string $filename): Observable {    return Observable::create(function($observer) use ($filename) {        $handle = fopen($filename, 'r');        if (!$handle) {            $observer->onError(new \\RuntimeException("Failed to open file"));            return;        }                while (($line = fgets($handle)) !== false) {            $observer->onNext(trim($line));        }        fclose($handle);        $observer->onCompleted();    });}

// Usage with EventLoop (requires 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 "Found error: $line";    });

$loop->run();?>