SocketIO::select() - Code Metrics - videlalvaro/php-amqplib - Measure and Improve Code Quality continuously with Scrutinizer

SocketIO::select() A
last analyzed 2016-02-02 18:10 UTC

↳ Parent: SocketIO

Complexity

Conditions	1
Paths	1

Size

Total Lines	8
Code Lines	5

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	0
CRAP Score	2

Importance

Changes	3
Bugs	2	Features	1

Metric	Value
c	3
b	2
f	1
dl	0
loc	8
ccs	0
cts	5
cp	0
rs	9.4285
cc	1
eloc	5
nc	1
nop	2
crap	2

<?php
namespace PhpAmqpLib\Wire\IO;

use PhpAmqpLib\Exception\AMQPIOException;
use PhpAmqpLib\Exception\AMQPRuntimeException;

class SocketIO extends AbstractIO
{
    /** @var string */
    protected $host;

    /** @var int */
    protected $port;

    /** @var int */
    protected $timeout;

    /** @var resource */
    private $sock;

    /** @var bool */
    private $keepalive;

    /**
     * @param string $host
     * @param int $port
     * @param int $timeout
     * @param bool $keepalive
     */
    public function __construct($host, $port, $timeout, $keepalive = false)
    {
        $this->host = $host;
        $this->port = $port;
        $this->timeout = $timeout;
        $this->keepalive = $keepalive;
    }

    /**
     * Sets up the socket connection
     *
     * @throws \Exception
     */
    public function connect()
    {
        $this->sock = socket_create(AF_INET, SOCK_STREAM, SOL_TCP);

        socket_set_option($this->sock, SOL_SOCKET, SO_RCVTIMEO, array('sec' => $this->timeout, 'usec' => 0));
        socket_set_option($this->sock, SOL_SOCKET, SO_SNDTIMEO, array('sec' => $this->timeout, 'usec' => 0));

        if (!socket_connect($this->sock, $this->host, $this->port)) {
            $errno = socket_last_error($this->sock);
            $errstr = socket_strerror($errno);
            throw new AMQPIOException(sprintf(
                'Error Connecting to server (%s): %s',
                $errno,
                $errstr
            ), $errno);
        }

        socket_set_block($this->sock);
        socket_set_option($this->sock, SOL_TCP, TCP_NODELAY, 1);

        if ($this->keepalive) {
            $this->enable_keepalive();
        }
    }

    /**
     * @return resource
     */
    public function getSocket()
    {
        return $this->sock;
    }

    /**
     * Reconnects the socket
     */
    public function reconnect()
    {
        $this->close();
        $this->connect();
    }

    /**
     * @param $n
     * @return mixed|string
     * @throws \PhpAmqpLib\Exception\AMQPIOException
     * @throws \PhpAmqpLib\Exception\AMQPRuntimeException
     */
    public function read($n)
    {
        $res = '';
        $read = 0;

        $buf = socket_read($this->sock, $n);
        while ($read < $n && $buf !== '' && $buf !== false) {
            // Null sockets are invalid, throw exception
            if (is_null($this->sock)) {
                throw new AMQPRuntimeException(sprintf(
                    'Socket was null! Last SocketError was: %s',
                    socket_strerror(socket_last_error())
                ));
            }

            $read += mb_strlen($buf, 'ASCII');
            $res .= $buf;
            $buf = socket_read($this->sock, $n - $read);
        }

        if (mb_strlen($res, 'ASCII') != $n) {
            throw new AMQPIOException(sprintf(
                'Error reading data. Received %s instead of expected %s bytes',
                mb_strlen($res, 'ASCII'),
                $n
            ));
        }

        return $res;
    }

