Event loop and RxPHP

Applying event loop to Observables would work in a similar way. We'd create Observables, start an event loop and periodically check their progress. Luckily for us, RxPHP is prepared for this. In combination with the ReactPHP library ( https://github.com/reactphp/react ), we can use a Scheduler that's designed exactly for what we need.

As an example, we can have a look at IntervalObservable that periodically emits values:

// rxphp_eventloop.php 
$loop = new ReactEventLoopStreamSelectLoop(); 
$scheduler = new RxSchedulerEventLoopScheduler($loop); 
 
RxObservable::interval(1000, $scheduler) 
    ->take(3) 
    ->subscribe(new DebugSubject()); 
 
$loop->run(); 

This prints three values with 1s delays:

$ php rxphp_eventloop.php
23:12:44 [] onNext: 0 (integer)
23:12:45 [] onNext: 1 (integer)
23:12:46 [] onNext: 2 (integer)
23:12:46 [] onCompleted

Note

In RxPHP 2, using event loops has been simplified and, most of the time, we don't even need to worry about starting the loop ourselves. We'll talk about differences between RxPHP 1.x and RxPHP 2 regarding event loops in Chapter 6, PHP Streams API and Higher-Order Observables.