I am trying to achieve mechanism where PHP pushes message to RabbitMQ (I don't want RabbitMQ to be directly exposed to user), RabbitMQ connects to RatchetPHP and Ratchet broadcasts it via websocket connections to users.
The issue I have is with accually making Ratchet server to simultanously listen for queue messages and transfer them further. Ratchet documentation assumes using ZeroMQ and after a long search through outdated documentations and libraries which do not have such methods anymore (eg. React\Stomp
) I need fresh eyes from someone who has experience with these solutions.
What I have is pusher.php
(standard example from RabbitMQ docs):
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$channel = $connection->channel();
$channel->queue_declare('hello', false, false, false, false);
$msg = new AMQPMessage('Hello World!');
$channel->basic_publish($msg, '', 'hello');
echo " [x] Sent 'Hello World!'
";
$channel->close();
$connection->close();
Just to simplify reproducing scenario I include also Chat
class:
use Ratchet\ConnectionInterface;
use Ratchet\MessageComponentInterface;
class Chat implements MessageComponentInterface
{
protected $clients;
public function __construct()
{
$this->clients = new \SplObjectStorage;
}
public function onOpen(ConnectionInterface $connection)
{
// Store the new connection to send messages to later
$this->clients->attach($connection);
echo "New connection! ({$connection->resourceId})
";
}
public function onMessage(ConnectionInterface $from, $msg)
{
$numRecv = count($this->clients) - 1;
echo sprintf('Connection %d sending message "%s" to %d other connection%s'."
"
, $from->resourceId, $msg, $numRecv, $numRecv == 1 ? '' : 's');
foreach($this->clients as $client)
{
/** @var \SplObjectStorage $client */
if($from !== $client)
{
// The sender is not the receiver, send to each client connected
$client->send($msg);
}
}
}
public function onClose(ConnectionInterface $conn)
{
// The connection is closed, remove it, as we can no longer send it messages
$this->clients->detach($conn);
echo "Connection {$conn->resourceId} has disconnected
";
}
public function onError(ConnectionInterface $conn, \Exception $e)
{
echo "An error has occurred: {$e->getMessage()}
";
$conn->close();
}
}
And Ratchet server.php
(standard Ratchet example and RabbitMQ receiver example):
use PhpAmqpLib\Connection\AMQPStreamConnection;
use Src\Chat;
// RABBIT_RECEIVER
$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$channel = $connection->channel();
$channel->queue_declare('hello', false, false, false, false);
echo ' [*] Waiting for messages. To exit press CTRL+C', "
";
$callback = function ($msg)
{
echo " [x] Received ", $msg->body, "
";
};
$channel->basic_consume('hello', '', false, true, false, false, $callback);
while(count($channel->callbacks))
{
$channel->wait();
}
$channel->close();
$connection->close();
// RABBIT_RECEIVER END
$server = new \Ratchet\App('sockets.dev');
$server->route('/', new Chat());
$server->run();
Current versions are basically 2 separate mechanisms listening for messages and they work great alone (so no issue there) except that they block each other and do not transfer messages between.
Question is how to make server.php
to make RabbitMQ receive message and plug it into running Ratchet server.