UnwrapWritableStream::__construct() - 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

UnwrapWritableStream::__construct() B

↳ Parent: UnwrapWritableStream

Complexity

Conditions	10
Paths	1

Size

Total Lines	72
Code Lines	42

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	10
eloc	42
c	1
b	0
f	0
nc	1
nop	1
dl	0
loc	72
rs	7.6666

How to fix Long Method Complexity

<?php

namespace React\Promise\Stream;

use Evenement\EventEmitter;
use InvalidArgumentException;
use React\Promise\CancellablePromiseInterface;
use React\Promise\PromiseInterface;
use React\Stream\WritableStreamInterface;

/**
 * @internal
 * @see unwrapWritable() instead
 */
class UnwrapWritableStream extends EventEmitter implements WritableStreamInterface
{
    private $promise;
    private $stream;
    private $buffer = array();
    private $closed = false;
    private $ending = false;

    /**
     * Instantiate new unwrapped writable stream for given `Promise` which resolves with a `WritableStreamInterface`.
     *
     * @param PromiseInterface $promise Promise<WritableStreamInterface, Exception>
     */
    public function __construct(PromiseInterface $promise)
    {
        $out = $this;
        $store  =& $this->stream;
        $buffer =& $this->buffer;
        $ending =& $this->ending;
        $closed =& $this->closed;

        $this->promise = $promise->then(
            function ($stream) {
                if (!$stream instanceof WritableStreamInterface) {
                    throw new InvalidArgumentException('Not a writable stream');
                }
                return $stream;
            }
        )->then(
            function (WritableStreamInterface $stream) use ($out, &$store, &$buffer, &$ending, &$closed) {
                // stream is already closed, make sure to close output stream
                if (!$stream->isWritable()) {
                    $out->close();
                    return $stream;
                }

                // resolves but output is already closed, make sure to close stream silently
                if ($closed) {
                    $stream->close();
                    return $stream;
                }

                // forward drain events for back pressure
                $stream->on('drain', function () use ($out) {
                    $out->emit('drain', array($out));
                });

                // error events cancel output stream
                $stream->on('error', function ($error) use ($out) {
                    $out->emit('error', array($error, $out));
                    $out->close();
                });

                // close both streams once either side closes
                $stream->on('close', array($out, 'close'));
                $out->on('close', array($stream, 'close'));

                if ($buffer) {
                    // flush buffer to stream and check if its buffer is not exceeded
                    $drained = true;
                    foreach ($buffer as $chunk) {
                        if (!$stream->write($chunk)) {
                            $drained = false;
                        }
                    }
                    $buffer = array();

                    if ($drained) {
                        // signal drain event, because the output stream previous signalled a full buffer
                        $out->emit('drain', array($out));
                    }
                }

                if ($ending) {
                    $stream->end();
                } else {
                    $store = $stream;
                }

                return $stream;
            },
            function ($e) use ($out, &$closed) {
                if (!$closed) {
                    $out->emit('error', array($e, $out));
                    $out->close();
                }
            }
        );
    }

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

        // forward to inner stream if possible
        if ($this->stream !== null) {
            return $this->stream->write($data);
        }

        // append to buffer and signal the buffer is full
        $this->buffer[] = $data;
        return false;
    }

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

        $this->ending = true;

        // forward to inner stream if possible
        if ($this->stream !== null) {
            return $this->stream->end($data);
        }

        // append to buffer
        if ($data !== null) {
            $this->buffer[] = $data;
        }
    }

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

    public function close()
    {
        if ($this->closed) {
            return;
        }

        $this->buffer = array();
        $this->ending = true;
        $this->closed = true;

        // try to cancel promise once the stream closes
        if ($this->promise instanceof CancellablePromiseInterface) {
            $this->promise->cancel();
        }
        $this->promise = $this->stream = null;

        $this->emit('close');
        $this->removeAllListeners();
    }
}


