stream_set_blocking - PHP code example

The php Function Stream_Set_Blocking

The php function stream_set_blocking is part of the Stream class which is a generalized way to handle file, network and data compression operations. It is an asynchronous mechanism that makes it possible to start, pause and resume the execution flow of an isolated portion of your program, essentially creating a virtual thread that only lives within one PHP process. This is different from regular threads that are managed by the operating system and that live in multiple processes or even in a separate kernel thread.

This function has a single argument, mode, which can be either blocking or non-blocking. If mode is 0, the given stream will be switched to non-blocking mode, if mode is 1 it will be returned to blocking mode. This affects calls like fgets() and fread() which will return right away if the stream is readable, or block if it is not.

tv_sec and tv_usec are timeout parameters in microseconds. If they are set to 0 stream_select() will never block, instead it will poll the streams for data and return as soon as any of them change status (or if the call is interrupted by an incoming signal). If tv_sec or tv_usec is not set, a timeout of -1 seconds will be used which will mean that stream_select() will block indefinitely until an event on one of the watched streams occurs. This can be quite expensive for loops, so it is a good idea to use tv_sec and tv_usec together.

stream_set_blocking example

catch (RuntimeException $e) {
                    if (!$question->isHiddenFallback()) {
                        throw $e;
                    }
                }
            }

            if (false === $ret) {
                $isBlocked = stream_get_meta_data($inputStream)['blocked'] ?? true;

                if (!$isBlocked) {
                    stream_set_blocking($inputStream, true);
                }

                $ret = $this->readInput($inputStream, $question);

                if (!$isBlocked) {
                    stream_set_blocking($inputStream, false);
                }

                if (false === $ret) {
                    throw new MissingInputException('Aborted.');
                }

return;
        } finally {
            $this->info['pretransfer_time'] = $this->info['total_time'] = microtime(true) - $this->info['start_time'];
            restore_error_handler();
        }

        if (isset($context['ssl']['capture_peer_cert_chain']) && isset(($context = stream_context_get_options($this->context))['ssl']['peer_certificate_chain'])) {
            $this->info['peer_certificate_chain'] = $context['ssl']['peer_certificate_chain'];
        }

        stream_set_blocking($h, false);
        unset($this->context, $this->resolver);

        // Create dechunk buffers         if (isset($this->headers['content-length'])) {
            $this->remaining = (int) $this->headers['content-length'][0];
        } elseif ('chunked' === ($this->headers['transfer-encoding'][0] ?? null)) {
            stream_filter_append($this->buffer, 'dechunk', \STREAM_FILTER_WRITE);
            $this->remaining = -1;
        } else {
            $this->remaining = -2;
        }

/** * Unblocks streams. */
    protected function unblock(): void
    {
        if (!$this->blocked) {
            return;
        }

        foreach ($this->pipes as $pipe) {
            stream_set_blocking($pipe, 0);
        }
        if (\is_resource($this->input)) {
            stream_set_blocking($this->input, 0);
        }

        $this->blocked = false;
    }

    /** * Writes input to stdin. * * @throws InvalidArgumentException When an input iterator yields a non supported value */

