StreamIO::createStreamContext() - Code Metrics - Inspection of "Small refectoring" - skolodyazhnyy/butter-amqplib - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( a7780f...563711 )

by Sergey

created 2016-09-13 21:10 UTC

StreamIO::createStreamContext() A

↳ Parent: StreamIO

Complexity

Conditions	3
Paths	3

Size

Total Lines	23
Code Lines	14

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	9
CRAP Score	3

Importance

Changes

Metric	Value
dl	0
loc	23
ccs	9
cts	9
cp	1
rs	9.0856
c	0
b	0
f	0
cc	3
eloc	14
nc	3
nop	1
crap	3

<?php

namespace ButterAMQP\IO;

use ButterAMQP\Exception\IOClosedException;
use ButterAMQP\Exception\IOException;
use ButterAMQP\IOInterface;
use ButterAMQP\Debug\ReadableBinaryData;
use Psr\Log\LoggerAwareInterface;
use Psr\Log\LoggerAwareTrait;
use Psr\Log\NullLogger;

class StreamIO implements IOInterface, LoggerAwareInterface
{
    use LoggerAwareTrait;

    /**
     * @var resource|null
     */
    private $stream;

    /**
     * @var string
     */
    private $buffer;

    /**
     * @var int
     */
    private $readAheadSize;

    /**
     * Initialize default logger.
     */
    public function __construct()
    {
        $this->logger = new NullLogger();
    }

    /**
     * {@inheritdoc}
     */
    public function open($protocol, $host, $port, array $parameters = [])
    {
        if ($this->stream && $this->isOpen()) {
            return $this;
        }

        $context = $this->createStreamContext($parameters);

        $this->stream = @stream_socket_client(
            sprintf('%s://%s:%d', $protocol, $host, $port),
            $errno,
            $errstr,
            isset($parameters['connection_timeout']) ? $parameters['connection_timeout'] : 30,
            STREAM_CLIENT_CONNECT,
            $context
        );

        if (!$this->stream) {
            throw new IOException(sprintf(
                'Unable to connect to "%s:%d" using stream socket: %s',
                $host,
                $port,
                $errstr
            ));
        }

        $this->buffer = '';

        if (isset($parameters['timeout'])) {
            list($sec, $usec) = explode('|', number_format($parameters['timeout'], 6, '|', ''));
            stream_set_timeout($this->stream, $sec, $usec);
        }

        if (isset($parameters['read_ahead'])) {
            $this->readAheadSize = $parameters['read_ahead'];
        }

        return $this;
    }

    /**
     * @param array $parameters
     *
     * @return resource
     */
    private function createStreamContext(array $parameters)
    {
        static $options = [
            'certfile' => 'local_cert',
            'keyfile' => 'local_pk',
            'cacertfile' => 'cafile',
            'passphrase' => 'passphrase',
            'verify' => 'verify_peer',
            'allow_self_signed' => 'allow_self_signed',
        ];

        $context = stream_context_create();

        foreach ($parameters as $name => $value) {
            if (!isset($options[$name])) {
                continue;
            }

            stream_context_set_option($context, 'ssl', $options[$name], $value);
        }

        return $context;
    }

    /**
     * @return $this
     */
    public function close()
    {
        if (!$this->stream) {
            return $this;
        }

        fclose($this->stream);

        $this->stream = null;

        return $this;
    }

    /**
     * @param string   $data
     * @param int|null $length
     *
     * @return $this
     *
     * @throws IOException
     */
    public function write($data, $length = null)
    {
        if ($this->stream === null) {
            throw new IOClosedException('Connection is not open');
        }

        if ($length === null) {
            $length = strlen($data);
        }

        $this->logger->debug(new ReadableBinaryData('Sending', $data));

        while ($length > 0) {
            if ($this->isOpen()) {
                throw new IOClosedException('Connection is closed');
            }

            $written = @fwrite($this->stream, $data, $length);
            if ($written === false) {
                throw new IOException('An error occur while writing to socket');
            }

            $length -= $written;
            $data = $length ? substr($data, $written, $length) : '';
        }

        return $this;
    }

