protected function readNextProcessValue(bool $valueShouldBeLogged = true)
    {
        $readTimeout = $this->options['read_timeout'];
        $payload = '';

        try {
            $startTimestamp = microtime(true);

            do {
                $this->client->selectRead($readTimeout);
                $packet = $this->client->read(static::SOCKET_PACKET_SIZE);

                $chunksLeft = (int) substr($packet, 0, static::SOCKET_HEADER_SIZE);
                $chunk = substr($packet, static::SOCKET_HEADER_SIZE);

                $payload .= $chunk;

                if ($chunksLeft > 0) {
                    // The next chunk might be an empty string if don't wait a short period on slow environments.
                    usleep(self::SOCKET_NEXT_CHUNK_DELAY * 1000);
                }
            } while ($chunksLeft > 0);
        } catch (SocketException $exception) {
            $this->waitForProcessTermination();
            $this->checkProcessStatus();

            // Extract the socket error code to throw more specific exceptions
            preg_match('/\(([A-Z_]+?)\)$/', $exception->getMessage(), $socketErrorMatches);
            $socketErrorCode = constant($socketErrorMatches[1]);

            $elapsedTime = microtime(true) - $startTimestamp;
            if ($socketErrorCode === SOCKET_EAGAIN && $readTimeout !== null && $elapsedTime >= $readTimeout) {
                throw new Exceptions\ReadSocketTimeoutException($readTimeout, $exception);
            }

            throw $exception;
        }

        $this->logProcessStandardStreams();

        ['logs' => $logs, 'value' => $value] = json_decode(base64_decode($payload), true);

        foreach ($logs ?: [] as $log) {
            $level = (new \ReflectionClass(LogLevel::class))->getConstant($log['level']);
            $messageContainsLineBreaks = strstr($log['message'], PHP_EOL) !== false;
            $formattedMessage = $messageContainsLineBreaks ? "\n{log}\n" : '{log}';

            $this->logger->log($level, "Received a $log[origin] log: $formattedMessage", [
                'pid' => $this->processPid,
                'port' => $this->serverPort(),
                'log' => $log['message'],
            ]);
        }

        $value = $this->unserialize($value);

        if ($valueShouldBeLogged) {
            $this->logger->debug('Received data from the port {port}...', [
                'pid' => $this->processPid,
                'port' => $this->serverPort(),
                'data' => $value,
            ]);
        }

        if ($value instanceof NodeException) {
            throw $value;
        }

        return $value;
    }
protected function readNextProcessValue(bool $valueShouldBeLogged = true)
    {
        $readTimeout = $this->options['read_timeout'];
        $payload = '';

        try {
            $startTimestamp = microtime(true);

            do {
                $this->client->selectRead($readTimeout);
                $packet = $this->client->read(static::SOCKET_PACKET_SIZE);

                $chunksLeft = (int) substr($packet, 0, static::SOCKET_HEADER_SIZE);
                $chunk = substr($packet, static::SOCKET_HEADER_SIZE);

                $payload .= $chunk;

                if ($chunksLeft > 0) {
                    // The next chunk might be an empty string if don't wait a short period on slow environments.
                    usleep(self::SOCKET_NEXT_CHUNK_DELAY * 1000);
                }
            } while ($chunksLeft > 0);
        } catch (SocketException $exception) {
            $this->waitForProcessTermination();
            $this->checkProcessStatus();

            // Extract the socket error code to throw more specific exceptions
            preg_match('/\(([A-Z_]+?)\)$/', $exception->getMessage(), $socketErrorMatches);
            $socketErrorCode = constant($socketErrorMatches[1]);

            $elapsedTime = microtime(true) - $startTimestamp;
            if ($socketErrorCode === SOCKET_EAGAIN && $readTimeout !== null && $elapsedTime >= $readTimeout) {
                throw new Exceptions\ReadSocketTimeoutException($readTimeout, $exception);
            }

            throw $exception;
        }

        $this->logProcessStandardStreams();

        ['logs' => $logs, 'value' => $value] = json_decode(base64_decode($payload), true);

        foreach ($logs ?: [] as $log) {
            $level = (new \ReflectionClass(LogLevel::class))->getConstant($log['level']);
            $messageContainsLineBreaks = strstr($log['message'], PHP_EOL) !== false;
            $formattedMessage = $messageContainsLineBreaks ? "\n{log}\n" : '{log}';

            $this->logger->log($level, "Received a $log[origin] log: $formattedMessage", [
                'pid' => $this->processPid,
                'port' => $this->serverPort(),
                'log' => $log['message'],
            ]);
        }

        $value = $this->unserialize($value);

        if ($valueShouldBeLogged) {
            $this->logger->debug('Received data from the port {port}...', [
                'pid' => $this->processPid,
                'port' => $this->serverPort(),
                'data' => $value,
            ]);
        }

        if ($value instanceof NodeException) {
            throw $value;
        }

        return $value;
    }
