StreamIO::connect() - Code Metrics - Inspection of "Heartbeat writetimeout fix" - videlalvaro/php-amqplib - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#348)

unknown

created 2015-12-15 15:39 UTC

StreamIO::connect() B

↳ Parent: StreamIO

Complexity

Conditions	7
Paths	9

Size

Total Lines	64
Code Lines	40

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	26
CRAP Score	11.0278

Importance

Changes	7
Bugs	4	Features	1

Metric	Value
c	7
b	4
f	1
dl	0
loc	64
ccs	26
cts	46
cp	0.5652
rs	7.2058
cc	7
eloc	40
nc	9
nop	0
crap	11.0278

How to fix Long Method

<?php
namespace PhpAmqpLib\Wire\IO;

use PhpAmqpLib\Exception\AMQPIOException;
use PhpAmqpLib\Exception\AMQPRuntimeException;
use PhpAmqpLib\Exception\AMQPTimeoutException;
use PhpAmqpLib\Helper\MiscHelper;
use PhpAmqpLib\Wire\AMQPWriter;

class StreamIO extends AbstractIO
{
    /** @var string */
    protected $protocol;

    /** @var string */
    protected $host;

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

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

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

    /** @var resource */
    protected $context;

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

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

    /** @var float */
    protected $last_read;

    /** @var float */
    protected $last_write;

    /** @var array */
    protected $last_error;

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

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

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

    /**
     * @param string $host
     * @param int $port
     * @param int $connection_timeout
     * @param int $read_write_timeout
     * @param null $context
     * @param bool $keepalive
     * @param int $heartbeat
     */
    public function __construct(
        $host,
        $port,
        $connection_timeout,
        $read_write_timeout,
        $context = null,
        $keepalive = false,
        $heartbeat = 0
    ) {
        $this->protocol = 'tcp';
        $this->host = $host;
        $this->port = $port;
        $this->connection_timeout = $connection_timeout;
        $this->read_write_timeout = $read_write_timeout;
        $this->context = $context;
        $this->keepalive = $keepalive;
        $this->heartbeat = $heartbeat;
        $this->canSelectNull = true;
        $this->canDispatchPcntlSignal = $this->isPcntlSignalEnabled();

        if (is_null($this->context)) {
            $this->context = stream_context_create();
        } else {
            $this->protocol = 'ssl';
            // php bugs 41631 & 65137 prevent select null from working on ssl streams
            if (PHP_VERSION_ID < 50436) {
                $this->canSelectNull = false;
            }
        }
    }

    /**
     * @return bool
     */
    private function isPcntlSignalEnabled()
    {
        return extension_loaded('pcntl')
            && function_exists('pcntl_signal_dispatch')
            && (defined('AMQP_WITHOUT_SIGNALS') && !AMQP_WITHOUT_SIGNALS);
    }