1			<?php
2
3			namespace React\Promise\Stream;
4
5			use Evenement\EventEmitter;
6			use InvalidArgumentException;
7			use React\Promise\CancellablePromiseInterface;
8			use React\Promise\PromiseInterface;
9			use React\Stream\WritableStreamInterface;
10
11			/**
12			* @internal
13			* @see unwrapWritable() instead
14			*/
15			class UnwrapWritableStream extends EventEmitter implements WritableStreamInterface
16			{
17			private $promise;
18			private $stream;
19			private $buffer = array();
20			private $closed = false;
21			private $ending = false;
22
23			/**
24			* Instantiate new unwrapped writable stream for given `Promise` which resolves with a `WritableStreamInterface`.
25			*
26			* @param PromiseInterface $promise Promise<WritableStreamInterface, Exception>
27			*/
28			public function __construct(PromiseInterface $promise)
29			{
30			$out = $this;
31			$store =& $this->stream;
32			$buffer =& $this->buffer;
33			$ending =& $this->ending;
34			$closed =& $this->closed;
35
36			$this->promise = $promise->then(
37			function ($stream) {
38			if (!$stream instanceof WritableStreamInterface) {
39			throw new InvalidArgumentException('Not a writable stream');
40			}
41			return $stream;
42			}
43			)->then(
44			function (WritableStreamInterface $stream) use ($out, &$store, &$buffer, &$ending, &$closed) {
45			// stream is already closed, make sure to close output stream
46			if (!$stream->isWritable()) {
47			$out->close();
48			return $stream;
49			}
50
51			// resolves but output is already closed, make sure to close stream silently
52			if ($closed) {
53			$stream->close();
54			return $stream;
55			}
56
57			// forward drain events for back pressure
58			$stream->on('drain', function () use ($out) {
59			$out->emit('drain', array($out));
60			});
61
62			// error events cancel output stream
63			$stream->on('error', function ($error) use ($out) {
64			$out->emit('error', array($error, $out));
65			$out->close();
66			});
67
68			// close both streams once either side closes
69			$stream->on('close', array($out, 'close'));
70			$out->on('close', array($stream, 'close'));
71
72			if ($buffer) {
73			// flush buffer to stream and check if its buffer is not exceeded
74			$drained = true;
75			foreach ($buffer as $chunk) {
76			if (!$stream->write($chunk)) {
77			$drained = false;
78			}
79			}
80			$buffer = array();
81
82			if ($drained) {
83			// signal drain event, because the output stream previous signalled a full buffer
84			$out->emit('drain', array($out));
85			}
86			}
87
88			if ($ending) {
89			$stream->end();
90			} else {
91			$store = $stream;
92			}
93
94			return $stream;
95			},
96			function ($e) use ($out, &$closed) {
97			if (!$closed) {
98			$out->emit('error', array($e, $out));
99			$out->close();
100			}
101			}
102			);
103			}
104
105			public function write($data)
106			{
107			if ($this->ending) {
108			return false;
109			}
110
111			// forward to inner stream if possible
112			if ($this->stream !== null) {
113			return $this->stream->write($data);
114			}
115
116			// append to buffer and signal the buffer is full
117			$this->buffer[] = $data;
118			return false;
119			}
120
121			public function end($data = null)
122			{
123			if ($this->ending) {
124			return;
125			}
126
127			$this->ending = true;
128
129			// forward to inner stream if possible
130			if ($this->stream !== null) {
131			return $this->stream->end($data);
132			}
133
134			// append to buffer
135			if ($data !== null) {
136			$this->buffer[] = $data;
137			}
138			}
139
140			public function isWritable()
141			{
142			return !$this->ending;
143			}
144
145			public function close()
146			{
147			if ($this->closed) {
148			return;
149			}
150
151			$this->buffer = array();
152			$this->ending = true;
153			$this->closed = true;
154
155			// try to cancel promise once the stream closes
156			if ($this->promise instanceof CancellablePromiseInterface) {
157			$this->promise->cancel();
158			}
159			$this->promise = $this->stream = null;
160
161			$this->emit('close');
162			$this->removeAllListeners();
163			}
164			}
165

deployphp / deployer

GitHub Access Token became invalid

Push — master ( 5cefd1...492078 )

UnwrapWritableStream::__construct() B

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Duplication Side-by-Side

Filter issues like