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