Слушатели и таймеры в ReactPHP?

Я пытаюсь создать небольшое приложение, которое будет делать несколько вещей.

Основной класс будет порождать, уничтожать, слушать и управлять рабочими …

<?php

namespace Queue;

class Controller extends Process
{
public $workers = array();
function __construct()
{
//Run the Process constructor before doing anything else.
parent::__construct();
}
function spawnWorker($businessId)
{
$this->workers[$businessId] = new \React\ChildProcess\Process("php index.php worker $businessId");
$this->workers[$businessId]->start($this->loop);
$this->workers[$businessId]->stdout->on("data", function($output) use(&$businessId)  {
$this->channels->out->write("Child $businessId Said: " . $output);
});
}
function outputToWorker($businessId, $string)
{
$this->workers[$businessId]->stdin->write($string);
}
function run()
{
$this->loop->run();
}
}

Группа рабочих, управляемых контроллером (в конечном итоге у них будет длительный процесс, который управляет очередью, определенной в базе данных) …

<?php

namespace Queue;

class Worker extends Process
{
function __construct($businessId)
{
//Run the Process constructor before doing anything else.
parent::__construct();
// Assign a business to this worker.
$this->businessId = $businessId;
// Create a queue object for this worker.
$this->queue = new \stdClass;
}
function run()
{
$this->channels->in->on("data", function($input) {
$this->sendOutput($input);
});
$this->loop->run();
}
}

Однако я столкнулся с ситуацией, когда код не сделал то, что я ожидал …

// Based on command line argument, decide which kind of process to spawn.
if (!isset($argv[1]) || $argv[1] === "controller")
{
$controller = new Controller;
$controller->spawnWorker(1);
$controller->outputToWorker(1, "Before Run");
$controller->run();
$controller->outputToWorker(1, "After Run");
} else if (strtolower($argv[1]) === "worker" && isset($argv[2]))
{
$worker = new Worker($argv[1]);
$worker->run();
}

Этот последний фрагмент кода является файлом, который я запускаю для запуска приложения. Он маршрутизирует, какой тип процесса должен быть создан на основе аргументов командной строки. (Таким образом, мне не нужно иметь файл для разных процессов).

Рабочий корректно появляется и первое сообщение (Child 1 Said: Before Run) отправляется работнику, работник отправляет его stdout и controller слышит это событие и отправляет его на свой stdout,

Однако кажется, что прослушиватель является одноразовым, и цикл обработки событий не обрабатывает его повторно. Смысл в любое время после первоначального run() на controller он не будет отвечать на stdout события от рабочего.

Должен ли я создавать своих собственных явных слушателей с таймерами?