    /**
     * {@inheritdoc}
     */
    public function peek($length, $blocking = true)
    {
        $received = strlen($this->buffer);

        if ($received >= $length) {
            return $this->buffer;
        }

        $this->buffer .= $this->recv($length - $received, $blocking);

        if (strlen($this->buffer) >= $length) {
            return $this->buffer;
        }

        return null;
    }

    /**
     * {@inheritdoc}
     */
    public function read($length, $blocking = true)
    {
        if ($this->peek($length, $blocking) === null) {
            return null;
        }

        $data = substr($this->buffer, 0, $length);

        $this->buffer = substr($this->buffer, $length, strlen($this->buffer) - $length);

        return $data;
    }

    /**
     * @param int  $length
     * @param bool $blocking
     *
     * @return string
     *
     * @throws IOException
     */
    private function recv($length, $blocking)
    {
        if ($this->stream === null) {
            throw new IOClosedException('Connection is not open');
        }

        if ($this->isOpen()) {
            throw new IOClosedException('Connection is closed');
        }

        if ($this->readAheadSize) {
            $meta = stream_get_meta_data($this->stream);

            if ($length < $meta['unread_bytes']) {
                $length = min($this->readAheadSize, $meta['unread_bytes']);
            }
        }

        stream_set_blocking($this->stream, $blocking);

        if (($received = fread($this->stream, $length)) === false) {
            throw new IOException('An error occur while reading from the socket');
        }

        return $received;
    }

    /**
     * @return bool
     */
    public function isOpen()
    {
        return is_resource($this->stream) && feof($this->stream);
    }
}


