Reactive Programming in PHP with RxPHP

Reactive programming is a paradigm focused on asynchronous data streams and the propagation of changes. In PHP, this paradigm is implemented through the RxPHP library (ReactiveX for PHP), which allows you to work with events, asynchronous requests, and data sequences as observable streams (Observables).

What is RxPHP?

RxPHP is a PHP port of the popular ReactiveX (Rx) framework. It provides tools for composing asynchronous and event-driven programs using observable sequences and LINQ-like query operators.

Key concepts of RxPHP:

Observable — a data source that can emit values, errors, or a completion signal.
Observer — a subscriber that reacts to events from the Observable.
Subscription — the connection between an Observable and an Observer, allowing unsubscription.
Operators — functions for transforming, filtering, and combining streams (map, filter, merge, zip, etc.).
Scheduler — execution control (synchronous, asynchronous, with delay).

Installing RxPHP

Installation is done via Composer:

composer require reactivex/rxphp

Example 1: Creating a Simple Observable

Let's create a stream that sequentially emits three numbers and completes:

<?phprequire 'vendor/autoload.php';

use Rx\\Observable;

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

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

Execution result:

Received: 10Received: 20Received: 30Completed!

Example 2: Using map and filter Operators

Let's transform a number stream: filter even numbers and multiply them by 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 "Result: $value";});

Output:

Result: 20Result: 40

Example 3: Asynchronous Work with HttpClient

RxPHP is great for working with asynchronous HTTP requests. Let's use Observable::fromPromise with the GuzzleHttp library:

<?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);    });

Example 4: Merging Streams (merge and zip)

Often you need to combine several streams into one. Let's look at two operators:

merge — combines streams, interleaving their elements as they arrive:

<?phpuse Rx\\Observable;

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

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