    /**
     * @param $data
     * @return mixed|void
     * @throws \PhpAmqpLib\Exception\AMQPIOException
     * @throws \PhpAmqpLib\Exception\AMQPRuntimeException
     */
    public function write($data)
    {
        $len = mb_strlen($data, 'ASCII');

        while (true) {
            // Null sockets are invalid, throw exception
            if (is_null($this->sock)) {
                throw new AMQPRuntimeException(sprintf(
                    'Socket was null! Last SocketError was: %s',
                    socket_strerror(socket_last_error())
                ));
            }

            $sent = socket_write($this->sock, $data, $len);
            if ($sent === false) {
                throw new AMQPIOException(sprintf(
                    'Error sending data. Last SocketError: %s',
                    socket_strerror(socket_last_error())
                ));
            }

            // Check if the entire message has been sent
            if ($sent < $len) {
                // If not sent the entire message.
                // Get the part of the message that has not yet been sent as message
                $data = mb_substr($data, $sent, mb_strlen($data, 'ASCII') - $sent, 'ASCII');
                // Get the length of the not sent part
                $len -= $sent;
            } else {
                break;
            }
        }
    }

    public function close()
    {
        if (is_resource($this->sock)) {
            socket_close($this->sock);
        }
        $this->sock = null;
    }

    /**
     * @param $sec
     * @param $usec
     * @return int|mixed
     */
    public function select($sec, $usec)
    {
        $read = array($this->sock);
        $write = null;
        $except = null;

        return socket_select($read, $write, $except, $sec, $usec);
    }

    /**
     * @throws \PhpAmqpLib\Exception\AMQPIOException
     */
    protected function enable_keepalive()
    {
        if (!defined('SOL_SOCKET') || !defined('SO_KEEPALIVE')) {
            throw new AMQPIOException('Can not enable keepalive: SOL_SOCKET or SO_KEEPALIVE is not defined');
        }

        socket_set_option($this->sock, SOL_SOCKET, SO_KEEPALIVE, 1);
    }
}


