# ReactiveX/RxPHP: Observables and Async Streams in PHP

> ReactiveX/RxPHP is a PHP library for composing asynchronous and event-based programs with observable streams. It suits developers who need scheduled values, operators, cancellation through disposal, or interoperability with ReactPHP promises, but it adds a scheduler and reactive programming model that ordinary procedural code may not need.

**ReactiveX/RxPHP** — Reactive extensions for PHP

- Repository: https://github.com/ReactiveX/RxPHP
- Stars: 1,731 · Forks: 140
- Language: PHP
- License: MIT
- Published: 2026-10-16 · Updated: 2026-10-16 · Language: en
- Canonical page: https://hysenlabs.com/projects/reactivex-rxphp

## The stream composition problem RxPHP addresses

RxPHP targets PHP programs that consume values over time. Those values may arrive at fixed intervals, originate in asynchronous operations, or require transformations after each notification. Managing the callback, timing, completion, and error paths separately can spread control flow across the application. RxPHP instead represents the sequence as an `Observable` and provides operators for composing those sequences.

The library is aimed at developers working with asynchronous or event-based behavior, particularly those willing to use an event loop. The README recommends ReactPHP's event loop but says any event loop should work. It also documents built-in support for ReactPHP promises, which gives RxPHP a transition path when an existing library exposes only a promise.

RxPHP is not a general-purpose replacement for arrays or ordinary function calls. Its clearest use is a pipeline whose lifetime and value flow matter. A pipeline that emits every second, merges asynchronous work, filters values, or waits for several streams before completing is a better candidate than a loop that prints three array entries immediately. That distinction matters because adopting RxPHP means accepting subscription and scheduler concepts as part of application design.

## Observable creation, operators, and subscription flow

The simplest documented flow creates an Observable from an array and subscribes callbacks to it. `Observable::fromArray()` produces a sequence, while `subscribe()` receives separate handlers for the next value, an `Exception`, and completion. The README's example emits `1`, `2`, `3`, and `4`, then invokes the completion callback.

This stream model is primarily lazy. The interval example creates `Observable::interval(1000)`, takes five values, and uses `flatMap()` to turn each value into another Observable. The operator chain therefore controls which values reach the final subscriber and what observable operation is applied at each stage.

The operator set is visible in the repository's `demo/` directories. Examples include `bufferWithCount`, `combineLatest`, `concat`, `count`, `defer`, `delay`, `distinct`, `groupBy`, and `forkJoin`. Based on that layout, RxPHP places more than one operator demo near the source package rather than requiring a separate example project for each behavior. The README does not claim that every ReactiveX language implementation offers an identical API, so developers should verify the exact operators they need against the repository documentation and source.

## Installing RxPHP and configuring its default scheduler

The installation instructions require Composer, an event loop, and one default scheduler setup for an application that uses RxPHP itself. The README recommends ReactPHP. Install the event loop first:

```bash
composer require react/event-loop
```

Then install RxPHP:

```bash
composer require reactivex/rxphp
```

The first scheduled program loads Composer's autoloader, creates a ReactPHP event loop, and registers an `EventLoopScheduler` as the default scheduler. The README states that the default scheduler only needs to be set once.

```php
<?php

require_once __DIR__ . '/vendor/autoload.php';

use Rx\Observable;
use React\EventLoop\Factory;
use Rx\Scheduler;

$loop = Factory::create();

Scheduler::setDefaultFactory(function() use($loop){
    return new Scheduler\EventLoopScheduler($loop);
});

Observable::interval(1000)
    ->take(5)
    ->flatMap(function ($i) {
        return Observable::of($i + 1);
    })
    ->subscribe(function ($e) {
        echo $e, PHP_EOL;
    });

$loop->run();
```

After `$loop->run()` drives the event loop, the interval and `take(5)` boundaries allow five values to reach the subscriber. The scheduler setup is not optional for this application-level example. The README specifically notes that demos bootstrap the scheduler automatically, while a separate project must set its default scheduler.

## Running the bundled RxPHP demos

The repository includes a `demo/` directory with executable examples, so the quickest first use does not require designing a pipeline from scratch. Clone the repository, install its dependencies, and run the interval example:

```bash
git clone https://github.com/ReactiveX/RxPHP.git
cd RxPHP
composer install
php demo/interval/interval.php
```

The clone command creates the local `RxPHP` directory, `cd RxPHP` enters it, `composer install` installs the repository's declared dependencies, and the final command starts the documented interval demo. Other first-use choices include `demo/fromArray`, `demo/filter`, `demo/flatMap`, and `demo/forkJoin`.

