DuplexResourceStream::pause() - Code Metrics - Inspection of "Add new config options: {{what}} and {{where}}" - deployphp/deployer - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 5cefd1...492078 )

by Anton

created 2024-10-22 06:54 UTC

DuplexResourceStream::pause() A

↳ Parent: DuplexResourceStream

Complexity

Conditions	2
Paths	2

Size

Total Lines	5
Code Lines	3

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
eloc	3
nc	2
nop	0
dl	0
loc	5
c	1
b	0
f	0
cc	2
rs	10

<?php

namespace React\Stream;

use Evenement\EventEmitter;
use React\EventLoop\Loop;
use React\EventLoop\LoopInterface;
use InvalidArgumentException;

final class DuplexResourceStream extends EventEmitter implements DuplexStreamInterface
{
    private $stream;

    /** @var LoopInterface */
    private $loop;

    /**
     * Controls the maximum buffer size in bytes to read at once from the stream.
     *
     * This can be a positive number which means that up to X bytes will be read
     * at once from the underlying stream resource. Note that the actual number
     * of bytes read may be lower if the stream resource has less than X bytes
     * currently available.
     *
     * This can be `-1` which means read everything available from the
     * underlying stream resource.
     * This should read until the stream resource is not readable anymore
     * (i.e. underlying buffer drained), note that this does not neccessarily
     * mean it reached EOF.
     *
     * @var int
     */
    private $bufferSize;
    private $buffer;

    private $readable = true;
    private $writable = true;
    private $closing = false;
    private $listening = false;

    public function __construct($stream, LoopInterface $loop = null, $readChunkSize = null, WritableStreamInterface $buffer = null)
    {
        if (!\is_resource($stream) || \get_resource_type($stream) !== "stream") {
             throw new InvalidArgumentException('First parameter must be a valid stream resource');
        }

        // ensure resource is opened for reading and wrting (fopen mode must contain "+")
        $meta = \stream_get_meta_data($stream);
        if (isset($meta['mode']) && $meta['mode'] !== '' && \strpos($meta['mode'], '+') === false) {
            throw new InvalidArgumentException('Given stream resource is not opened in read and write mode');
        }

        // this class relies on non-blocking I/O in order to not interrupt the event loop
        // e.g. pipes on Windows do not support this: https://bugs.php.net/bug.php?id=47918
        if (\stream_set_blocking($stream, false) !== true) {
            throw new \RuntimeException('Unable to set stream resource to non-blocking mode');
        }

        // Use unbuffered read operations on the underlying stream resource.
        // Reading chunks from the stream may otherwise leave unread bytes in
        // PHP's stream buffers which some event loop implementations do not
        // trigger events on (edge triggered).
        // This does not affect the default event loop implementation (level
        // triggered), so we can ignore platforms not supporting this (HHVM).
        // Pipe streams (such as STDIN) do not seem to require this and legacy
        // PHP versions cause SEGFAULTs on unbuffered pipe streams, so skip this.
        if (\function_exists('stream_set_read_buffer') && !$this->isLegacyPipe($stream)) {
            \stream_set_read_buffer($stream, 0);
        }

        if ($buffer === null) {
            $buffer = new WritableResourceStream($stream, $loop);
        }

        $this->stream = $stream;
        $this->loop = $loop ?: Loop::get();
        $this->bufferSize = ($readChunkSize === null) ? 65536 : (int)$readChunkSize;
        $this->buffer = $buffer;

        $that = $this;

        $this->buffer->on('error', function ($error) use ($that) {
            $that->emit('error', array($error));
        });

        $this->buffer->on('close', array($this, 'close'));

        $this->buffer->on('drain', function () use ($that) {
            $that->emit('drain');
        });

        $this->resume();
    }

    public function isReadable()
    {
        return $this->readable;
    }

    public function isWritable()
    {
        return $this->writable;
    }

    public function pause()
    {
        if ($this->listening) {
            $this->loop->removeReadStream($this->stream);
            $this->listening = false;
        }
    }

    public function resume()
    {
        if (!$this->listening && $this->readable) {
            $this->loop->addReadStream($this->stream, array($this, 'handleData'));
            $this->listening = true;
        }
    }

    public function write($data)
    {
        if (!$this->writable) {
            return false;
        }

        return $this->buffer->write($data);
    }