protected function readNextProcessValue ( bool $ valueShouldBeLogged = true ) { $ readTimeout = $ this -> options [ 'read_timeout' ] ; $ payload = '' ; try { $ startTimestamp = microtime ( true ) ; do { $ this -> client -> selectRead ( $ readTimeout ) ; $ packet = $ this -> client -> read ( static :: SOCKET_PACKET_SIZE ) ; $ chunksLeft = ( int ) substr ( $ packet , 0 , static :: SOCKET_HEADER_SIZE ) ; $ chunk = substr ( $ packet , static :: SOCKET_HEADER_SIZE ) ; $ payload .= $ chunk ; if ( $ chunksLeft > 0 ) { // The next chunk might be an empty string if don't wait a short period on slow environments. usleep ( self :: SOCKET_NEXT_CHUNK_DELAY * 1000 ) ; } } while ( $ chunksLeft > 0 ) ; } catch ( SocketException $ exception ) { $ this -> waitForProcessTermination ( ) ; $ this -> checkProcessStatus ( ) ; // Extract the socket error code to throw more specific exceptions preg_match ( '/\(([A-Z_]+?)\)$/' , $ exception -> getMessage ( ) , $ socketErrorMatches ) ; $ socketErrorCode = constant ( $ socketErrorMatches [ 1 ] ) ; $ elapsedTime = microtime ( true ) - $ startTimestamp ; if ( $ socketErrorCode === SOCKET_EAGAIN && $ readTimeout !== null && $ elapsedTime >= $ readTimeout ) { throw new Exceptions \ ReadSocketTimeoutException ( $ readTimeout , $ exception ) ; } throw $ exception ; } $ this -> logProcessStandardStreams ( ) ; [ 'logs' => $ logs , 'value' => $ value ] = json_decode ( base64_decode ( $ payload ) , true ) ; foreach ( $ logs ? : [ ] as $ log ) { $ level = ( new \ ReflectionClass ( LogLevel :: class ) ) -> getConstant ( $ log [ 'level' ] ) ; $ messageContainsLineBreaks = strstr ( $ log [ 'message' ] , PHP_EOL ) !== false ; $ formattedMessage = $ messageContainsLineBreaks ? "\n{log}\n" : '{log}' ; $ this -> logger -> log ( $ level , "Received a $log[origin] log: $formattedMessage" , [ 'pid' => $ this -> processPid , 'port' => $ this -> serverPort ( ) , 'log' => $ log [ 'message' ] , ] ) ; } $ value = $ this -> unserialize ( $ value ) ; if ( $ valueShouldBeLogged ) { $ this -> logger -> debug ( 'Received data from the port {port}...' , [ 'pid' => $ this -> processPid , 'port' => $ this -> serverPort ( ) , 'data' => $ value , ] ) ; } if ( $ value instanceof NodeException ) { throw $ value ; } return $ value ; }
Read the next value written by the process.

@throws \Nesk\Rialto\Exceptions\ReadSocketTimeoutException if reading the socket exceeded the timeout.
@throws \Nesk\Rialto\Exceptions\Node\Exception if the process returned an error.
Read the next value written by the process .