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

Completed

Push — master ( dc4ce9...cc0810 )

by Sergey

created 2016-09-14 08:33 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;

class StreamIO implements IOInterface
{
    /**
     * @var resource|null
     */
    private $stream;

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

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

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

        $parameters = array_merge(['connection_timeout' => 30, 'read_ahead' => 13000, 'timeout' => 1], $parameters);

        $remote = sprintf('%s://%s:%d', $protocol, $host, $port);
        $timeout = $parameters['connection_timeout'];
        $context = $this->createStreamContext($parameters);

        $this->stream = @stream_socket_client($remote, $errno, $errstr, $timeout, STREAM_CLIENT_CONNECT, $context);

        if (!$this->stream) {
            throw new IOException(sprintf('An error occur while connecting to "%s:%d": %s', $host, $port, $errstr));
        }

        $this->buffer = '';

        $this->setReadingTimeout($parameters['timeout']);
        $this->setReadAheadSize($parameters['read_ahead']);

        return $this;
    }

    /**
     * @param float $timeout
     */
    private function setReadingTimeout($timeout)
    {
        list($sec, $usec) = explode('|', number_format($timeout, 6, '|', ''));
        stream_set_timeout($this->stream, $sec, $usec);
    }

    /**
     * @param int $size
     */
    private function setReadAheadSize($size)
    {
        $this->readAheadSize = $size;
    }

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

skolodyazhnyy / butter-amqplib

Push — master ( dc4ce9...cc0810 )

StreamIO::createStreamContext() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like