Rxphp Laravel Package

Getting Started

Install

composer require reactivex/rxphp
# only if you need real async (timers, I/O concurrency):
composer require react/event-loop

RxPHP has no Laravel service provider, no config file, and no facade — it is a pure library. There is nothing to register in config/app.php; you use Rx\Observable; and go. That is the whole setup, and it is the main reason the package feels "invisible" on a first pass.

Where to look first

• src/Observable.php — the static factories (fromArray, fromIterator, create, interval, timer, range, defer, merge, concat, zip, combineLatest, empty, never, error) and every operator method. It is a very large file; treat it as the API index.
• src/Observer/* and src/Disposable/* — how values leave a stream and how you cancel it.
• src/Scheduler/* — this is where 80% of the surprises live, especially EventLoopScheduler vs ImmediateScheduler.
• The demo/ directory and tests/ in the repo — the tests double as marble-diagram documentation and are the fastest way to learn operator semantics.
• For operator names/semantics, the RxJS docs are the de facto reference; RxPHP is a port and the names mostly line up (map, filter, flatMap, debounce, scan, zip, …), with a few select/where-style aliases.

First use case

The honest first use case is a lazy, cancellable pipeline over a data source that is awkward as a loop — paginated API results, CSV rows, a queue of jobs:

use Rx\Observable;
use Rx\Observer\CallbackObserver;

Observable::fromIterator($this->fetchAllPagesAsGenerator())   // lazy
    ->filter(fn ($row) => $row['status'] === 'active')
    ->map(fn ($row) => $this->transform($row))
    ->bufferWithCount(500)                                     // batch writes
    ->subscribe(new CallbackObserver(
        fn (array $batch) => DB::table('rows')->insert($batch),
        fn (Throwable $e) => report($e),
        fn () => $this->info('done'),
    ));

Note what did not happen: nothing ran until subscribe. That laziness is the package's core mental model and the source of most bugs.

Implementation Patterns

Wrap side-effectful sources in Observable::create (or defer). Each subscription should re-execute the source, not reuse a stale array or an already-consumed generator. Observable::defer(fn () => Observable::fromIterator(...)) guarantees a fresh iterator per subscriber and is the safe default when you are not sure.

Fan-out / fan-in with merge, zip, combineLatest. Aggregating three independent API calls or three DB shards:

Observable::forkJoin([            // or zip() if you want pairwise semantics
    $this->observableUsers(),
    $this->observableOrders(),
    $this->observableInvoices(),
])->map(fn (array $results) => $this->buildDashboard(...$results));

zip pairs strictly and stops at the shortest stream; combineLatest emits on every change and needs every source to have emitted once. Pick deliberately — the difference bites when one source is slow or empty.

Multicast with a Subject for an in-process event bus. A Subject is both an observer and an observable, which makes it the natural bridge between imperative Laravel code (Event::listen, a queue job) and a reactive pipeline. Push with $subject->onNext($event), subscribe N consumers, and use ReplaySubject when late subscribers need history.

Keep flatMap for concurrency, concatMap for order. flatMap interleaves inner streams and does not preserve order; when you page through results and order matters, concatMap is the correct operator. For "latest wins" (search-as-you-type, debounced queries) combine debounce/throttle with a switch-style operator rather than hand-rolling flags.

Batching is the backpressure story. RxPHP has no real backpressure. When a producer outruns a consumer, the idiomatic fix is bufferWithCount() or a windowed operator before the slow stage, not sleep(). This is also how you turn per-row writes into bulk inserts.

Error handling belongs in the pipeline. ->retry(3), ->catch(fn ($e, $obs) => $obs->startWith($fallback)), ->finally(...) compose far more cleanly than try/catch around a loop, and they keep the retry policy attached to the operation it describes. Always pass an error callback to subscribe — an unhandled onError is a fatal RuntimeException in RxPHP, not a silent log line.

Scheduling: default is synchronous. Out of the box, Scheduler::getDefault() is an immediate/trampoline scheduler, so everything runs on the calling stack. For timers or concurrency in an Artisan worker or a ReactPHP process, install the event loop scheduler globally:

use Rx\Scheduler;
use Rx\Scheduler\EventLoopScheduler;

Scheduler::setDefaultFactory(fn () => new EventLoopScheduler($loop));

Then ->subscribeOn($scheduler) / ->observeOn($scheduler) control where work starts vs where values are delivered — a distinction worth internalising early.

Testing with TestScheduler. Inject a TestScheduler, drive virtual time forward, and assert on recorded notifications instead of sleep()-ing. This is the only way to test debounce/timeout/retry logic without flaky, slow suites.

In Laravel specifically: keep RxPHP out of the request/response hot path unless the pipeline is synchronous and short. Use it inside queued jobs, Artisan commands (php artisan worker:consume), and long-running consumers. Wrap the pipeline in a job when you need retries/visibility from the queue system; use RxPHP inside the job for the fan-out and streaming logic.

Gotchas and Tips

• Cold by default. Every subscribe re-runs the source. Three subscribers to an HTTP observable = three HTTP calls. Use ->share(), or ->publish()->refCount(), when you want one execution fanned out. Conversely, do not sprinkle share() on cold observables you want to re-run per subscriber.
• Nothing happens without subscribe. A pipeline you built but never subscribed is dead code that looks alive. Conversely, subscribing inside another operator's closure creates nested subscriptions that are easy to leak.
• The error callback is mandatory in practice. If you omit it, RxPHP throws on onError. In a daemon, that kills the process. Pass an explicit error observer or use CallbackObserver consistently.
• interval/timer/delay block or fail without an event loop scheduler. With the default scheduler they can spin the process. If a timer seems to hang, check which scheduler is installed before anything else.
• Disposal discipline. Keep the returned Disposable and dispose it in finally, or hold a CompositeDisposable at the worker level and dispose on shutdown/SIGTERM. Long-running workers that never dispose are the classic RxPHP memory leak.
• Unbounded accumulation. toArray(), ReplaySubject without a buffer size, and scan over an infinite stream all grow memory silently. Cap replay buffers.
• Exceptions inside operator closures become onError. A try/catch wrapped around subscribe will not see them; you must handle them through the stream or with a terminal error callback. This trips up everyone coming from imperative code.
• Operator names diverge slightly from RxJS. Aliases (select, where, doOnNext) exist alongside the canonical names, and a few RxJS operators are absent or named differently. Confirm against Observable.php rather than assuming. Also note that RxPHP 2.x changed subscribe to take an ObserverInterface; if a tutorial shows subscribe($onNext, $onError, $onCompleted), you are looking at 1.x — use CallbackObserver/subscribeCallback instead.
• PHP's native array functions beat RxPHP for small in-memory data. If the collection is already a finite array and you do not need cancellation or laziness, array_map/array_filter is faster and more familiar. Reach for RxPHP when the source is streaming, async, cancellable, or composed of multiple sources.
• Debugging: there is no devtools. Use tap-style side effects (doOnNext, doOnError, doOnCompleted) liberally, and add a DebugObserver-style pass-through if you need to log every notification. Marble tests in the repo's test suite are the best reference for expected timing.
• Reentrancy: emitting into a Subject from inside that subject's own subscriber recurses. The trampoline/current-thread scheduler exists to flatten this; do not bypass it with immediate scheduling in event-bus code.
• null values, empty vs never. Observable::empty() completes immediately; Observable::never() never emits or completes and will hang a blocking ->wait() or a test that does not use virtual time. Mixing these up produces deadlocks that look like logic bugs.