    public function close()
    {
        if (!$this->writable && !$this->closing) {
            return;
        }

        $this->closing = false;

        $this->readable = false;
        $this->writable = false;

        $this->emit('close');
        $this->pause();
        $this->buffer->close();
        $this->removeAllListeners();

        if (\is_resource($this->stream)) {
            \fclose($this->stream);
        }
    }

    public function end($data = null)
    {
        if (!$this->writable) {
            return;
        }

        $this->closing = true;

        $this->readable = false;
        $this->writable = false;
        $this->pause();

        $this->buffer->end($data);
    }

    public function pipe(WritableStreamInterface $dest, array $options = array())
    {
        return Util::pipe($this, $dest, $options);
    }

    /** @internal */
    public function handleData($stream)
    {
        $error = null;
        \set_error_handler(function ($errno, $errstr, $errfile, $errline) use (&$error) {
            $error = new \ErrorException(
                $errstr,
                0,
                $errno,
                $errfile,
                $errline
            );
        });

        $data = \stream_get_contents($stream, $this->bufferSize);

        \restore_error_handler();

        if ($error !== null) {
            $this->emit('error', array(new \RuntimeException('Unable to read from stream: ' . $error->getMessage(), 0, $error)));
            $this->close();
            return;
        }

        if ($data !== '') {
            $this->emit('data', array($data));
        } elseif (\feof($this->stream)) {
            // no data read => we reached the end and close the stream
            $this->emit('end');
            $this->close();
        }
    }

    /**
     * Returns whether this is a pipe resource in a legacy environment
     *
     * This works around a legacy PHP bug (#61019) that was fixed in PHP 5.4.28+
     * and PHP 5.5.12+ and newer.
     *
     * @param resource $resource
     * @return bool
     * @link https://github.com/reactphp/child-process/issues/40
     *
     * @codeCoverageIgnore
     */
    private function isLegacyPipe($resource)
    {
        if (\PHP_VERSION_ID < 50428 || (\PHP_VERSION_ID >= 50500 && \PHP_VERSION_ID < 50512)) {
            $meta = \stream_get_meta_data($resource);

            if (isset($meta['stream_type']) && $meta['stream_type'] === 'STDIO') {
                return true;
            }
        }
        return false;
    }
}


