public static function runSharedWorker(SharedWorkerImplementationInterface $workerImpl, $socketAddress)
    {
        $server = self::startListening($socketAddress);
        try {
            $loop = self::getLoop();
            $loop->addReadStream($server, function () use ($loop, $server, $workerImpl) {
                $socket = stream_socket_accept($server);
                $peerName = stream_socket_get_name($socket, true);
                $connection = Source::fromStream($socket, true, null, false);
                $channel = self::getChannelFactory()->createChannel(new BufferedSource($connection), $connection);
                Selectable::registerRead($loop, $channel, function () use ($loop, $channel, $peerName, $workerImpl) {
                    try {
                        $message = $channel->receiveMessage();
                    } catch (IOException\UnderflowException $e) {
                        Selectable::unregisterRead($loop, $channel);
                        $workerImpl->onDisconnect($channel);

                        return;
                    }
                    if (AdminEncoding::isStopMessage($message, self::$adminCookie, $privileged)) {
                        if ($privileged) {
                            $lock = Lock::acquire();
                            self::stopListening();
                            $workerImpl->onStop();
                            $lock->release();
                        }
                    } elseif (AdminEncoding::isQueryMessage($message, self::$adminCookie, $privileged)) {
                        $result = $workerImpl->onQuery($privileged);
                        if (!($result instanceof WorkerStatus)) {
                            $result = new WorkerStatus($result);
                        }
                        AdminEncoding::sendStatusMessage($channel, $result);
                    } else {
                        $workerImpl->onMessage($message, $channel, $peerName);
                    }
                });
                $workerImpl->onConnect($channel, $peerName);
            });
            $workerImpl->setLoop($loop);
            $workerImpl->initialize();
            $loop->run();
            $workerImpl->terminate();
        } finally {
            self::stopListening();
        }
    }
public static function runSharedWorker(SharedWorkerImplementationInterface $workerImpl, $socketAddress)
    {
        $server = self::startListening($socketAddress);
        try {
            $loop = self::getLoop();
            $loop->addReadStream($server, function () use ($loop, $server, $workerImpl) {
                $socket = stream_socket_accept($server);
                $peerName = stream_socket_get_name($socket, true);
                $connection = Source::fromStream($socket, true, null, false);
                $channel = self::getChannelFactory()->createChannel(new BufferedSource($connection), $connection);
                Selectable::registerRead($loop, $channel, function () use ($loop, $channel, $peerName, $workerImpl) {
                    try {
                        $message = $channel->receiveMessage();
                    } catch (IOException\UnderflowException $e) {
                        Selectable::unregisterRead($loop, $channel);
                        $workerImpl->onDisconnect($channel);

                        return;
                    }
                    if (AdminEncoding::isStopMessage($message, self::$adminCookie, $privileged)) {
                        if ($privileged) {
                            $lock = Lock::acquire();
                            self::stopListening();
                            $workerImpl->onStop();
                            $lock->release();
                        }
                    } elseif (AdminEncoding::isQueryMessage($message, self::$adminCookie, $privileged)) {
                        $result = $workerImpl->onQuery($privileged);
                        if (!($result instanceof WorkerStatus)) {
                            $result = new WorkerStatus($result);
                        }
                        AdminEncoding::sendStatusMessage($channel, $result);
                    } else {
                        $workerImpl->onMessage($message, $channel, $peerName);
                    }
                });
                $workerImpl->onConnect($channel, $peerName);
            });
            $workerImpl->setLoop($loop);
            $workerImpl->initialize();
            $loop->run();
            $workerImpl->terminate();
        } finally {
            self::stopListening();
        }
    }
@param SharedWorkerImplementationInterface $workerImpl
@param string                              $socketAddress

@throws Exception\BindOrListenException
@throws Exception\RuntimeException