public function onWorkerReplyMessage(string $message)
    {
        try {
            $data = json_decode($message, true);
            if (empty($data) || !isset($data['request'])) {
                throw new ServerException('Invalid worker request received.');
            }
            $result = null;
            switch ($data['request']) {
                case 'register_job':
                    $result = $this->registerJob($data['job_name'], $data['worker_id']);
                    break;
                case 'unregister_job':
                    $result = $this->unregisterJob($data['job_name'], $data['worker_id']);
                    break;
                case 'change_state':
                    $result = $this->changeWorkerState($data['worker_id'], $data['state']);
                    break;
                case 'job_completed':
                    $this->onJobCompleted($data['worker_id'], $data['job_id'], $data['result']);
                    break;
            }
            $response = ($result === true) ? 'ok' : 'error';
            $this->workerReplySocket->send($response);
        } catch (ServerException $e) {
            $errorMsgPattern = 'Worker message error: %s (%s:%d)';
            $this->logger->error(sprintf($errorMsgPattern, $e->getMessage(), $e->getFile(), $e->getLine()));
        }
    }
public function onWorkerReplyMessage(string $message)
    {
        try {
            $data = json_decode($message, true);
            if (empty($data) || !isset($data['request'])) {
                throw new ServerException('Invalid worker request received.');
            }
            $result = null;
            switch ($data['request']) {
                case 'register_job':
                    $result = $this->registerJob($data['job_name'], $data['worker_id']);
                    break;
                case 'unregister_job':
                    $result = $this->unregisterJob($data['job_name'], $data['worker_id']);
                    break;
                case 'change_state':
                    $result = $this->changeWorkerState($data['worker_id'], $data['state']);
                    break;
                case 'job_completed':
                    $this->onJobCompleted($data['worker_id'], $data['job_id'], $data['result']);
                    break;
            }
            $response = ($result === true) ? 'ok' : 'error';
            $this->workerReplySocket->send($response);
        } catch (ServerException $e) {
            $errorMsgPattern = 'Worker message error: %s (%s:%d)';
            $this->logger->error(sprintf($errorMsgPattern, $e->getMessage(), $e->getFile(), $e->getLine()));
        }
    }
public function onWorkerReplyMessage ( string $ message ) { try { $ data = json_decode ( $ message , true ) ; if ( empty ( $ data ) || ! isset ( $ data [ 'request' ] ) ) { throw new ServerException ( 'Invalid worker request received.' ) ; } $ result = null ; switch ( $ data [ 'request' ] ) { case 'register_job' : $ result = $ this -> registerJob ( $ data [ 'job_name' ] , $ data [ 'worker_id' ] ) ; break ; case 'unregister_job' : $ result = $ this -> unregisterJob ( $ data [ 'job_name' ] , $ data [ 'worker_id' ] ) ; break ; case 'change_state' : $ result = $ this -> changeWorkerState ( $ data [ 'worker_id' ] , $ data [ 'state' ] ) ; break ; case 'job_completed' : $ this -> onJobCompleted ( $ data [ 'worker_id' ] , $ data [ 'job_id' ] , $ data [ 'result' ] ) ; break ; } $ response = ( $ result === true ) ? 'ok' : 'error' ; $ this -> workerReplySocket -> send ( $ response ) ; } catch ( ServerException $ e ) { $ errorMsgPattern = 'Worker message error: %s (%s:%d)' ; $ this -> logger -> error ( sprintf ( $ errorMsgPattern , $ e -> getMessage ( ) , $ e -> getFile ( ) , $ e -> getLine ( ) ) ) ; } }
Handles incoming worker requests.

@param string $message Json-encoded data received from worker.
@throws ServerException
Handles incoming worker requests .