    /**
     * Sets up the stream connection
     *
     * @throws \PhpAmqpLib\Exception\AMQPRuntimeException
     * @throws \Exception
     */
    public function connect()
    {
        $errstr = $errno = null;

        $remote = sprintf(
            '%s://%s:%s',
            $this->protocol,
            $this->host,
            $this->port
        );

        set_error_handler(array($this, 'error_handler'));

        $this->sock = stream_socket_client(
            $remote,
            $errno,
            $errstr,
            $this->connection_timeout,
            STREAM_CLIENT_CONNECT,
            $this->context
        );

        restore_error_handler();

        if (false === $this->sock) {
            throw new AMQPRuntimeException(
                sprintf(
                    'Error Connecting to server(%s): %s ',
                    $errno,
                    $errstr
                ),
                $errno
            );
        }

        if (false === stream_socket_get_name($this->sock, true)) {
            throw new AMQPRuntimeException(
                sprintf(
                    'Connection refused: %s ',
                    $remote
                )
            );
        }

        list($sec, $uSec) = MiscHelper::splitSecondsMicroseconds($this->read_write_timeout);
        if (!stream_set_timeout($this->sock, $sec, $uSec)) {
            throw new AMQPIOException('Timeout could not be set');
        }

        // php cannot capture signals while streams are blocking
        if ($this->canDispatchPcntlSignal) {
            stream_set_blocking($this->sock, 0);
            stream_set_write_buffer($this->sock, 0);
            if (function_exists('stream_set_read_buffer')) {
                stream_set_read_buffer($this->sock, 0);
            }
        } else {
            stream_set_blocking($this->sock, 1);
        }

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

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

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

        while ($read < $len) {
            $this->check_heartbeat();

            if (!is_resource($this->sock) || feof($this->sock)) {
                throw new AMQPRuntimeException('Broken pipe or closed connection');
            }

            set_error_handler(array($this, 'error_handler'));
            $buffer = fread($this->sock, ($len - $read));
            restore_error_handler();

            if ($buffer === false) {
                throw new AMQPRuntimeException('Error receiving data');
            }

            if ($buffer === '') {
                if ($this->canDispatchPcntlSignal) {
                    // prevent cpu from being consumed while waiting
                    if ($this->canSelectNull) {
                        $this->select($this->heartbeat ?  $this->heartbeat + 1 : null, null);
                        pcntl_signal_dispatch();
                    } else {
                        usleep(100000);
                        pcntl_signal_dispatch();
                    }
                }
                continue;
            }

            $read += mb_strlen($buffer, 'ASCII');
            $data .= $buffer;
        }

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

        $this->last_read = microtime(true);
        return $data;
    }

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

        while ($written < $len) {
            if ($buffer !== 0 && $this->read_write_timeout) {
                $tw = time();
            }

            if (!is_resource($this->sock)) {
                throw new AMQPRuntimeException('Broken pipe or closed connection');
            }

            set_error_handler(array($this, 'error_handler'));
            // OpenSSL's C library function SSL_write() can balk on buffers > 8192
            // bytes in length, so we're limiting the write size here. On both TLS
            // and plaintext connections, the write loop will continue until the
            // buffer has been fully written.
            // This behavior has been observed in OpenSSL dating back to at least
            // September 2002:
            // http://comments.gmane.org/gmane.comp.encryption.openssl.user/4361
            $buffer = fwrite($this->sock, $data, 8192);
            restore_error_handler();

            if ($buffer === false) {
                throw new AMQPRuntimeException('Error sending data');
            }

            if ($buffer === 0 && feof($this->sock)) {
                throw new AMQPRuntimeException('Broken pipe or closed connection');
            }

            if ($buffer === 0 && $tw && (time() - $tw > $this->read_write_timeout)) {
0   == false // true
0   == null  // true
123 == false // false
123 == null  // false

// It is often better to use strict comparison
0 === false // false
0 === null  // false
                throw new AMQPTimeoutException('Error sending data. write timed out');
            }

            if ($this->timed_out()) {
                throw new AMQPTimeoutException('Error sending data. Socket connection timed out');
            }

            $written += $buffer;

            if ($buffer > 0) {
                $data = substr($data, $buffer, strlen($data) - $buffer);
            }
        }

        $this->last_write = microtime(true);
    }

    /**
     * Internal error handler to deal with stream and socket errors that need to be ignored
     *
     * @param  int $errno
     * @param  string $errstr
     * @param  string $errfile
     * @param  int $errline
     * @param  array $errcontext
     * @return null
     * @throws \ErrorException
     */
    public function error_handler($errno, $errstr, $errfile, $errline, $errcontext = null)
    {
        $this->last_error = compact('errno', 'errstr', 'errfile', 'errline', 'errcontext');

        // fwrite notice that the stream isn't ready
        if (strstr($errstr, 'Resource temporarily unavailable')) {
             // it's allowed to retry
            return null;
        }

        // stream_select warning that it has been interrupted by a signal
        if (strstr($errstr, 'Interrupted system call')) {
             // it's allowed while processing signals
            return null;
        }

        restore_error_handler();

        // raise all other issues to exceptions
        throw new \ErrorException($errstr, 0, $errno, $errfile, $errline);
    }

    /**
     * Heartbeat logic: check connection health here
     */
    protected function check_heartbeat()
    {
        // ignore unless heartbeat interval is set
        if ($this->heartbeat !== 0 && $this->last_read && $this->last_write) {
            $t = microtime(true);
            $t_read = round($t - $this->last_read);
            $t_write = round($t - $this->last_write);

            // server has gone away
            if (($this->heartbeat * 2) < $t_read) {
                $this->reconnect();
            }

            // time for client to send a heartbeat
            if (($this->heartbeat / 2) < $t_write) {
                $this->write_heartbeat();
            }
        }
    }

    /**
     * Sends a heartbeat message
     */
    protected function write_heartbeat()
    {
        $pkt = new AMQPWriter();
        $pkt->write_octet(8);
        $pkt->write_short(0);
        $pkt->write_long(0);
        $pkt->write_octet(0xCE);
        $this->write($pkt->getvalue());
    }

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

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

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

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