The demo bootstrap matters when comparing this path with application setup. The README says the demos automatically bootstrap the scheduler, whereas application code needs its own default scheduler. A demo running correctly therefore confirms the local example and dependency installation, but it does not remove the scheduler configuration requirement from an external project.

## ReactPHP promise bridges and the alternative model

RxPHP documents two directions of interoperability with ReactPHP promises. One mixes a promise into an Observable pipeline with `Observable::fromPromise()`. In the documented case, each interval value is combined with `React\Promise\resolve(42 + $i)` inside `flatMap()`.

```php
Observable::interval(1000)
    ->flatMap(function ($i) {
        return Observable::fromPromise(\React\Promise\resolve(42 + $i));
    })
    ->subscribe(function ($v) {
        echo $v . PHP_EOL;
    });
```

The other direction converts an Observable into a promise with `toPromise()`. The README also notes that this can be useful for libraries that use generators or coroutines.

```php
$observable = Observable::interval(1000)
    ->take(10)
    ->toArray()
    ->map('json_encode');

$promise = $observable->toPromise();
```

A promise remains the simpler alternative when the required behavior is a single eventual result. An Observable represents a stream of notifications and adds operators for ongoing sequences. RxPHP is the relevant choice when multiple values, transformations, or asynchronous sources must be composed. A promise remains the relevant boundary when an existing consumer expects a promise, which is why RxPHP exposes conversion in both directions.

## RxPHP maintenance, upgrades, and operational limits

The repository is not archived. Its last push was on 2026-04-08, so it should not be described as actively maintained based on that date alone. The most recent listed release is `2.1.0`, dated 2025-10-27, following `2.0.13` from 2024-09-21 and `2.0.12` from 2023-11-27. The repository includes `CHANGELOG.md`, `CONTRIBUTING.md`, `composer.json`, `phpstan.neon.dist`, `phpunit.xml.dist`, `rector.php`, `src/`, and `test/`, but the README does not define a compatibility policy for older PHP or Composer versions.

Upgrading therefore requires checking the dependency and API changes associated with the selected release rather than assuming that all ReactiveX versions are interchangeable. The Composer package name is `reactivex/rxphp`, and dependency management remains tied to Composer.

RxPHP is licensed under the MIT License, with details in `LICENSE`. That permits use under the license's terms, but this review does not interpret legal obligations for a particular deployment. The practical limits are architectural: applications need a scheduler for the documented setup, event-driven execution requires an event loop to be driven, and operators that change stream boundaries can alter timing and completion behavior.

## Conclusion

RxPHP is a reasonable choice for PHP developers building scheduled streams, transformations across asynchronous events, or bridges between Observable pipelines and ReactPHP promises. It is a poor fit for a short synchronous procedure or code whose complexity would increase once operators, subscription lifecycles, and scheduler configuration enter the picture. Verify first that your PHP application can support Composer, an event loop, and the default scheduler configuration shown in the project's installation example, then inspect whether the required operators and promise behavior match the pipeline you intend to build.

## FAQ

### How do I install RxPHP in a PHP project?

Install `react/event-loop` and `reactivex/rxphp` with Composer, then configure a default scheduler in application code. The demos bootstrap the scheduler automatically, but external projects must set it themselves.

### How do I create and consume an Observable in RxPHP?

Create one with methods such as `Observable::fromArray()` or `Observable::interval()`, then pass next-value, error, and completion callbacks to `subscribe()`. Operators such as `take()` and `flatMap()` can be inserted into the stream before subscription.

### Can RxPHP work with ReactPHP promises?

Yes. `Observable::fromPromise()` brings a ReactPHP promise into an Observable pipeline, while `toPromise()` converts an Observable into a promise for consumers that expect that interface.

## Sources

- [Issues](https://github.com/ReactiveX/RxPHP/issues)
- [License: MIT](https://github.com/ReactiveX/RxPHP/blob/master/LICENSE)
- [ReactiveX/RxPHP on GitHub](https://github.com/ReactiveX/RxPHP)
- [README](https://github.com/ReactiveX/RxPHP/blob/master/README.md)
- [Releases](https://github.com/ReactiveX/RxPHP/releases)

---

Hysen Labs editorial analysis, written from the project's own repository and release notes. Cite the canonical page: https://hysenlabs.com/projects/reactivex-rxphp