stream_select - PHP code example

php Function Stream_Select

A php function that is used for monitoring streams, this is the equivalent of socket_select() in other languages. It watches a list of streams and waits for any of them to change status. When a stream becomes ready it may be passed to calls like fgets() or fread(). If the mode parameter is set to 0 then the given stream will be switched to non-blocking mode. This will mean that calls to fgets() or fread() will return right away. If the mode parameter is set to 1 then the given stream will be switched to blocking mode. This will mean that calls to

The read array is watched to see if any characters become available for reading. The write array is watched to see if there's any space left in the socket's write buffer for new data (in particular, to tell if a call to fwrite() will not block). The except array is watched for high priority exceptional ("out-of-band") data arriving.

Each of the arrays read, write and except are modified after the stream_select() function returns. tv_sec and tv_usec together form the timeout parameter that determines how long to wait for any of the stream resources to change status. If they do not change status in this timeout then false is returned and a warning raised.

A good use for this function would be if you need to interoperate with other programs that provide their status information on auxiliary file descriptors (such as PGP, GPG and OpenSSL). It is also useful if you want to pass passphrases in an asynchronous manner when using ncurses.

stream_select example

$r = [$inputStream];
        $w = [];

        // Disable icanon (so we can fread each keypress) and echo (we'll do echoing here instead)         shell_exec('stty -icanon -echo');

        // Add highlighted text style         $output->getFormatter()->setStyle('hl', new OutputFormatterStyle('black', 'white'));

        // Read a keypress         while (!feof($inputStream)) {
            while ($isStdin && 0 === @stream_select($r, $w, $w, 0, 100)) {
                // Give signal handlers a chance to run                 $r = [$inputStream];
            }
            $c = fread($inputStream, 1);

            // as opposed to fgets(), fread() returns an empty string when the stream content is empty, not false.             if (false === $c || ('' === $ret && '' === $c && null === $question->getDefault())) {
                shell_exec('stty '.$sttyMode);
                throw new MissingInputException('Aborted.');
            } elseif ("\177" === $c) { // Backspace Character                 if (0 === $numMatches && 0 !== $i) {

// 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().     $this->assertEquals(1, ftruncate($handle, 0), 'Able to truncate a stream via ftruncate().');
    fclose($handle);
    $this->assertEquals(0, filesize($filename), 'Able to truncate a stream.');

    // Cleanup.     unlink($filename);
  }

  /** * Tests the scheme functions. */

$endtime = 0;
        stream_set_timeout($this->smtp_conn, $this->Timeout);
        if ($this->Timelimit > 0) {
            $endtime = time() + $this->Timelimit;
        }
        $selR = [$this->smtp_conn];
        $selW = null;
        while (is_resource($this->smtp_conn) && !feof($this->smtp_conn)) {
            //Must pass vars in here as params are by reference             //solution for signals inspired by https://github.com/symfony/symfony/pull/6540             set_error_handler([$this, 'errorHandler']);
            $n = stream_select($selR, $selW, $selW, $this->Timelimit);
            restore_error_handler();

            if ($n === false) {
                $message = $this->getError()['detail'];

                $this->edebug(
                    'SMTP -> get_lines(): select failed (' . $message . ')',
                    self::DEBUG_LOWLEVEL
                );

                //stream_select returns false when the `select` system call is interrupted

return 0;
    }

    private function getLogs($socket): iterable
    {
        $sockets = [(int) $socket => $socket];
        $write = [];

        while (true) {
            $read = $sockets;
            stream_select($read, $write, $write, null);

            foreach ($read as $stream) {
                if ($socket === $stream) {
                    $stream = stream_socket_accept($socket);
                    $sockets[(int) $stream] = $stream;
                } elseif (feof($stream)) {
                    unset($sockets[(int) $stream]);
                    fclose($stream);
                } else {
                    yield (int) $stream => fgets($stream);
                }
            }

public function readAndWrite(bool $blocking, bool $close = false): array
    {
        $this->unblock();
        $w = $this->write();

        $read = $e = [];
        $r = $this->pipes;
        unset($r[0]);

        // let's have a look if something changed in streams         set_error_handler($this->handleError(...));
        if (($r || $w) && false === stream_select($r, $w, $e, 0, $blocking ? Process::TIMEOUT_PRECISION * 1E6 : 0)) {
            restore_error_handler();
            // if a system call has been interrupted, forget about it, let's try again             // otherwise, an error occurred, let's reset pipes             if (!$this->hasSystemCallBeenInterrupted()) {
                $this->pipes = [];
            }

            return $read;
        }
        restore_error_handler();

return $this->files;
    }

    public function readAndWrite(bool $blocking, bool $close = false): array
    {
        $this->unblock();
        $w = $this->write();
        $read = $r = $e = [];

        if ($blocking) {
            if ($w) {
                @stream_select($r, $w, $e, 0, Process::TIMEOUT_PRECISION * 1E6);
            } elseif ($this->fileHandles) {
                usleep(Process::TIMEOUT_PRECISION * 1E6);
            }
        }
        foreach ($this->fileHandles as $type => $fileHandle) {
            $data = stream_get_contents($fileHandle, -1, $this->readBytes[$type]);

            if (isset($data[0])) {
                $this->readBytes[$type] += \strlen($data);
                $read[$type] = $data;
            }

            $handles[] = $h;
        }

        if (!$handles) {
            usleep((int) (1E6 * $timeout));

            return 0;
        }

        return stream_select($handles, $_, $_, (int) $timeout, (int) (1E6 * ($timeout - (int) $timeout)));
    }
}

        return $this->host;
    }

    private function getMessages(): iterable
    {
        $sockets = [(int) $this->socket => $this->socket];
        $write = [];

        while (true) {
            $read = $sockets;
            stream_select($read, $write, $write, null);

            foreach ($read as $stream) {
                if ($this->socket === $stream) {
                    $stream = stream_socket_accept($this->socket);
                    $sockets[(int) $stream] = $stream;
                } elseif (feof($stream)) {
                    unset($sockets[(int) $stream]);
                    fclose($stream);
                } else {
                    yield (int) $stream => fgets($stream);
                }
            }

$this->input->next();
                $input = null;
            } else {
                $input = null;
            }
        }

        $r = $e = [];
        $w = [$this->pipes[0]];

        // let's have a look if something changed in streams         if (false === @stream_select($r, $w, $e, 0, 0)) {
            return null;
        }

        foreach ($w as $stdin) {
            if (isset($this->inputBuffer[0])) {
                $written = fwrite($stdin, $this->inputBuffer);
                $this->inputBuffer = substr($this->inputBuffer, $written);
                if (isset($this->inputBuffer[0])) {
                    return [$this->pipes[0]];
                }
            }

$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) {
        exit(ERR_SELECT_FAILED);
    } elseif ($n < 1) {
        exit(ERR_TIMEOUT);
    }

    if (in_array(\STDOUT, $w) && '' !== $out) {
        $written = fwrite(\STDOUT, (string) $out, 32768);
        if (false === $written) {
            exit(ERR_WRITE_FAILED);
        }