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...