AbstractSocket::setupSocketResource() - Code Metrics - Inspection of "Alive connections between requests" - edefimov/async-sockets - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( c97103...9edfc5 )

by Evgenij

created 2017-03-12 20:32 UTC

AbstractSocket::setupSocketResource() B

↳ Parent: AbstractSocket

Complexity

Conditions	2
Paths	2

Size

Total Lines	27
Code Lines	15

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	16
CRAP Score	2

Importance

Changes

Metric	Value
dl	0
loc	27
c	0
b	0
f	0
ccs	16
cts	16
cp	1
rs	8.8571
cc	2
eloc	15
nc	2
nop	2
crap	2

<?php
/**
 * Async sockets
 *
 * @copyright Copyright (c) 2015-2017, Efimov Evgenij <[email protected]>
 *
 * This source file is subject to the MIT license that is bundled
 * with this source code in the file LICENSE.
 */

namespace AsyncSockets\Socket;

use AsyncSockets\Exception\BadResourceException;
use AsyncSockets\Exception\ConnectionException;
use AsyncSockets\Frame\FramePickerInterface;
use AsyncSockets\Socket\Io\Context;
use AsyncSockets\Socket\Io\DisconnectedIo;
use AsyncSockets\Socket\Io\IoInterface;

/**
 * Class AbstractSocket
 */
abstract class AbstractSocket implements SocketInterface
{
    /**
     * Tcp socket type
     */
    const SOCKET_TYPE_TCP = 'tcp';

    /**
     * Udp socket type
     */
    const SOCKET_TYPE_UDP = 'udp';

    /**
     * Unix socket type
     */
    const SOCKET_TYPE_UNIX = 'unix';

    /**
     * Unix datagram socket type
     */
    const SOCKET_TYPE_UDG = 'udg';

    /**
     * Unknown type of socket
     */
    const SOCKET_TYPE_UNKNOWN = '';

    /**
     * Array of socket resources
     *
     * @var resource[]
     */
    private static $resources;

    /**
     * Flag if we have shutdown function
     *
     * @var bool
     */
    private static $hasShutdownClearer = false;

    /**
     * Gracefully disconnects sockets when application terminating
     *
     * @return void
     */
    public static function shutdownSocketCleaner()
    {
        foreach (self::$resources as $resource) {
            try {
                $socket = self::fromResource($resource);
                if (!($socket instanceof PersistentClientSocket)) {
                    $socket->close();
                }
            } catch (BadResourceException $e) {
                // nothing required
            }
        }
    }

    /**
     * Creates socket object from given resource
     *
     * @param resource $resource Opened socket resource
     *
     * @return SocketInterface
     */
    public static function fromResource($resource)
    {
        $map = [
            'stream'            => 0,
            'persistent stream' => 1,
        ];

        $type = get_resource_type($resource);
        if (!isset($map[$type])) {
            throw BadResourceException::notSocketResource($type);
        }

        $isServer     = 0;
        $isPersistent = $map[$type];
        $localAddress = stream_socket_get_name($resource, false);
        if ($localAddress === false) {
            throw BadResourceException::canNotObtainLocalAddress();
        }

        try {
            if (($s = stream_socket_client(
                    $localAddress,
                    $errno,
                    $errstr,
                    (int) ini_get('default_socket_timeout'),
                    STREAM_CLIENT_CONNECT,
                    stream_context_get_default()
                )) !== false
            ) {
                $isServer = 1;
                fclose($s);
            }
        } catch (\Exception $e) {
            // do nothing
        } catch (\Throwable $e) {

            // do nothing
        }

        // [isServer][isPersistent] => class

        $objectMap = [
            0 => [
                0 => 'AsyncSockets\Socket\ClientSocket',
                1 => 'AsyncSockets\Socket\PersistentClientSocket',
            ],
            1 => [
                0 => 'AsyncSockets\Socket\ServerSocket',
                1 => 'AsyncSockets\Socket\ServerSocket',
            ]
        ];

        /** @var AbstractSocket $result */
        $result = new $objectMap[$isServer][$isPersistent]();
        $result->setupSocketResource($resource, (string) stream_socket_get_name($resource, true));

        return $result;
    }