if (rand(1, 6) > 3) {
    $higher = true;
} else {
    $higher = false;
}

        set_error_handler(array($this, 'error_handler'));
        $result = stream_select($read, $write, $except, $sec, $usec);
        restore_error_handler();

        return $result;
    }

    /**
     * @return mixed
     */
    protected function timed_out()
    {
        // get status of socket to determine whether or not it has timed out
        $info = stream_get_meta_data($this->sock);

        return $info['timed_out'];
    }

    /**
     * @throws \PhpAmqpLib\Exception\AMQPIOException
     */
    protected function enable_keepalive()
    {
        if (!function_exists('socket_import_stream')) {
            throw new AMQPIOException('Can not enable keepalive: function socket_import_stream does not exist');
        }

        if (!defined('SOL_SOCKET') || !defined('SO_KEEPALIVE')) {
            throw new AMQPIOException('Can not enable keepalive: SOL_SOCKET or SO_KEEPALIVE is not defined');
        }

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


1		<?php
2		namespace PhpAmqpLib\Wire\IO;
3
4		use PhpAmqpLib\Exception\AMQPIOException;
5		use PhpAmqpLib\Exception\AMQPRuntimeException;
6		use PhpAmqpLib\Exception\AMQPTimeoutException;
7		use PhpAmqpLib\Helper\MiscHelper;
8		use PhpAmqpLib\Wire\AMQPWriter;
9
10		class StreamIO extends AbstractIO
11		{
12		/** @var string */
13		protected $protocol;
14
15		/** @var string */
16		protected $host;
17
18		/** @var int */
19		protected $port;
20
21		/** @var int */
22		protected $connection_timeout;
23
24		/** @var int */
25		protected $read_write_timeout;
26
27		/** @var resource */
28		protected $context;
29
30		/** @var bool */
31		protected $keepalive;
32
33		/** @var int */
34		protected $heartbeat;
35
36		/** @var float */
37		protected $last_read;
38
39		/** @var float */
40		protected $last_write;
41
42		/** @var array */
43		protected $last_error;
44
45		/** @var resource */
46		private $sock;
47
48		/** @var bool */
49		private $canSelectNull;
50
51		/** @var bool */
52		private $canDispatchPcntlSignal;
53
54		/**
55		* @param string $host
56		* @param int $port
57		* @param int $connection_timeout
58		* @param int $read_write_timeout
59		* @param null $context
60		* @param bool $keepalive
61		* @param int $heartbeat
62		*/
63	40	public function __construct(
64		$host,
65		$port,
66		$connection_timeout,
67		$read_write_timeout,
68		$context = null,
69		$keepalive = false,
70		$heartbeat = 0
71		) {
72	40	$this->protocol = 'tcp';
73	40	$this->host = $host;
74	40	$this->port = $port;
75	40	$this->connection_timeout = $connection_timeout;
76	40	$this->read_write_timeout = $read_write_timeout;
77	40	$this->context = $context;
78	40	$this->keepalive = $keepalive;
79	40	$this->heartbeat = $heartbeat;
80	40	$this->canSelectNull = true;
81	40	$this->canDispatchPcntlSignal = $this->isPcntlSignalEnabled();
82
83	40	if (is_null($this->context)) {
84	40	$this->context = stream_context_create();
85	32	} else {
86		$this->protocol = 'ssl';
87		// php bugs 41631 & 65137 prevent select null from working on ssl streams
88		if (PHP_VERSION_ID < 50436) {
89		$this->canSelectNull = false;
90		}
91		}
92	40	}
93
94		/**
95		* @return bool
96		*/
97	40	private function isPcntlSignalEnabled()
98		{
99	40	return extension_loaded('pcntl')
100	40	&& function_exists('pcntl_signal_dispatch')
101	40	&& (defined('AMQP_WITHOUT_SIGNALS') && !AMQP_WITHOUT_SIGNALS);
102		}
103
104		/**
105		* Sets up the stream connection
106		*
107		* @throws \PhpAmqpLib\Exception\AMQPRuntimeException
108		* @throws \Exception
109		*/
110	40	public function connect()
111		{
112	40	$errstr = $errno = null;
113
114	40	$remote = sprintf(
115	40	'%s://%s:%s',
116	40	$this->protocol,
117	40	$this->host,
118	40	$this->port
119	32	);
120
121	40	set_error_handler(array($this, 'error_handler'));
122
123	40	$this->sock = stream_socket_client(
124	32	$remote,
125	32	$errno,
126	32	$errstr,
127	40	$this->connection_timeout,
128	40	STREAM_CLIENT_CONNECT,
129	40	$this->context
130	32	);
131
132	40	restore_error_handler();
133
134	40	if (false === $this->sock) {
135		throw new AMQPRuntimeException(
136		sprintf(
137		'Error Connecting to server(%s): %s ',
138		$errno,
139		$errstr
140		),
141		$errno
142		);
143		}
144
145	40	if (false === stream_socket_get_name($this->sock, true)) {
146		throw new AMQPRuntimeException(
147		sprintf(
148		'Connection refused: %s ',
149		$remote
150		)
151		);
152		}
153
154	40	list($sec, $uSec) = MiscHelper::splitSecondsMicroseconds($this->read_write_timeout);
155	40	if (!stream_set_timeout($this->sock, $sec, $uSec)) {
156		throw new AMQPIOException('Timeout could not be set');
157		}
158
159		// php cannot capture signals while streams are blocking
160	40	if ($this->canDispatchPcntlSignal) {
161		stream_set_blocking($this->sock, 0);
162		stream_set_write_buffer($this->sock, 0);
163		if (function_exists('stream_set_read_buffer')) {
164		stream_set_read_buffer($this->sock, 0);
165		}
166		} else {
167	40	stream_set_blocking($this->sock, 1);
168		}
169
170	40	if ($this->keepalive) {
171		$this->enable_keepalive();
172		}
173	40	}
174
175		/**
176		* Reconnects the socket
177		*/
178	10	public function reconnect()
179		{
180	10	$this->close();
181	10	$this->connect();
182	10	}
183
184		/**
185		* @param $len
186		* @throws \PhpAmqpLib\Exception\AMQPIOException
187		* @return mixed\|string
188		*/
189	40	public function read($len)
190		{
191	40	$read = 0;
192	40	$data = '';
193
194	40	while ($read < $len) {
195	40	$this->check_heartbeat();
196
197	40	if (!is_resource($this->sock) \|\| feof($this->sock)) {
198		throw new AMQPRuntimeException('Broken pipe or closed connection');
199		}
200
201	40	set_error_handler(array($this, 'error_handler'));
202	40	$buffer = fread($this->sock, ($len - $read));
203	40	restore_error_handler();
204
205	40	if ($buffer === false) {
206		throw new AMQPRuntimeException('Error receiving data');
207		}
208
209	40	if ($buffer === '') {
210		if ($this->canDispatchPcntlSignal) {
211		// prevent cpu from being consumed while waiting
212		if ($this->canSelectNull) {
213		$this->select($this->heartbeat ? $this->heartbeat + 1 : null, null);
214		pcntl_signal_dispatch();
215		} else {
216		usleep(100000);
217		pcntl_signal_dispatch();
218		}
219		}
220		continue;
221		}
222
223	40	$read += mb_strlen($buffer, 'ASCII');
224	40	$data .= $buffer;
225	32	}
226
227	40	if (mb_strlen($data, 'ASCII') !== $len) {
228		throw new AMQPRuntimeException(
229		sprintf(
230		'Error reading data. Received %s instead of expected %s bytes',
231		mb_strlen($data, 'ASCII'),
232		$len
233		)
234		);
235		}
236
237	40	$this->last_read = microtime(true);
238	40	return $data;
239		}
240
241		/**
242		* @param $data
243		* @return mixed\|void
244		* @throws \PhpAmqpLib\Exception\AMQPRuntimeException
245		* @throws \PhpAmqpLib\Exception\AMQPTimeoutException
246		*/
247	40	public function write($data)
248		{
249	40	$written = 0;
250	40	$buffer = false;
251		$tw = false;
252	40	$len = mb_strlen($data, 'ASCII');
253
254	40	while ($written < $len) {
255		if ($buffer !== 0 && $this->read_write_timeout) {
256		$tw = time();
257		}
258	40
259		if (!is_resource($this->sock)) {
260		throw new AMQPRuntimeException('Broken pipe or closed connection');
261		}
262
263		set_error_handler(array($this, 'error_handler'));
264		// OpenSSL's C library function SSL_write() can balk on buffers > 8192
265		// bytes in length, so we're limiting the write size here. On both TLS
266	40	// and plaintext connections, the write loop will continue until the
267	40	// buffer has been fully written.
268		// This behavior has been observed in OpenSSL dating back to at least
269	40	// September 2002:
270		// http://comments.gmane.org/gmane.comp.encryption.openssl.user/4361
271		$buffer = fwrite($this->sock, $data, 8192);
272		restore_error_handler();
273	40
274		if ($buffer === false) {
275		throw new AMQPRuntimeException('Error sending data');
276		}
277	40
278		if ($buffer === 0 && feof($this->sock)) {
279		throw new AMQPRuntimeException('Broken pipe or closed connection');
280		}
281	40
282		if ($buffer === 0 && $tw && (time() - $tw > $this->read_write_timeout)) {
		0 ignored issues – show Bug Best Practice introduced 2015-12-15 13:15 UTC by Report Bug Copy Issue Report The expression `$tw` of type `integer\|false` is loosely compared to `true`; this is ambiguous if the integer can be zero. You might want to explicitly use `!== null` instead. In PHP, under loose comparison (like `==`, or `!=`, or `switch` conditions), values of different types might be equal. For `integer` values, zero is a special case, in particular the following results might be unexpected: 0 == false // true 0 == null // true 123 == false // false 123 == null // false // It is often better to use strict comparison 0 === false // false 0 === null // false Loading history...
283	40	throw new AMQPTimeoutException('Error sending data. write timed out');
284	40	}
285	32
286	32	if ($this->timed_out()) {
287		throw new AMQPTimeoutException('Error sending data. Socket connection timed out');
288	40	}
289	40
290		$written += $buffer;
291
292		if ($buffer > 0) {
293		$data = substr($data, $buffer, strlen($data) - $buffer);
294		}
295		}
296
297		$this->last_write = microtime(true);
298		}
299
300		/**
301		* Internal error handler to deal with stream and socket errors that need to be ignored
302		*
303		* @param int $errno
304		* @param string $errstr
305		* @param string $errfile
306		* @param int $errline
307		* @param array $errcontext
308		* @return null
309		* @throws \ErrorException
310		*/
311		public function error_handler($errno, $errstr, $errfile, $errline, $errcontext = null)
312		{
313		$this->last_error = compact('errno', 'errstr', 'errfile', 'errline', 'errcontext');
314
315		// fwrite notice that the stream isn't ready
316		if (strstr($errstr, 'Resource temporarily unavailable')) {
317		// it's allowed to retry
318		return null;
319		}
320
321		// stream_select warning that it has been interrupted by a signal
322		if (strstr($errstr, 'Interrupted system call')) {
323		// it's allowed while processing signals
324		return null;
325		}
326
327	40	restore_error_handler();
328
329		// raise all other issues to exceptions
330	40	throw new \ErrorException($errstr, 0, $errno, $errfile, $errline);
331		}
332
333		/**
334		* Heartbeat logic: check connection health here
335		*/
336		protected function check_heartbeat()
337		{
338		// ignore unless heartbeat interval is set
339		if ($this->heartbeat !== 0 && $this->last_read && $this->last_write) {
340		$t = microtime(true);
341		$t_read = round($t - $this->last_read);
342		$t_write = round($t - $this->last_write);
343
344		// server has gone away
345	40	if (($this->heartbeat * 2) < $t_read) {
346		$this->reconnect();
347		}
348
349		// time for client to send a heartbeat
350		if (($this->heartbeat / 2) < $t_write) {
351		$this->write_heartbeat();
352		}
353		}
354		}
355
356		/**
357		* Sends a heartbeat message
358		*/
359		protected function write_heartbeat()
360	40	{
361		$pkt = new AMQPWriter();
362	40	$pkt->write_octet(8);
363	40	$pkt->write_short(0);
364	32	$pkt->write_long(0);
365	40	$pkt->write_octet(0xCE);
366	40	$this->write($pkt->getvalue());
367		}
368
369		public function close()
370		{
371		if (is_resource($this->sock)) {
372		fclose($this->sock);
373		}
374		$this->sock = null;
375		}
376
377		/**
378		* @return resource
379		*/
380		public function get_socket()
381		{
382		return $this->sock;
383		}
384
385		/**
386		* @return resource
387		*/
388		public function getSocket()
389		{
390		return $this->get_socket();
391		}
392
393		/**
394		* @param $sec
395		* @param $usec
396		* @return int\|mixed
397		*/
398		public function select($sec, $usec)
399		{
400		$read = array($this->sock);
401		$write = null;
402		$except = null;
403		$result = false;
		0 ignored issues – show Unused Code introduced 2015-11-27 00:01 UTC by Report Bug Copy Issue Report `$result` is not used, you could remove the assignment. This check looks for variable assignements that are either overwritten by other assignments or where the variable is not used subsequently. $myVar = 'Value'; $higher = false; if (rand(1, 6) > 3) { $higher = true; } else { $higher = false; } Both the `$myVar` assignment in line 1 and the `$higher` assignment in line 2 are dead. The first because `$myVar` is never used and the second because `$higher` is always overwritten for every possible time line. Loading history...
404
405		set_error_handler(array($this, 'error_handler'));
406	40	$result = stream_select($read, $write, $except, $sec, $usec);
407		restore_error_handler();
408
409	40	return $result;
410		}
411	40
412		/**
413		* @return mixed
414		*/
415		protected function timed_out()
416		{
417		// get status of socket to determine whether or not it has timed out
418		$info = stream_get_meta_data($this->sock);
419
420		return $info['timed_out'];
421		}
422
423		/**
424		* @throws \PhpAmqpLib\Exception\AMQPIOException
425		*/
426		protected function enable_keepalive()
427		{
428		if (!function_exists('socket_import_stream')) {
429		throw new AMQPIOException('Can not enable keepalive: function socket_import_stream does not exist');
430		}
431	4
432		if (!defined('SOL_SOCKET') \|\| !defined('SO_KEEPALIVE')) {
433		throw new AMQPIOException('Can not enable keepalive: SOL_SOCKET or SO_KEEPALIVE is not defined');
434		}
435
436		$socket = socket_import_stream($this->sock);
437		socket_set_option($socket, SOL_SOCKET, SO_KEEPALIVE, 1);
438		}
439		}
440

videlalvaro / php-amqplib

Pull Request — master (#348)

StreamIO::connect() B

Complexity

Size

Duplication

Code Coverage

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like