Как отправить сообщение с ZMQ точному абоненту?

Question

Я пытаюсь отправить клиенту уведомление с помощью Autobahn JS и ThruwayPHP.

Когда пользователь входит в систему, я вижу из консоли, что подписался на тему, и мой сервер веб-сокетов запущен и работает.

2018-01-25T12:36:19.2809790 debug      [Thruway\Transport\RatchetTransportProvider 9264] onMessage: ([32,1566442348733370,{},"ermp.confirmationform.1013542907194252"])
2018-01-25T12:36:19.2815330 debug      [Thruway\Subscription\SubscriptionGroup 9264] Added subscription to 'exact':'ermp.confirmationform.1013542907194252'
2018-01-25T12:36:19.2818770 debug      [Thruway\Transport\RatchetTransportProvider 9264] onMessage: ([32,501539158051958,{},"ermp.prescriptionform.1013542907194252"])
2018-01-25T12:36:19.2820810 debug      [Thruway\Subscription\SubscriptionGroup 9264] Added subscription to 'exact':'ermp.prescriptionform.1013542907194252'

Теперь, когда мой клиент редактирует свой профиль, я хотел бы отправить сообщение администратору, что пользователь X изменил свой профиль. Для этого я использую ZMQ, который прекрасно работал без частных каналов.

Вот мой WebsocketServer.php

class WebSocketServer extends Command
    {

        protected $signature   = 'websocket:init';
        protected $description = 'Start websocket';

        public function __construct()
        {
            parent::__construct();
        }

        public function handle()
        {
            $loop   = \React\EventLoop\Factory::create();
            $pusher = new Client("admin_realm", $loop);

            //on open
            $pusher->on('open', function (ClientSession $session) use ($loop, $pusher) {

                $context = new Context($loop);
                $pull    = $context->getSocket(\ZMQ::SOCKET_PULL);
                $pull->bind('tcp://127.0.0.1:9090');

                // 1) subscribe to a topic

                $onEvent = function ($data) {
                    $eventData = $data[0];
                    echo json_encode($eventData);
                };

                //subscribe to confirmationForm notis
                $session->subscribe('ermp.confirmationform', $onEvent);

                //subscribe to prescriptionForm notis
                $session->subscribe('ermp.prescriptionform', $onEvent);

                //on error
                $pull->on('error', function ($e) {
                    var_dump($e->getMessage());
                });

                //on message
                $pull->on('message', function ($msg) use ($session) {
                    $data = json_decode($msg, true);
                    //for testing purposes with a string
                    $session->publish('ermp.confirmationform', [$data]);

                    if (isset($data['status'])) {

                        if ($data['type'] === 'confirmationform') {
                            $session->publish('ermp.confirmationform', [$data]);

                        }
                        elseif ($data['type'] === 'prescriptionform') {
                            $session->publish('ermp.prescriptionform', [$data]);

                        }
                    }
                });

            });

            //start the router
            $router = new Router($loop);
            $router->registerModules([
                // Authentication
                new \Thruway\Authentication\AuthenticationManager(),
                // Authorization for a specific realm 'realm1'
                new \Thruway\Authentication\AuthorizationManager('realm1'),
            ]);
            $router->addInternalClient($pusher);
            $transportProvider = new RatchetTransportProvider("127.0.0.1", 8090);
            $router->addTransportProvider($transportProvider);
            //$router->setAuthorizationManager(new MyAuthorizationManager());
            $router->start();
        }
    }

И вот как я отправляю сообщение.

        $context = new \ZMQContext();
        $socket = $context->getSocket(\ZMQ::SOCKET_PUSH, 'ermp.confirmationform');
        $socket->connect("tcp://127.0.0.1:9090");

        $socket->send("hello");

Я вижу данные с dump() в моем коде сервера, но ничего не публикуется. Как я могу отправить сообщение соответствующему пользователю?