многопоточность — использование реактивного PHP в блокирующем приложении

В настоящее время я работаю над приложением PHP, которое будет использовать некоторые веб-сокеты для связи с другим сервисом.

Чтобы поговорить с этим сервисом веб-сокетов, мы используем Ratchet — библиотеку PHP, основанную на реагирующем PHP.

Этот фрагмент кода должен отправлять и отвечать на пару запросов, а после этого должен возвращать информацию в «основной поток».

Пример потока:

HTTP-запрос -> контроллер -> запускает службу, которая открывает клиент веб-сокета -> клиент веб-сокета общается с сервером -> как только это будет сделано, он должен вернуть результат в код контроллера -> выходные данные контроллера пользователю

У меня проблема в том, что я не знаком с Reactive PHP и не знаю, как с этим справиться.

Я пробовал;

    $service = new WebsocketService();
$startTimer = time();
$service->getList(44);
while($service->getResponse() == null) {
usleep(500);
if (time() > $startTimer + 10) {
continue; //Timeout on 10 seconds
}
}
var_dump($service->getResponse());

Сервисный код установит свою переменную «response» в значение, отличное от нуля, как только это будет сделано. Это, очевидно, терпит неудачу, потому что метод сна блокирует поток. Кроме того, кажется, что цикл while блокирует ввод-вывод, а реактивный код дает сбой.

Решением было бы открыть новый поток и запустить там код веб-сокета, но я не был бы доволен этим.

Я чувствую, что мне нужно реализовать своего рода «наблюдатель» вокруг процесса websocket, но я не уверен, как это сделать.

Наш клиентский код службы Websocket выглядит следующим образом;

private $response = null;

/**
* @return null|object
*/
public function getResponse() {
return $this->response;
}

public function getList($accountId) {
$this->response = null;
\Ratchet\Client\connect('ws://192.168.56.1:8080')->then(function(\Ratchet\Client\WebSocket $conn) use ($accountId) {
$login = new \stdClass();
$login->action = 'login';
$conn->on('message', function($msg) use ($conn, $login, $accountId) {
try {
$response = json_decode($msg);
if ($response->result_id == 100) {
//Succesfully logged in to websocket server

//Do our request now.
$message = new \stdClass();
$message->target = 'test';
$conn->send(json_encode($message));
}

if (isset($response->reply) && $response->reply == 'list') {
$this->response = $response; //This is the content I need returned in the controller
$conn->close(); //Dont need it anymore
}

} catch (\Exception $e) {
echo 'response exception!';
//Do nothing for now
}
});

$conn->send(json_encode($login));
}, function ($e) {
echo "Could not connect: {$e->getMessage()}\n";
});
}

Запуск такого кода также не работает;

    $service = new WebsocketService();
$service->getList(44);
echo 'Test';
var_dump($service->getResponse());

потому что «тестовое» эхо приходит еще до того, как я получу ответ от сервера веб-сокетов.

Пожалуйста, просветите меня! Я не уверен, что искать.