$this->expectException(ClientException::class);
        $response->toStream();
    }

    public function testNonBlockingStream()
    {
        $client = $this->getHttpClient(__FUNCTION__);
        $response = $client->request('GET', 'http://localhost:8057/timeout-body');
        $stream = $response->toStream();
        usleep(10000);

        $this->assertTrue(stream_set_blocking($stream, false));
        $this->assertSame('<1>', fread($stream, 8192));
        $this->assertFalse(feof($stream));

        $this->assertTrue(stream_set_blocking($stream, true));
        $this->assertSame('<2>', fread($stream, 8192));
        $this->assertSame('', fread($stream, 8192));
        $this->assertTrue(feof($stream));
    }

    public function testSeekAsyncStream()
    {

  public function testFileFunctions() {
    $filename = 'public://' . $this->randomMachineName();
    file_put_contents($filename, str_repeat('d', 1000));

    // Open for rw and place pointer at beginning of file so select will return.     $handle = fopen($filename, 'c+');
    $this->assertNotFalse($handle, 'Able to open a file for appending, reading and writing.');

    // Attempt to change options on the file stream: should all fail.     $this->assertFalse(@stream_set_blocking($handle, 0), 'Unable to set to non blocking using a local stream wrapper.');
    $this->assertFalse(@stream_set_blocking($handle, 1), 'Unable to set to blocking using a local stream wrapper.');
    $this->assertFalse(@stream_set_timeout($handle, 1), 'Unable to set read time out using a local stream wrapper.');
    $this->assertEquals(-1 /*EOF*/, @stream_set_write_buffer($handle, 512), 'Unable to set write buffer using a local stream wrapper.');

    // This will test stream_cast().     $read = [$handle];
    $write = NULL;
    $except = NULL;
    $this->assertEquals(1, stream_select($read, $write, $except, 0), 'Able to cast a stream via stream_select.');

    // This will test stream_truncate().

$timeout = $this->getTimeout();
        set_error_handler(function D$type, $msg) {
            throw new TransportException(sprintf('Connection could not be established with host "%s": ', $this->url).$msg);
        });
        try {
            $this->stream = stream_socket_client($this->url, $errno, $errstr, $timeout, \STREAM_CLIENT_CONNECT, $streamContext);
        } finally {
            restore_error_handler();
        }

        stream_set_blocking($this->stream, true);
        stream_set_timeout($this->stream, $timeout);
        $this->in = &$this->stream;
        $this->out = &$this->stream;
    }

    public function startTLS(): bool
    {
        set_error_handler(function D$type, $msg) {
            throw new TransportException('Unable to connect with STARTTLS: '.$msg);
        });
        try {

if ( ! $stream ) {
            $this->errors->add(
                'command',
                sprintf(
                    /* translators: %s: Command. */
                    __( 'Unable to perform command: %s' ),
                    $command
                )
            );
        } else {
            stream_set_blocking( $stream, true );
            stream_set_timeout( $stream, FS_TIMEOUT );
            $data = stream_get_contents( $stream );
            fclose( $stream );

            if ( $returnbool ) {
                return ( false === $data ) ? false : '' !== trim( $data );
            } else {
                return $data;
            }
        }

define('ERR_SELECT_FAILED', 1);
define('ERR_TIMEOUT', 2);
define('ERR_READ_FAILED', 3);
define('ERR_WRITE_FAILED', 4);

$read = [\STDIN];
$write = [\STDOUT, \STDERR];

stream_set_blocking(\STDIN, 0);
stream_set_blocking(\STDOUT, 0);
stream_set_blocking(\STDERR, 0);

$out = $err = '';
while ($read || $write) {
    $r = $read;
    $w = $write;
    $e = null;
    $n = stream_select($r, $w, $e, 5);

    if (false === $n) {

return fread($this->handle, $count);
    }

    public function stream_eof(): bool
    {
        return feof($this->handle);
    }

    public function stream_set_option(int $option, int $arg1, int $arg2): bool
    {
        return match ($option) {
            \STREAM_OPTION_BLOCKING => stream_set_blocking($this->handle, $arg1),
            \STREAM_OPTION_READ_TIMEOUT => stream_set_timeout($this->handle, $arg1, $arg2),
            \STREAM_OPTION_WRITE_BUFFER => 0 === stream_set_write_buffer($this->handle, $arg2),
            default => false,
        };
    }

    public function stream_stat(): array|false
    {
        if (!$stat = stat($this->path)) {
            return false;
        }

        $headers .= "\r\n";

        if ( ! is_null( $parsed_args['body'] ) ) {
            $headers .= $parsed_args['body'];
        }

        fwrite( $handle, $headers );

        if ( ! $parsed_args['blocking'] ) {
            stream_set_blocking( $handle, 0 );
            fclose( $handle );
            return array(
                'headers'  => array(),
                'body'     => '',
                'response' => array(
                    'code'    => false,
                    'message' => false,
                ),
                'cookies'  => array(),
            );
        }

    public function initialize(): void
    {
        $descriptorSpec = [
            0 => ['pipe', 'r'],
            1 => ['pipe', 'w'],
            2 => ['pipe', '\\' === \DIRECTORY_SEPARATOR ? 'a' : 'w'],
        ];
        $pipes = [];
        $this->stream = proc_open($this->command, $descriptorSpec, $pipes);
        stream_set_blocking($pipes[2], false);
        if ($err = stream_get_contents($pipes[2])) {
            throw new TransportException('Process could not be started: '.$err);
        }
        $this->in = &$pipes[0];
        $this->out = &$pipes[1];
    }

    public function terminate(): void
    {
        if (null !== $this->stream) {
            fclose($this->in);

return new VarDumperFormatter();
    }

    /** * @return resource */
    private function createSocket()
    {
        $socket = stream_socket_client($this->host, $errno, $errstr, 0, \STREAM_CLIENT_CONNECT | \STREAM_CLIENT_ASYNC_CONNECT | \STREAM_CLIENT_PERSISTENT, $this->context);

        if ($socket) {
            stream_set_blocking($socket, false);
        }

        return $socket;
    }

    private function formatRecord(array|LogRecord $record): string
    {
        $recordFormatted = $record['formatted'];

        foreach (['log_uuid', 'uuid', 'uid'] as $key) {
            if (isset($record['extra'][$key])) {