как продолжить другой процесс при прослушивании генераторов

Я использую Rethinkdb для прослушивания изменений, внесенных в таблицу. Я также использую Ratchet, чтобы получать уведомления в режиме реального времени о полученных изменениях.

Проблема сейчас в том, что rethinkdb возвращает / возвращает результат, но храповик не может запустить следующую строку для запуска сервера, поэтому клиенты не подключены к веб-сокету. Но сервер может получать обновления от rethinkdb. Просто это не может установить связь с клиентами.

Это связано с тем, что генератор используется для прослушивания изменений с помощью rethinkdb.

Вот как я слушаю изменения в rethinkdb:

function subscribeUpdates()
{
$updates = r\table("transactions")->changes()->run(self::getRethinkDB());
foreach ($updates as $up) {
yield $up['new_val'];
}

}

С другой стороны, так я запускаю сервер Ratchet, получая обновления от rethinkdb.

use Ratchet\Server\IoServer;
use Ratchet\Http\HttpServer;
use Ratchet\WebSocket\WsServer;
use MyApp\Chat;

require dirname(__DIR__) . '/doropu/vendor/autoload.php';


$server = IoServer::factory(

new HttpServer(
new WsServer(
new Chat()
)
),

8280,
'0.0.0.0'

);


$changes = db::subscribeUpdates();
foreach ($changes as $value) {
var_dump($value);
}

echo "It never comes here";
$server->run();

Обратите внимание, что я использовал yield $up['new_val']; в функции, которая прослушивает изменения таблицы rethinkdb. Делая это, Ratchet получает мгновенные обновления. Просто это никогда не переходит на следующую строку $server->run(); таким образом, клиент не может установить соединение.

Если бы я использовал return вместо yield, он переходит на следующую строку, но процесс прослушивания на этом заканчивается. Он больше не обращает внимания на изменения в rethinkdb.

Как запустить сервер Ratchet, а затем прослушать
изменения?