1			<?php
2
3			namespace React\Stream;
4
5			use Evenement\EventEmitter;
6			use React\EventLoop\Loop;
7			use React\EventLoop\LoopInterface;
8			use InvalidArgumentException;
9
10			final class DuplexResourceStream extends EventEmitter implements DuplexStreamInterface
11			{
12			private $stream;
13
14			/** @var LoopInterface */
15			private $loop;
16
17			/**
18			* Controls the maximum buffer size in bytes to read at once from the stream.
19			*
20			* This can be a positive number which means that up to X bytes will be read
21			* at once from the underlying stream resource. Note that the actual number
22			* of bytes read may be lower if the stream resource has less than X bytes
23			* currently available.
24			*
25			* This can be `-1` which means read everything available from the
26			* underlying stream resource.
27			* This should read until the stream resource is not readable anymore
28			* (i.e. underlying buffer drained), note that this does not neccessarily
29			* mean it reached EOF.
30			*
31			* @var int
32			*/
33			private $bufferSize;
34			private $buffer;
35
36			private $readable = true;
37			private $writable = true;
38			private $closing = false;
39			private $listening = false;
40
41			public function __construct($stream, LoopInterface $loop = null, $readChunkSize = null, WritableStreamInterface $buffer = null)
42			{
43			if (!\is_resource($stream) \|\| \get_resource_type($stream) !== "stream") {
44			throw new InvalidArgumentException('First parameter must be a valid stream resource');
45			}
46
47			// ensure resource is opened for reading and wrting (fopen mode must contain "+")
48			$meta = \stream_get_meta_data($stream);
49			if (isset($meta['mode']) && $meta['mode'] !== '' && \strpos($meta['mode'], '+') === false) {
50			throw new InvalidArgumentException('Given stream resource is not opened in read and write mode');
51			}
52
53			// this class relies on non-blocking I/O in order to not interrupt the event loop
54			// e.g. pipes on Windows do not support this: https://bugs.php.net/bug.php?id=47918
55			if (\stream_set_blocking($stream, false) !== true) {
56			throw new \RuntimeException('Unable to set stream resource to non-blocking mode');
57			}
58
59			// Use unbuffered read operations on the underlying stream resource.
60			// Reading chunks from the stream may otherwise leave unread bytes in
61			// PHP's stream buffers which some event loop implementations do not
62			// trigger events on (edge triggered).
63			// This does not affect the default event loop implementation (level
64			// triggered), so we can ignore platforms not supporting this (HHVM).
65			// Pipe streams (such as STDIN) do not seem to require this and legacy
66			// PHP versions cause SEGFAULTs on unbuffered pipe streams, so skip this.
67			if (\function_exists('stream_set_read_buffer') && !$this->isLegacyPipe($stream)) {
68			\stream_set_read_buffer($stream, 0);
69			}
70
71			if ($buffer === null) {
72			$buffer = new WritableResourceStream($stream, $loop);
73			}
74
75			$this->stream = $stream;
76			$this->loop = $loop ?: Loop::get();
77			$this->bufferSize = ($readChunkSize === null) ? 65536 : (int)$readChunkSize;
78			$this->buffer = $buffer;
79
80			$that = $this;
81
82			$this->buffer->on('error', function ($error) use ($that) {
83			$that->emit('error', array($error));
84			});
85
86			$this->buffer->on('close', array($this, 'close'));
87
88			$this->buffer->on('drain', function () use ($that) {
89			$that->emit('drain');
90			});
91
92			$this->resume();
93			}
94
95			public function isReadable()
96			{
97			return $this->readable;
98			}
99
100			public function isWritable()
101			{
102			return $this->writable;
103			}
104
105			public function pause()
106			{
107			if ($this->listening) {
108			$this->loop->removeReadStream($this->stream);
109			$this->listening = false;
110			}
111			}
112
113			public function resume()
114			{
115			if (!$this->listening && $this->readable) {
116			$this->loop->addReadStream($this->stream, array($this, 'handleData'));
117			$this->listening = true;
118			}
119			}
120
121			public function write($data)
122			{
123			if (!$this->writable) {
124			return false;
125			}
126
127			return $this->buffer->write($data);
128			}
129
130			public function close()
131			{
132			if (!$this->writable && !$this->closing) {
133			return;
134			}
135
136			$this->closing = false;
137
138			$this->readable = false;
139			$this->writable = false;
140
141			$this->emit('close');
142			$this->pause();
143			$this->buffer->close();
144			$this->removeAllListeners();
145
146			if (\is_resource($this->stream)) {
147			\fclose($this->stream);
148			}
149			}
150
151			public function end($data = null)
152			{
153			if (!$this->writable) {
154			return;
155			}
156
157			$this->closing = true;
158
159			$this->readable = false;
160			$this->writable = false;
161			$this->pause();
162
163			$this->buffer->end($data);
164			}
165
166			public function pipe(WritableStreamInterface $dest, array $options = array())
167			{
168			return Util::pipe($this, $dest, $options);
169			}
170
171			/** @internal */
172			public function handleData($stream)
173			{
174			$error = null;
175			\set_error_handler(function ($errno, $errstr, $errfile, $errline) use (&$error) {
176			$error = new \ErrorException(
177			$errstr,
178			0,
179			$errno,
180			$errfile,
181			$errline
182			);
183			});
184
185			$data = \stream_get_contents($stream, $this->bufferSize);
186
187			\restore_error_handler();
188
189			if ($error !== null) {
190			$this->emit('error', array(new \RuntimeException('Unable to read from stream: ' . $error->getMessage(), 0, $error)));
191			$this->close();
192			return;
193			}
194
195			if ($data !== '') {
196			$this->emit('data', array($data));
197			} elseif (\feof($this->stream)) {
198			// no data read => we reached the end and close the stream
199			$this->emit('end');
200			$this->close();
201			}
202			}
203
204			/**
205			* Returns whether this is a pipe resource in a legacy environment
206			*
207			* This works around a legacy PHP bug (#61019) that was fixed in PHP 5.4.28+
208			* and PHP 5.5.12+ and newer.
209			*
210			* @param resource $resource
211			* @return bool
212			* @link https://github.com/reactphp/child-process/issues/40
213			*
214			* @codeCoverageIgnore
215			*/
216			private function isLegacyPipe($resource)
217			{
218			if (\PHP_VERSION_ID < 50428 \|\| (\PHP_VERSION_ID >= 50500 && \PHP_VERSION_ID < 50512)) {
219			$meta = \stream_get_meta_data($resource);
220
221			if (isset($meta['stream_type']) && $meta['stream_type'] === 'STDIO') {
222			return true;
223			}
224			}
225			return false;
226			}
227			}
228

deployphp / deployer

GitHub Access Token became invalid

Push — master ( 5cefd1...492078 )

DuplexResourceStream::pause() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like