StreamIO::open() - Code Metrics - skolodyazhnyy/butter-amqplib - Measure and Improve Code Quality continuously with Scrutinizer

StreamIO::open() A
last analyzed 2018-06-14 19:28 UTC

↳ Parent: StreamIO

Complexity

Conditions	2
Paths	2

Size

Total Lines	17
Code Lines	10

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	9
CRAP Score	2.004

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
dl	0
loc	17
ccs	9
cts	10
cp	0.9
rs	9.4285
c	1
b	0
f	0
cc	2
eloc	10
nc	2
nop	4
crap	2.004

<?php

namespace ButterAMQP\IO;

use ButterAMQP\Exception\IOClosedException;
use ButterAMQP\Exception\IOException;
use ButterAMQP\IOInterface;

class StreamIO implements IOInterface
{
    const DEFAULT_CONNECTION_TIMEOUT = 30;
    const DEFAULT_READING_TIMEOUT = 1;

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

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

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

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

        $this->stream = $this->openConnection(
            $protocol,
            $host,
            $port,
            $parameters
        );

        $this->tuneConnection($parameters);

        return $this;
    }

    /**
     * @param string $protocol
     * @param string $host
     * @param int    $port
     * @param array  $parameters
     *
     * @return resource
     *
     * @throws IOException
     */
    private function openConnection($protocol, $host, $port, array $parameters = [])
    {
        $timeout = isset($parameters['connection_timeout']) ?
            $parameters['connection_timeout'] : self::DEFAULT_CONNECTION_TIMEOUT;

        $stream = @stream_socket_client(
            sprintf('%s://%s:%d', $protocol, $host, $port),
            $errno,
            $errstr,
            $timeout,
            STREAM_CLIENT_CONNECT,
            $this->createStreamContext($parameters)
        );

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

        return $stream;
    }

    /**
     * @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;
    }

    /**
     * @param array $parameters
     */
    private function tuneConnection(array $parameters)
    {
        $timeout = isset($parameters['timeout']) ? $parameters['timeout'] : self::DEFAULT_READING_TIMEOUT;

        list($sec, $usec) = explode('|', number_format($timeout, 6, '|', ''));

        $this->timeoutSec = $sec;
        $this->timeoutUsec = $usec;

        stream_set_timeout($this->stream, $this->timeoutSec, $this->timeoutUsec);
        stream_set_read_buffer($this->stream, 0);
        stream_set_write_buffer($this->stream, 0);
    }

    /**
     * @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) {
            $written = @fwrite($this->stream, $data, $length);
            if (!$written) {
                throw new IOException('An error occur while writing to socket');
            }

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

        return $this;
    }

    /**
     * {@inheritdoc}
     */
    public function read($length, $blocking = true)
    {
        stream_set_blocking($this->stream, $blocking);

        if (!$this->isOpen()) {
            throw new IOClosedException('Socket is closed or was not open');
        }

        $r = [$this->stream];
        $w = null;
        $e = null;

        if ($blocking && @stream_select($r, $w, $e, $this->timeoutSec, $this->timeoutUsec) === false) {
            throw new IOException('An error occur while selecting stream');
        }

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

skolodyazhnyy / butter-amqplib

StreamIO::open() A last analyzed 2018-06-14 19:28 UTC

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like

StreamIO::open() A
last analyzed 2018-06-14 19:28 UTC