1			<?php
2
3			namespace ButterAMQP\IO;
4
5			use ButterAMQP\Exception\IOClosedException;
6			use ButterAMQP\Exception\IOException;
7			use ButterAMQP\IOInterface;
8			use ButterAMQP\Debug\ReadableBinaryData;
9			use Psr\Log\LoggerAwareInterface;
10			use Psr\Log\LoggerAwareTrait;
11			use Psr\Log\NullLogger;
12
13			class StreamIO implements IOInterface, LoggerAwareInterface
14			{
15			use LoggerAwareTrait;
16
17			/**
18			* @var resource\|null
19			*/
20			private $stream;
21
22			/**
23			* @var string
24			*/
25			private $buffer;
26
27			/**
28			* @var int
29			*/
30			private $readAheadSize;
31
32			/**
33			* Initialize default logger.
34			*/
35	25		public function __construct()
36			{
37	25		$this->logger = new NullLogger();
38	25		}
39
40			/**
41			* {@inheritdoc}
42			*/
43	25		public function open($protocol, $host, $port, array $parameters = [])
44			{
45	25		if ($this->stream && $this->isOpen()) {
46			return $this;
47			}
48
49	25		$context = $this->createStreamContext($parameters);
50
51	25		$this->stream = @stream_socket_client(
52	25		sprintf('%s://%s:%d', $protocol, $host, $port),
53	25		$errno,
54	25		$errstr,
55	25		isset($parameters['connection_timeout']) ? $parameters['connection_timeout'] : 30,
56	25		STREAM_CLIENT_CONNECT,
57			$context
58	25		);
59
60	25		if (!$this->stream) {
61			throw new IOException(sprintf(
62			'Unable to connect to "%s:%d" using stream socket: %s',
63			$host,
64			$port,
65			$errstr
66			));
67			}
68
69	25		$this->buffer = '';
70
71	25		if (isset($parameters['timeout'])) {
72	7		list($sec, $usec) = explode('\|', number_format($parameters['timeout'], 6, '\|', ''));
73	7		stream_set_timeout($this->stream, $sec, $usec);
74	7		}
75
76	25		if (isset($parameters['read_ahead'])) {
77			$this->readAheadSize = $parameters['read_ahead'];
78			}
79
80	25		return $this;
81			}
82
83			/**
84			* @param array $parameters
85			*
86			* @return resource
87			*/
88	25		private function createStreamContext(array $parameters)
89			{
90			static $options = [
91			'certfile' => 'local_cert',
92			'keyfile' => 'local_pk',
93			'cacertfile' => 'cafile',
94			'passphrase' => 'passphrase',
95			'verify' => 'verify_peer',
96			'allow_self_signed' => 'allow_self_signed',
97	25		];
98
99	25		$context = stream_context_create();
100
101	25		foreach ($parameters as $name => $value) {
102	7		if (!isset($options[$name])) {
103	7		continue;
104			}
105
106	1		stream_context_set_option($context, 'ssl', $options[$name], $value);
107	25		}
108
109	25		return $context;
110			}
111
112			/**
113			* @return $this
114			*/
115	12		public function close()
116			{
117	12		if (!$this->stream) {
118	1		return $this;
119			}
120
121	12		fclose($this->stream);
122
123	12		$this->stream = null;
124
125	12		return $this;
126			}
127
128			/**
129			* @param string $data
130			* @param int\|null $length
131			*
132			* @return $this
133			*
134			* @throws IOException
135			*/
136	22		public function write($data, $length = null)
137			{
138	22		if ($this->stream === null) {
139			throw new IOClosedException('Connection is not open');
140			}
141
142	22		if ($length === null) {
143	22		$length = strlen($data);
144	22		}
145
146	22		$this->logger->debug(new ReadableBinaryData('Sending', $data));
147
148	22		while ($length > 0) {
149	22		if ($this->isOpen()) {
150	1		throw new IOClosedException('Connection is closed');
151			}
152
153	22		$written = @fwrite($this->stream, $data, $length);
154	22		if ($written === false) {
155			throw new IOException('An error occur while writing to socket');
156			}
157
158	22		$length -= $written;
159	22		$data = $length ? substr($data, $written, $length) : '';
160	22		}
161
162	22		return $this;
163			}
164
165			/**
166			* {@inheritdoc}
167			*/
168	23		public function peek($length, $blocking = true)
169			{
170	23		$received = strlen($this->buffer);
171
172	23		if ($received >= $length) {
173	2		return $this->buffer;
174			}
175
176	23		$this->buffer .= $this->recv($length - $received, $blocking);
177
178	23		if (strlen($this->buffer) >= $length) {
179	22		return $this->buffer;
180			}
181
182	4		return null;
183			}
184
185			/**
186			* {@inheritdoc}
187			*/
188	22	View Code Duplication	public function read($length, $blocking = true)
189			{
190	22		if ($this->peek($length, $blocking) === null) {
191	3		return null;
192			}
193
194	21		$data = substr($this->buffer, 0, $length);
195
196	21		$this->buffer = substr($this->buffer, $length, strlen($this->buffer) - $length);
197
198	21		return $data;
199			}
200
201			/**
202			* @param int $length
203			* @param bool $blocking
204			*
205			* @return string
206			*
207			* @throws IOException
208			*/
209	23		private function recv($length, $blocking)
210			{
211	23		if ($this->stream === null) {
212			throw new IOClosedException('Connection is not open');
213			}
214
215	23		if ($this->isOpen()) {
216	1		throw new IOClosedException('Connection is closed');
217			}
218
219	23		if ($this->readAheadSize) {
220			$meta = stream_get_meta_data($this->stream);
221
222			if ($length < $meta['unread_bytes']) {
223			$length = min($this->readAheadSize, $meta['unread_bytes']);
224			}
225			}
226
227	23		stream_set_blocking($this->stream, $blocking);
228
229	23		if (($received = fread($this->stream, $length)) === false) {
230			throw new IOException('An error occur while reading from the socket');
231			}
232
233	23		return $received;
234			}
235
236			/**
237			* @return bool
238			*/
239	25		public function isOpen()
240			{
241	25		return is_resource($this->stream) && feof($this->stream);
242			}
243			}
244

skolodyazhnyy / butter-amqplib

Push — master ( a7780f...563711 )

StreamIO::createStreamContext() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like