StreamedClientIo::canReachFrame() - Code Metrics - Inspection of "Bytes received counter added" - edefimov/async-sockets - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Branch — 0.4-dev (54e67b)

by Evgenij

created 2016-06-15 20:07 UTC

StreamedClientIo::canReachFrame() A

↳ Parent: StreamedClientIo

Complexity

Conditions	2
Paths	2

Size

Total Lines	4
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	2
CRAP Score	2

Importance

Changes	2
Bugs	0	Features	0

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

<?php
/**
 * Async sockets
 *
 * @copyright Copyright (c) 2015-2016, 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\Io;

use AsyncSockets\Frame\FramePickerInterface;

/**
 * Class StreamedClientIo
 */
class StreamedClientIo extends AbstractClientIo
{
    /**
     * Read attempts count
     */
    const READ_ATTEMPTS = 2;

    /**
     * Amount of read attempts
     *
     * @var int
     */
    private $readAttempts = self::READ_ATTEMPTS;

    /**
     * Remote socket address
     *
     * @var string
     */
    private $remoteAddress;

    /** {@inheritdoc} */
    protected function readRawDataIntoPicker(FramePickerInterface $picker, $isOutOfBand)
    {
        return $isOutOfBand ? $this->readOobData($picker) : $this->readRegularData($picker);
    }

    /**
     * Read OOB data from socket
     *
     * @param FramePickerInterface $picker
     *
     * @return string
     */
    private function readOobData(FramePickerInterface $picker)
    {
        $data = stream_socket_recvfrom(
            $this->socket->getStreamResource(),
            self::SOCKET_BUFFER_SIZE,
            STREAM_OOB
        );

        return $picker->pickUpData($data, $this->getRemoteAddress());
    }

    /**
     * Read regular data
     *
     * @param FramePickerInterface $picker Picker to read data into
     *
     * @return string
     */
    private function readRegularData(FramePickerInterface $picker)
    {
        // work-around https://bugs.php.net/bug.php?id=52602
        $resource         = $this->socket->getStreamResource();
        $readContext      = [
            'countCycles'       => 0,
            'dataBeforeIo'      => $this->getDataInSocket(),
            'isStreamDataEmpty' => false,
        ];

        do {
            $data = fread($resource, self::SOCKET_BUFFER_SIZE);
            $this->throwNetworkSocketExceptionIf($data === false, 'Failed to read data.', true);
            $isDataEmpty = $data === '';
            $result      = $picker->pickUpData($data, $this->getRemoteAddress());

            $readContext['countCycles']      += 1;
            $readContext['isStreamDataEmpty'] = $this->isReadDataActuallyEmpty($data);
            $this->readAttempts               = $this->resolveReadAttempts($readContext, $this->readAttempts);
        } while (!$picker->isEof() && !$isDataEmpty);

        return $result;
    }

    /**
     * Calculate attempts value
     *
     * @param array $context Read context
     * @param int   $currentAttempts Current attempts counter
     *
     * @return int
     */
    private function resolveReadAttempts(array $context, $currentAttempts)
    {
        return ($context['countCycles'] === 1 && empty($context['dataBeforeIo'])) ||
               ($context['countCycles'] > 1   && $context['isStreamDataEmpty']) ?
            $currentAttempts - 1 :
            self::READ_ATTEMPTS;

    }

    /** {@inheritdoc} */
    protected function writeRawData($data, $isOutOfBand)
    {
        $resource = $this->socket->getStreamResource();
        $test     = stream_socket_sendto($resource, '');
        $this->throwNetworkSocketExceptionIf($test !== 0, 'Failed to send data.', true);

        $written = $isOutOfBand ?
            $this->writeOobData($resource, $data) :
            fwrite($resource, $data, strlen($data));

        $this->throwNetworkSocketExceptionIf($written === false, 'Failed to send data.', true);

        if ($written === 0) {
            $this->throwExceptionIfNotConnected('Remote connection has been lost.');
        }

        return $written;
    }

    /** {@inheritdoc} */
    protected function isConnected()
    {
        return $this->resolveRemoteAddress() !== false;
    }

