Internal/WorkerRunner.php
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();
        }
    }
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
@param SharedWorkerImplementationInterface $workerImpl @param string $socketAddress