Using rabbitmq or redis for pub/sub

[tolew1]

How can you implement this using rabbitmq as pub/sub so you can user multiple servers? Or does this library support socket-io-redis? https://github.com/socketio/socket.io-redis

I'd prefer using rabbitmq with this https://github.com/sensibill/socket.io-amqp but using amqp for PHP https://github.com/php-amqplib/php-amqplib

Any idea on how to convert those socket.io.amqp to PHP and utilize them with this library to use Rabbitmq as explained here https://github.com/rajaraodv/rabbitpubsub.

TIA

[walkor]

PHPSocket.io already support multiple servers by Channel Adapter and do not need redis or rabbitmq.

For example you have 3 servers server_A server_B server_C.
Server_A is a channel server and start script like this.

use Workerman\Worker;
require_once __DIR__ . '/vendor/autoload.php';

// =======ChannelAdapter server========
$channel = new Channel\Server('0.0.0.0', 3206);
Worker::runAll();

Server_B and Server_C is phpsocket.io server and codes like this.

use Workerman\Worker;
use PHPSocketIO\SocketIO;
require_once __DIR__ . '/vendor/autoload.php';

$io = new SocketIO(2020);
$io->on('workerStart', function()use($io){
    ChannelAdapter::$ip = 'ip_of_server_A';
    ChannelAdapter::$port = 3206;
    $io->adapter('\PHPSocketIO\ChannelAdapter');
});
$io->on('connection', function($socket)use($io){
    $socket->on('chat message', function($msg)use($io){
        $io->emit('chat message', $msg);
  });
});
Worker::runAll();

[walkor]

But of course if you want to use redis or rabbitmq in phpsocket.io here is the solution.

For redis implement, you can use react/redis to pub/sub in phpsocket.io.
First you need to install react/redis by command composer require clue/redis-react.
Then you can pub/sub like this.

<?php
require_once __DIR__ . '/vendor/autoload.php';
use Clue\React\Redis\Factory;
use Clue\React\Redis\Client;
use Workerman\Worker;
use PHPSocketIO\SocketIO;

// listen port 2021 for socket.io client
$io = new SocketIO(2021);
$io->on('workerStart', function() use ($io){
    // for react redis clinet
    global $factory;
    $loop    = Worker::getEventLoop();
    $factory = new Factory($loop);
    $factory->createClient('127.0.0.1:6379')->then(function (Client $client) use ($io) {
        $client->subscribe('some_event')->then(function () {
            echo 'Now subscribed to channel ' . PHP_EOL;
        });
        $client->on('message', function ($event_name, $message) use ($io){
           echo 'Message on ' . $event_name . ': ' . $message . PHP_EOL;
           $io->emit($event_name, $message);
        });
    });
});

$io->on('connection', function($socket)use($io){
  $socket->on('chat message', function($msg)use($io){
    $io->emit('chat message', $msg);
  });
  .......................
});

Worker::runAll();

And for rabbitmq.
composer require react/stomp

<?php
require_once __DIR__ . '/vendor/autoload.php';
use Workerman\Worker;
use PHPSocketIO\SocketIO;

// listen port 2021 for socket.io client
$io = new SocketIO(2021);
$io->on('workerStart', function() use ($io){
    // for react rabbitmq clinet
    global   $client;
    $loop    = Worker::getEventLoop();
    $factory = new React\Stomp\Factory($loop);
    $client  = $factory->createClient(array('vhost' => '/', 'login' => 'guest', 'passcode' => 'guest'));

    $client
        ->connect()
        ->then(function ($client) use ($loop) use ($io) {
            $client->subscribe('/topic/foo', function ($frame) use ($io) {
                echo "Message received: {$frame->body}\n";
                $io->emit(...........);
            });
        });
});

$io->on('connection', function($socket)use($io){
  $socket->on('chat message', function($msg)use($io){
    $io->emit('chat message', $msg);
  });
  ...............
});

Worker::runAll();

[tolew1]

That is interesting thank you for all the options. For instance I like the first option that is already built in as I don't have to do much more configuration.

However so I don't waste a server resource. Can server_A in your example also be a PHPSocketIO server too so that I can have 3 servers to load balance on and not just Server_B and Server_C?

Also option one presents a single point of failure. If server a goes down you lose your cluster.

[walkor]

Yes server_A can also be a PHPSocketIO server.

Yes there is a single point of failure , If you need high availability, you may have to use rabbitmq with out channleAdapter.

[tolew1]