    /** {@inheritdoc} */
    protected function getRemoteAddress()
    {
        if ($this->remoteAddress === null) {
            $this->remoteAddress = $this->resolveRemoteAddress();
        }

        return $this->remoteAddress;
    }

    /**
     * Checks whether data read from stream buffer can be filled later
     *
     * @param string $data Read data
     *
     * @return bool
     */
    private function isReadDataActuallyEmpty($data)
    {
        $result = false;
        if ($data === '') {
            $dataInSocket = $this->getDataInSocket();
            $result       = $dataInSocket === '' || $dataInSocket === false;
        }

        return $result;
    }

    /** {@inheritdoc} */
    protected function canReachFrame()
    {
        return $this->readAttempts > 0 && $this->isConnected();
    }

    /**
     * Return first byte from socket buffer
     *
     * @return string
     */
    private function getDataInSocket()
    {
        return stream_socket_recvfrom($this->socket->getStreamResource(), 1, STREAM_PEEK);
    }

    /**
     * Return remote address if we connected or false otherwise
     *
     * @return string|bool
     */
    private function resolveRemoteAddress()
    {
        return stream_socket_get_name($this->socket->getStreamResource(), true);
    }

    /**
     * Write out-of-band data
     *
     * @param resource $socket Socket resource
     * @param string   $data Data to write
     *
     * @return int Amount of written bytes
     */
    private function writeOobData($socket, $data)
    {
        $result     = 0;
        $dataLength = strlen($data);
        for ($i = 0; $i < $dataLength; $i++) {
            $written = stream_socket_sendto($socket, $data[$i], STREAM_OOB);
            $this->throwNetworkSocketExceptionIf($written < 0, 'Failed to send data.', true);
            if ($written === 0) {
                break;
            }

            $result += $written;
        }

        return $result;
    }
}