1		<?php
2		namespace PhpAmqpLib\Wire\IO;
3
4		use PhpAmqpLib\Exception\AMQPIOException;
5		use PhpAmqpLib\Exception\AMQPRuntimeException;
6
7		class SocketIO extends AbstractIO
8		{
9		/** @var string */
10		protected $host;
11
12		/** @var int */
13		protected $port;
14
15		/** @var int */
16		protected $timeout;
17
18		/** @var resource */
19		private $sock;
20
21		/** @var bool */
22		private $keepalive;
23
24		/**
25		* @param string $host
26		* @param int $port
27		* @param int $timeout
28		* @param bool $keepalive
29		*/
30	25	public function __construct($host, $port, $timeout, $keepalive = false)
31		{
32	25	$this->host = $host;
33	25	$this->port = $port;
34	25	$this->timeout = $timeout;
35	25	$this->keepalive = $keepalive;
36	25	}
37
38		/**
39		* Sets up the socket connection
40		*
41		* @throws \Exception
42		*/
43	25	public function connect()
44		{
45	25	$this->sock = socket_create(AF_INET, SOCK_STREAM, SOL_TCP);
46
47	25	socket_set_option($this->sock, SOL_SOCKET, SO_RCVTIMEO, array('sec' => $this->timeout, 'usec' => 0));
48	25	socket_set_option($this->sock, SOL_SOCKET, SO_SNDTIMEO, array('sec' => $this->timeout, 'usec' => 0));
49
50	25	if (!socket_connect($this->sock, $this->host, $this->port)) {
51		$errno = socket_last_error($this->sock);
52		$errstr = socket_strerror($errno);
53		throw new AMQPIOException(sprintf(
54		'Error Connecting to server (%s): %s',
55		$errno,
56		$errstr
57		), $errno);
58		}
59
60	25	socket_set_block($this->sock);
61	25	socket_set_option($this->sock, SOL_TCP, TCP_NODELAY, 1);
62
63	25	if ($this->keepalive) {
64		$this->enable_keepalive();
65		}
66	25	}
67
68		/**
69		* @return resource
70		*/
71		public function getSocket()
72		{
73		return $this->sock;
74		}
75
76		/**
77		* Reconnects the socket
78		*/
79	10	public function reconnect()
80		{
81	10	$this->close();
82	10	$this->connect();
83	10	}
84
85		/**
86		* @param $n
87		* @return mixed\|string
88		* @throws \PhpAmqpLib\Exception\AMQPIOException
89		* @throws \PhpAmqpLib\Exception\AMQPRuntimeException
90		*/
91	25	public function read($n)
92		{
93	25	$res = '';
94	25	$read = 0;
95
96	25	$buf = socket_read($this->sock, $n);
97	25	while ($read < $n && $buf !== '' && $buf !== false) {
98		// Null sockets are invalid, throw exception
99	25	if (is_null($this->sock)) {
100		throw new AMQPRuntimeException(sprintf(
101		'Socket was null! Last SocketError was: %s',
102		socket_strerror(socket_last_error())
103		));
104		}
105
106	25	$read += mb_strlen($buf, 'ASCII');
107	25	$res .= $buf;
108	25	$buf = socket_read($this->sock, $n - $read);
109	20	}
110
111	25	if (mb_strlen($res, 'ASCII') != $n) {
112		throw new AMQPIOException(sprintf(
113		'Error reading data. Received %s instead of expected %s bytes',
114		mb_strlen($res, 'ASCII'),
115		$n
116		));
117		}
118
119	25	return $res;
120		}
121
122		/**
123		* @param $data
124		* @return mixed\|void
125		* @throws \PhpAmqpLib\Exception\AMQPIOException
126		* @throws \PhpAmqpLib\Exception\AMQPRuntimeException
127		*/
128	25	public function write($data)
129		{
130	25	$len = mb_strlen($data, 'ASCII');
131
132	25	while (true) {
133		// Null sockets are invalid, throw exception
134	25	if (is_null($this->sock)) {
135		throw new AMQPRuntimeException(sprintf(
136		'Socket was null! Last SocketError was: %s',
137		socket_strerror(socket_last_error())
138		));
139		}
140
141	25	$sent = socket_write($this->sock, $data, $len);
142	25	if ($sent === false) {
143		throw new AMQPIOException(sprintf(
144		'Error sending data. Last SocketError: %s',
145		socket_strerror(socket_last_error())
146		));
147		}
148
149		// Check if the entire message has been sent
150	25	if ($sent < $len) {
151		// If not sent the entire message.
152		// Get the part of the message that has not yet been sent as message
153		$data = mb_substr($data, $sent, mb_strlen($data, 'ASCII') - $sent, 'ASCII');
154		// Get the length of the not sent part
155		$len -= $sent;
156		} else {
157	25	break;
158		}
159		}
160	25	}
161
162	25	public function close()
163		{
164	25	if (is_resource($this->sock)) {
165	25	socket_close($this->sock);
166	20	}
167	25	$this->sock = null;
168	25	}
169
170		/**
171		* @param $sec
172		* @param $usec
173		* @return int\|mixed
174		*/
175		public function select($sec, $usec)
176		{
177		$read = array($this->sock);
178		$write = null;
179		$except = null;
180
181		return socket_select($read, $write, $except, $sec, $usec);
182		}
183
184		/**
185		* @throws \PhpAmqpLib\Exception\AMQPIOException
186		*/
187		protected function enable_keepalive()
188		{
189		if (!defined('SOL_SOCKET') \|\| !defined('SO_KEEPALIVE')) {
190		throw new AMQPIOException('Can not enable keepalive: SOL_SOCKET or SO_KEEPALIVE is not defined');
191		}
192
193		socket_set_option($this->sock, SOL_SOCKET, SO_KEEPALIVE, 1);
194		}
195		}
196

videlalvaro / php-amqplib

SocketIO::select() A last analyzed 2016-02-02 18:10 UTC

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like

SocketIO::select() A
last analyzed 2016-02-02 18:10 UTC