Do you have an example of how to use rabbitmq without an adapter? I haven't been successful and it doesn't seem to be reaching the queue. This been really challenging just to get high availability. Not sure why this is so difficult. I would think more people would need this functionality.

Problem with the REACT is I need to create queues based on user connection. So the queue as to be created after the user has connected as each queue needs to be unique so I can use them as rooms.

[walkor]

Sorry, In addition to the examples above, I have no other examples using rabbitmq.

[tolew1]

Ok no problem however I'm trying your redis example above it never makes a connection. When I try to connect to with the client, the connection to the socketio. I'm getting 503 unavailable in the chrome dev tools on the chat demo example. I'm using your exact redis code above,

Nevermind on this, I had to change the port back to 2020 as your example used 2021.

[walkor]

Phpsocket.io never responses 503.

[tolew1]

Yes I found the problem it was the port number.

Last question, I'm sorry so many. How do I call the Redis client above inside the connection function

Something like this. Because I want to use redis commands when socket command is detected

$io->on('connection', function($socket)use($io){
  $socket->on('chat message', function($msg)use($io){
      $client->subscribe($msg')->then(function () {
            echo 'Now subscribed to channel ' . PHP_EOL;
        });
    $io->emit('chat message', $msg);
  });

[walkor]

<?php
require_once __DIR__ . '/vendor/autoload.php';
use Clue\React\Redis\Factory;
use Clue\React\Redis\Client;
use Workerman\Worker;
use PHPSocketIO\SocketIO;

// listen port 2021 for socket.io client
$io = new SocketIO(2021);
$io->on('workerStart', function() use ($io){
    $loop    = Worker::getEventLoop();
    $io->factory = new Factory($loop);
    $io->factory->createClient('127.0.0.1:6379')->then(function (Client $client) use ($io) {
        // ==== Assign to $io->redis ==== 
        $io->redis = $client;
        $io->redis->subscribe('some_event')->then(function () {/*...*/});
    });
});

$io->on('connection', function($socket)use($io){
  $socket->on('chat message', function($msg)use($io){
      $io->redis->subscribe($msg')->then(function () {
            echo 'Now subscribed to channel ' . PHP_EOL;
       });
      $io->emit('chat message', $msg);
  });

Worker::runAll();

You can store redis client to $io or global variable or other static members of the class.

[tolew1]

@walkor Thanks for the help and all the many questions. I was having a hard time understanding socket io in general but with your help I've gotten it working with regular rabbitmq and it's a great library for PHP.

Awesome work. I now have the ability to add as many servers as i need any not have a single point of failure.

[walkor]

You are welcome.

[RayHughes]

@tolew1 can you provide an example please. This is an awesome library and will likely grow tremendously.

[tolew1]

@RayHughes Sorry got busy. I can give high level overview as it's a lot of info.
Use PHPSocketio and let each server be it's own channel server (Needed for emitting from external php scripts). So don't connect other servers to it or then it becomes a SPOF. So if you have servers A, B, C all of them will be siloed channel servers. Then you can use a Redis cluster and/or MQ cluster that all 3 can talk to for pub/sub. You can store room info or whatever you want in redis. So you can then pass a broadcast through MQ and have all 3 servers listening to the queue. so whatever server the other user is connected to will get the broadcast through the use of MQ and the normal PHPSocketio emits. It can be a complicated setup but that's the gist. That way it's easy to scale horizontally also.

[igo9go]

$io = new SocketIO(2021);
$io->on('workerStart', function () use ($io) {
    $loop = Worker::getEventLoop();
    $io->factory = new Factory($loop);
    $io->factory->createClient('127.0.0.1:6379')->then(function (Client $client) use ($io) {
        $io->redis = $client;
        // ==== Assign to $io->redis ====
        $io->redis->set('test', 'test');

        $io->redis->subscribe('some_event')->then(function () {/*...*/
        });
    });
});

$io->on('connection', function ($socket) use ($io) {
    $io->redis->set('test2', 'test2');
    $socket->on('chat message', function ($msg) use ($io) {
        $io->redis->subscribe($msg)->then(function () {
            echo 'Now subscribed to channel ' . PHP_EOL;
       });
      $io->emit('chat message', $msg);
  });
});
Worker::runAll();

在connection里面，$io->redis->set('test2', 'test2'); 无法执行

[chengloy]

hi, walkor, I am so impressed by your professional skills and technology, I wanna ask about, can you details some demo about how rabbitmq pub/sub messages to PHPSocket.io, is there need write a service to connect them both?

[chengloy]

how can rabbitmq pub messages to phpsocket.io?