1		<?php
2		/**
3		* Async sockets
4		*
5		* @copyright Copyright (c) 2015-2016, 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		namespace AsyncSockets\Socket\Io;
11
12		use AsyncSockets\Frame\FramePickerInterface;
13
14		/**
15		* Class StreamedClientIo
16		*/
17		class StreamedClientIo extends AbstractClientIo
18		{
19		/**
20		* Read attempts count
21		*/
22		const READ_ATTEMPTS = 2;
23
24		/**
25		* Amount of read attempts
26		*
27		* @var int
28		*/
29		private $readAttempts = self::READ_ATTEMPTS;
30
31		/**
32		* Remote socket address
33		*
34		* @var string
35		*/
36		private $remoteAddress;
37
38		/** {@inheritdoc} */
39	26	protected function readRawDataIntoPicker(FramePickerInterface $picker, $isOutOfBand)
40		{
41	26	return $isOutOfBand ? $this->readOobData($picker) : $this->readRegularData($picker);
42		}
43
44		/**
45		* Read OOB data from socket
46		*
47		* @param FramePickerInterface $picker
48		*
49		* @return string
50		*/
51	1	private function readOobData(FramePickerInterface $picker)
52		{
53	1	$data = stream_socket_recvfrom(
54	1	$this->socket->getStreamResource(),
55	1	self::SOCKET_BUFFER_SIZE,
56		STREAM_OOB
57	1	);
58
59	1	return $picker->pickUpData($data, $this->getRemoteAddress());
60		}
61
62		/**
63		* Read regular data
64		*
65		* @param FramePickerInterface $picker Picker to read data into
66		*
67		* @return string
68		*/
69	25	private function readRegularData(FramePickerInterface $picker)
70		{
71		// work-around https://bugs.php.net/bug.php?id=52602
72	25	$resource = $this->socket->getStreamResource();
73		$readContext = [
74	25	'countCycles' => 0,
75	25	'dataBeforeIo' => $this->getDataInSocket(),
76	25	'isStreamDataEmpty' => false,
77	25	];
78
79		do {
80	25	$data = fread($resource, self::SOCKET_BUFFER_SIZE);
81	25	$this->throwNetworkSocketExceptionIf($data === false, 'Failed to read data.', true);
82	24	$isDataEmpty = $data === '';
83	24	$result = $picker->pickUpData($data, $this->getRemoteAddress());
84
85	24	$readContext['countCycles'] += 1;
86	24	$readContext['isStreamDataEmpty'] = $this->isReadDataActuallyEmpty($data);
87	24	$this->readAttempts = $this->resolveReadAttempts($readContext, $this->readAttempts);
88	24	} while (!$picker->isEof() && !$isDataEmpty);
89
90	24	return $result;
91		}
92
93		/**
94		* Calculate attempts value
95		*
96		* @param array $context Read context
97		* @param int $currentAttempts Current attempts counter
98		*
99		* @return int
100		*/
101	24	private function resolveReadAttempts(array $context, $currentAttempts)
102		{
103	24	return ($context['countCycles'] === 1 && empty($context['dataBeforeIo'])) \|\|
104	11	($context['countCycles'] > 1 && $context['isStreamDataEmpty']) ?
105	24	$currentAttempts - 1 :
106	24	self::READ_ATTEMPTS;
107
108		}
109
110		/** {@inheritdoc} */
111	5	protected function writeRawData($data, $isOutOfBand)
112		{
113	5	$resource = $this->socket->getStreamResource();
114	5	$test = stream_socket_sendto($resource, '');
115	5	$this->throwNetworkSocketExceptionIf($test !== 0, 'Failed to send data.', true);
116
117		$written = $isOutOfBand ?
118	4	$this->writeOobData($resource, $data) :
119	4	fwrite($resource, $data, strlen($data));
120
121	4	$this->throwNetworkSocketExceptionIf($written === false, 'Failed to send data.', true);
122
123	3	if ($written === 0) {
124	2	$this->throwExceptionIfNotConnected('Remote connection has been lost.');
125	1	}
126
127	2	return $written;
128		}
129
130		/** {@inheritdoc} */
131	37	protected function isConnected()
132		{
133	37	return $this->resolveRemoteAddress() !== false;
134		}
135
136		/** {@inheritdoc} */
137	25	protected function getRemoteAddress()
138		{
139	25	if ($this->remoteAddress === null) {
140	25	$this->remoteAddress = $this->resolveRemoteAddress();
141	25	}
142
143	25	return $this->remoteAddress;
144		}
145
146		/**
147		* Checks whether data read from stream buffer can be filled later
148		*
149		* @param string $data Read data
150		*
151		* @return bool
152		*/
153	24	private function isReadDataActuallyEmpty($data)
154		{
155	24	$result = false;
156	24	if ($data === '') {
157	5	$dataInSocket = $this->getDataInSocket();
158	5	$result = $dataInSocket === '' \|\| $dataInSocket === false;
159	5	}
160
161	24	return $result;
162		}
163
164		/** {@inheritdoc} */
165	2	protected function canReachFrame()
166		{
167	2	return $this->readAttempts > 0 && $this->isConnected();
168		}
169
170		/**
171		* Return first byte from socket buffer
172		*
173		* @return string
174		*/
175	25	private function getDataInSocket()
176		{
177	25	return stream_socket_recvfrom($this->socket->getStreamResource(), 1, STREAM_PEEK);
178		}
179
180		/**
181		* Return remote address if we connected or false otherwise
182		*
183		* @return string\|bool
184		*/
185	37	private function resolveRemoteAddress()
186		{
187	37	return stream_socket_get_name($this->socket->getStreamResource(), true);
188		}
189
190		/**
191		* Write out-of-band data
192		*
193		* @param resource $socket Socket resource
194		* @param string $data Data to write
195		*
196		* @return int Amount of written bytes
197		*/
198	1	private function writeOobData($socket, $data)
199		{
200	1	$result = 0;
201	1	$dataLength = strlen($data);
202	1	for ($i = 0; $i < $dataLength; $i++) {
203	1	$written = stream_socket_sendto($socket, $data[$i], STREAM_OOB);
204	1	$this->throwNetworkSocketExceptionIf($written < 0, 'Failed to send data.', true);
205	1	if ($written === 0) {
206		break;
207		}
208
209	1	$result += $written;
210	1	}
211
212	1	return $result;
213		}
214		}
215

edefimov / async-sockets

Branch — 0.4-dev (54e67b)

StreamedClientIo::canReachFrame() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like