    /**
     * This socket resource
     *
     * @var resource
     */
    private $resource;

    /**
     * I/O interface
     *
     * @var IoInterface
     */
    private $ioInterface;

    /**
     * Socket address
     *
     * @var string
     */
    private $remoteAddress;

    /**
     * Context for this socket
     *
     * @var Context
     */
    private $context;

    /**
     * AbstractSocket constructor.
     */
    public function __construct()
    {
        $this->setDisconnectedState();
        $this->context = new Context();

        if (!self::$hasShutdownClearer) {
            self::$hasShutdownClearer = true;
            register_shutdown_function(['AsyncSockets\Socket\AbstractSocket', 'shutdownSocketCleaner']);
        }
    }

    /**
     * Set disconnected state for socket
     *
     * @return void
     */
    private function setDisconnectedState()
    {
        $this->ioInterface = new DisconnectedIo($this);
    }

    /** {@inheritdoc} */
    public function open($address, $context = null)
    {
        $resource = $this->createSocketResource(
            $address,
            $context ?: stream_context_get_default()
        );

        $this->setupSocketResource($resource, $address);
    }

    /** {@inheritdoc} */
    public function close()
    {
        if ($this->resource) {
            unset(self::$resources[(string) $this->resource]);
            $this->setDisconnectedState();
            stream_socket_shutdown($this->resource, STREAM_SHUT_RDWR);
            fclose($this->resource);
            $this->resource      = null;
            $this->remoteAddress = null;
        }
    }

    /** {@inheritdoc} */
    public function read(FramePickerInterface $picker, $isOutOfBand = false)
    {
        try {
            return $this->ioInterface->read($picker, $this->context, $isOutOfBand);
        } catch (ConnectionException $e) {
            $this->setDisconnectedState();
            throw $e;
        }
    }

    /** {@inheritdoc} */
    public function write($data, $isOutOfBand = false)
    {
        try {
            return $this->ioInterface->write($data, $this->context, $isOutOfBand);
        } catch (ConnectionException $e) {
            $this->setDisconnectedState();
            throw $e;
        }
    }

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

    /**
     * @inheritDoc
     */
    public function __toString()
    {
        return $this->remoteAddress ?
            sprintf(
                '[#%s, %s]',
                preg_replace('/Resource id #(\d+)/i', '$1', (string) $this->resource),
                $this->remoteAddress
            ) :
            '[closed socket]';
    }

    /**
     * Create certain socket resource
     *
     * @param string   $address Network address to open in form transport://path:port
     * @param resource $context Valid stream context created by function stream_context_create or null
     *
     * @return resource
     */
    abstract protected function createSocketResource($address, $context);

    /**
     * Create I/O interface for socket
     *
     * @param string $type Type of this socket, one of SOCKET_TYPE_* consts
     * @param string $address Address passed to open method
     *
     * @return IoInterface
     */
    abstract protected function createIoInterface($type, $address);

    /**
     * Get current socket type
     *
     * @return string One of SOCKET_TYPE_* consts
     */
    private function resolveSocketType()
    {
        $info = stream_get_meta_data($this->resource);
        if (!isset($info['stream_type'])) {
            return self::SOCKET_TYPE_UNKNOWN;
        }

        $parts = explode('/', $info['stream_type']);
        $map   = [
            'tcp'  => self::SOCKET_TYPE_TCP,
            'udp'  => self::SOCKET_TYPE_UDP,
            'udg'  => self::SOCKET_TYPE_UDG,
            'unix' => self::SOCKET_TYPE_UNIX,
        ];

        $regexp = '#^('. implode('|', array_keys($map)) . ')_socket$#';
        foreach ($parts as $part) {
            if (preg_match($regexp, $part, $pockets)) {
                return $map[$pockets[1]];
            }
        }

        return self::SOCKET_TYPE_UNKNOWN;
    }

    /**
     * Set up internal data for given resource
     *
     * @param resource $resource Socket resource
     * @param string   $address Socket address
     *
     * @return void
     */
    private function setupSocketResource($resource, $address)
    {
        if (!is_resource($resource)) {
            throw new ConnectionException(
                $this,
                'Can not allocate socket resource.'
            );
        }

        $this->resource      = $resource;
        $this->remoteAddress = $address;

        // https://bugs.php.net/bug.php?id=51056
        stream_set_blocking($this->resource, 0);

        // https://bugs.php.net/bug.php?id=52602
        stream_set_timeout($this->resource, 0, 0);
        stream_set_chunk_size($this->resource, IoInterface::SOCKET_BUFFER_SIZE);

        $this->ioInterface = $this->createIoInterface(
            $this->resolveSocketType(),
            $address
        );

        self::$resources[(string) $resource] = $resource;
        $this->context->reset();
    }

    /**
     * @inheritDoc
     */
    public function isConnected()
    {
        return $this->ioInterface->isConnected();
    }
}


1		<?php
2		/**
3		* Async sockets
4		*
5		* @copyright Copyright (c) 2015-2017, Efimov Evgenij <[email protected]>
6		*
7		* This source file is subject to the MIT license that is bundled
8		* with this source code in the file LICENSE.
9		*/
10
11		namespace AsyncSockets\Socket;
12
13		use AsyncSockets\Exception\BadResourceException;
14		use AsyncSockets\Exception\ConnectionException;
15		use AsyncSockets\Frame\FramePickerInterface;
16		use AsyncSockets\Socket\Io\Context;
17		use AsyncSockets\Socket\Io\DisconnectedIo;
18		use AsyncSockets\Socket\Io\IoInterface;
19
20		/**
21		* Class AbstractSocket
22		*/
23		abstract class AbstractSocket implements SocketInterface
24		{
25		/**
26		* Tcp socket type
27		*/
28		const SOCKET_TYPE_TCP = 'tcp';
29
30		/**
31		* Udp socket type
32		*/
33		const SOCKET_TYPE_UDP = 'udp';
34
35		/**
36		* Unix socket type
37		*/
38		const SOCKET_TYPE_UNIX = 'unix';
39
40		/**
41		* Unix datagram socket type
42		*/
43		const SOCKET_TYPE_UDG = 'udg';
44
45		/**
46		* Unknown type of socket
47		*/
48		const SOCKET_TYPE_UNKNOWN = '';
49
50		/**
51		* Array of socket resources
52		*
53		* @var resource[]
54		*/
55		private static $resources;
56
57		/**
58		* Flag if we have shutdown function
59		*
60		* @var bool
61		*/
62		private static $hasShutdownClearer = false;
63
64		/**
65		* Gracefully disconnects sockets when application terminating
66		*
67		* @return void
68		*/
69		public static function shutdownSocketCleaner()
70		{
71		foreach (self::$resources as $resource) {
72		try {
73		$socket = self::fromResource($resource);
74		if (!($socket instanceof PersistentClientSocket)) {
75		$socket->close();
76		}
77		} catch (BadResourceException $e) {
78		// nothing required
79		}
80		}
81		}
82
83		/**
84		* Creates socket object from given resource
85		*
86		* @param resource $resource Opened socket resource
87		*
88		* @return SocketInterface
89		*/
90		public static function fromResource($resource)
91		{
92		$map = [
93		'stream' => 0,
94		'persistent stream' => 1,
95		];
96
97		$type = get_resource_type($resource);
98		if (!isset($map[$type])) {
99		throw BadResourceException::notSocketResource($type);
100		}
101
102		$isServer = 0;
103		$isPersistent = $map[$type];
104		$localAddress = stream_socket_get_name($resource, false);
105		if ($localAddress === false) {
106		throw BadResourceException::canNotObtainLocalAddress();
107		}
108
109		try {
110		if (($s = stream_socket_client(
111		$localAddress,
112		$errno,
113		$errstr,
114		(int) ini_get('default_socket_timeout'),
115		STREAM_CLIENT_CONNECT,
116		stream_context_get_default()
117		)) !== false
118		) {
119		$isServer = 1;
120		fclose($s);
121		}
122		} catch (\Exception $e) {
123		// do nothing
124		} catch (\Throwable $e) {
		0 ignored issues – show Bug introduced 2017-03-12 20:16 UTC by Report Bug Copy Issue Report The class `Throwable` does not exist. Did you forget a USE statement, or did you not list all dependencies? Scrutinizer analyzes your `composer.json`/`composer.lock` file if available to determine the classes, and functions that are defined by your dependencies. It seems like the listed class was neither found in your dependencies, nor was it found in the analyzed files in your repository. If you are using some other form of dependency management, you might want to disable this analysis. Loading history...
125		// do nothing
126		}
127
128		// [isServer][isPersistent] => class
		0 ignored issues – show Unused Code Comprehensibility introduced 2017-03-12 20:16 UTC by Report Bug Copy Issue Report `46%` of this comment could be valid code. Did you maybe forget this after debugging? Sometimes obsolete code just ends up commented out instead of removed. In this case it is better to remove the code once you have checked you do not need it. The code might also have been commented out for debugging purposes. In this case it is vital that someone uncomments it again or your project may behave in very unexpected ways in production. This check looks for comments that seem to be mostly valid code and reports them. Loading history...
129		$objectMap = [
130		0 => [
131		0 => 'AsyncSockets\Socket\ClientSocket',
132		1 => 'AsyncSockets\Socket\PersistentClientSocket',
133		],
134		1 => [
135		0 => 'AsyncSockets\Socket\ServerSocket',
136		1 => 'AsyncSockets\Socket\ServerSocket',
137		]
138		];
139
140		/** @var AbstractSocket $result */
141		$result = new $objectMap[$isServer][$isPersistent]();
142		$result->setupSocketResource($resource, (string) stream_socket_get_name($resource, true));
143
144		return $result;
145		}
146
147		/**
148		* This socket resource
149		*
150		* @var resource
151		*/
152		private $resource;
153
154		/**
155		* I/O interface
156		*
157		* @var IoInterface
158		*/
159		private $ioInterface;
160
161		/**
162		* Socket address
163		*
164		* @var string
165		*/
166		private $remoteAddress;
167
168		/**
169		* Context for this socket
170		*
171		* @var Context
172		*/
173		private $context;
174
175		/**
176		* AbstractSocket constructor.
177		*/
178	139	public function __construct()
179		{
180	139	$this->setDisconnectedState();
181	139	$this->context = new Context();
182
183	139	if (!self::$hasShutdownClearer) {
184	1	self::$hasShutdownClearer = true;
185	1	register_shutdown_function(['AsyncSockets\Socket\AbstractSocket', 'shutdownSocketCleaner']);
186	1	}
187	139	}
188
189		/**
190		* Set disconnected state for socket
191		*
192		* @return void
193		*/
194	139	private function setDisconnectedState()
195		{
196	139	$this->ioInterface = new DisconnectedIo($this);
197	139	}
198
199		/** {@inheritdoc} */
200	75	public function open($address, $context = null)
201		{
202	75	$resource = $this->createSocketResource(
203	75	$address,
204	63	$context ?: stream_context_get_default()
205	75	);
206
207	69	$this->setupSocketResource($resource, $address);
208	60	}
209
210		/** {@inheritdoc} */
211	12	public function close()
212		{
213	12	if ($this->resource) {
214	8	unset(self::$resources[(string) $this->resource]);
215	8	$this->setDisconnectedState();
216	8	stream_socket_shutdown($this->resource, STREAM_SHUT_RDWR);
217	8	fclose($this->resource);
218	8	$this->resource = null;
219	8	$this->remoteAddress = null;
220	8	}
221	12	}
222
223		/** {@inheritdoc} */
224	31	public function read(FramePickerInterface $picker, $isOutOfBand = false)
225		{
226		try {
227	31	return $this->ioInterface->read($picker, $this->context, $isOutOfBand);
228	10	} catch (ConnectionException $e) {
229	5	$this->setDisconnectedState();
230	5	throw $e;
231		}
232		}
233
234		/** {@inheritdoc} */
235	10	public function write($data, $isOutOfBand = false)
236		{
237		try {
238	10	return $this->ioInterface->write($data, $this->context, $isOutOfBand);
239	10	} catch (ConnectionException $e) {
240	5	$this->setDisconnectedState();
241	5	throw $e;
242		}
243		}
244
245		/** {@inheritdoc} */
246	45	public function getStreamResource()
247		{
248	45	return $this->resource;
249		}
250
251		/**
252		* @inheritDoc
253		*/
254	10	public function __toString()
255		{
256	10	return $this->remoteAddress ?
257	10	sprintf(
258	5	'[#%s, %s]',
259	5	preg_replace('/Resource id #(\d+)/i', '$1', (string) $this->resource),
260	5	$this->remoteAddress
261	5	) :
262	10	'[closed socket]';
263		}
264
265		/**
266		* Create certain socket resource
267		*
268		* @param string $address Network address to open in form transport://path:port
269		* @param resource $context Valid stream context created by function stream_context_create or null
270		*
271		* @return resource
272		*/
273		abstract protected function createSocketResource($address, $context);
274
275		/**
276		* Create I/O interface for socket
277		*
278		* @param string $type Type of this socket, one of SOCKET_TYPE_* consts
279		* @param string $address Address passed to open method
280		*
281		* @return IoInterface
282		*/
283		abstract protected function createIoInterface($type, $address);
284
285		/**
286		* Get current socket type
287		*
288		* @return string One of SOCKET_TYPE_* consts
289		*/
290	64	private function resolveSocketType()
291		{
292	64	$info = stream_get_meta_data($this->resource);
293	64	if (!isset($info['stream_type'])) {
294	4	return self::SOCKET_TYPE_UNKNOWN;
295		}
296
297	60	$parts = explode('/', $info['stream_type']);
298		$map = [
299	60	'tcp' => self::SOCKET_TYPE_TCP,
300	60	'udp' => self::SOCKET_TYPE_UDP,
301	60	'udg' => self::SOCKET_TYPE_UDG,
302	60	'unix' => self::SOCKET_TYPE_UNIX,
303	60	];
304
305	60	$regexp = '#^('. implode('\|', array_keys($map)) . ')_socket$#';
306	60	foreach ($parts as $part) {
307	60	if (preg_match($regexp, $part, $pockets)) {
308	45	return $map[$pockets[1]];
309		}
310	15	}
311
312	15	return self::SOCKET_TYPE_UNKNOWN;
313		}
314
315		/**
316		* Set up internal data for given resource
317		*
318		* @param resource $resource Socket resource
319		* @param string $address Socket address
320		*
321		* @return void
322		*/
323	69	private function setupSocketResource($resource, $address)
324		{
325	69	if (!is_resource($resource)) {
326	5	throw new ConnectionException(
327	5	$this,
328		'Can not allocate socket resource.'
329	5	);
330		}
331
332	64	$this->resource = $resource;
333	64	$this->remoteAddress = $address;
334
335		// https://bugs.php.net/bug.php?id=51056
336	64	stream_set_blocking($this->resource, 0);
337
338		// https://bugs.php.net/bug.php?id=52602
339	64	stream_set_timeout($this->resource, 0, 0);
340	64	stream_set_chunk_size($this->resource, IoInterface::SOCKET_BUFFER_SIZE);
341
342	64	$this->ioInterface = $this->createIoInterface(
343	64	$this->resolveSocketType(),
344		$address
345	64	);
346
347	60	self::$resources[(string) $resource] = $resource;
348	60	$this->context->reset();
349	60	}
350
351		/**
352		* @inheritDoc
353		*/
354	8	public function isConnected()
355		{
356	8	return $this->ioInterface->isConnected();
357		}
358		}
359

edefimov / async-sockets

Push — master ( c97103...9edfc5 )

AbstractSocket::setupSocketResource() B

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like