AbstractOobHandler::handleTransferCounter() - Code Metrics - Inspection of "Refactored some methods" - edefimov/async-sockets - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Branch — 0.4-dev (25e57a)

by Evgenij

created 2017-01-15 13:38 UTC

AbstractOobHandler::handleTransferCounter() B

↳ Parent: AbstractOobHandler

Complexity

Conditions	4
Paths	7

Size

Total Lines	32
Code Lines	20

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	7
CRAP Score	4.0312

Importance

Changes

Metric	Value
dl	0
loc	32
rs	8.5806
c	0
b	0
f	0
ccs	7
cts	8
cp	0.875
cc	4
eloc	20
nc	7
nop	3
crap	4.0312

<?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\RequestExecutor\Pipeline;

use AsyncSockets\Event\ReadEvent;
use AsyncSockets\Frame\RawFramePicker;
use AsyncSockets\Operation\OperationInterface;
use AsyncSockets\RequestExecutor\EventHandlerInterface;
use AsyncSockets\RequestExecutor\IoHandlerInterface;
use AsyncSockets\RequestExecutor\Metadata\RequestDescriptor;
use AsyncSockets\RequestExecutor\Metadata\SpeedRateCounter;
use AsyncSockets\RequestExecutor\RequestExecutorInterface;

/**
 * Class AbstractOobHandler
 */
abstract class AbstractOobHandler implements IoHandlerInterface
{
    /**
     * Operation to descriptor state map
     *
     * @var array
     */
    private static $stateMap = [
        OperationInterface::OPERATION_READ  => RequestDescriptor::RDS_READ,
        OperationInterface::OPERATION_WRITE => RequestDescriptor::RDS_WRITE,
    ];

    /**
     * {@inheritdoc}
     */
    final public function handle(
        RequestDescriptor $descriptor,
        RequestExecutorInterface $executor,
        EventHandlerInterface $eventHandler
    ) {
        $result = $this->handleOobData($descriptor, $executor, $eventHandler);
        if ($result) {
            return $result;
        }

        $state = $this->getInvokeState($descriptor->getOperation());
        if ($descriptor->hasState($state)) {
            $descriptor->clearState($state);

            $result = $this->handleOperation($descriptor, $executor, $eventHandler);
        }

        return $result;
    }

    /**
     * Process given operation
     *
     * @param RequestDescriptor        $descriptor Request descriptor
     * @param RequestExecutorInterface $executor Executor, processing operation
     * @param EventHandlerInterface    $eventHandler Event handler for this operation
     *
     * @return OperationInterface|null Next operation to pass in socket. Return null,
     *      if next operation is not required. Return $operation parameter, if operation is not completed yet
     */
    abstract protected function handleOperation(
        RequestDescriptor $descriptor,
        RequestExecutorInterface $executor,
        EventHandlerInterface $eventHandler
    );

    /**
     * Return state which should invoke this operation
     *
     * @param OperationInterface $operation Operation object
     *
     * @return int RequestDescriptor::RDS_* constant
     */
    private function getInvokeState(OperationInterface $operation)
    {
        $type = $operation->getType();

        return isset(self::$stateMap[$type]) ? self::$stateMap[$type] : 0;
    }

    /**
     * Handle OOB data
     *
     * @param RequestDescriptor        $descriptor Request descriptor
     * @param RequestExecutorInterface $executor Executor, processing operation
     * @param EventHandlerInterface    $eventHandler Event handler for this operation
     *
     * @return OperationInterface|null Operation to return to user or null to continue normal processing
     */
    private function handleOobData(
        RequestDescriptor $descriptor,
        RequestExecutorInterface $executor,
        EventHandlerInterface $eventHandler
    ) {
        if (!$descriptor->hasState(RequestDescriptor::RDS_OOB)) {
            return null;
        }

        $descriptor->clearState(RequestDescriptor::RDS_OOB);

        $picker = new RawFramePicker();
        $socket = $descriptor->getSocket();
        $meta   = $descriptor->getMetadata();
        $frame  = $socket->read($picker, true);
        $event  = new ReadEvent(
            $executor,
            $socket,
            $meta[ RequestExecutorInterface::META_USER_CONTEXT ],
            $frame,
            true
        );

        $eventHandler->invokeEvent($event);

        return $event->getNextOperation();
    }

    /**
     * Creates transfer rate counter with given parameters
     *
     * @param string            $name Counter name to create
     * @param int               $minRate Minimum speed setting for counter
     * @param int               $duration Max duration of low speed
     * @param RequestDescriptor $descriptor Request descriptor for counter
     *
     * @return SpeedRateCounter
     */
    private function createRateCounter($name, $minRate, $duration, RequestDescriptor $descriptor)
    {
        $counter = $descriptor->getCounter($name);

        if (!$counter) {
            $meta    = $descriptor->getMetadata();
            $counter = new SpeedRateCounter($minRate, $duration);
            $time    = $meta[RequestExecutorInterface::META_LAST_IO_START_TIME] ?:
                        $meta[RequestExecutorInterface::META_CONNECTION_FINISH_TIME];
            $counter->advance($time, 0);
            $descriptor->registerCounter($name, $counter);
        }

        return $counter;
    }

    /**
     * Process transfer bytes counter
     *
     * @param string            $name Counter name operating this transfer
     * @param RequestDescriptor $descriptor Socket descriptor
     * @param int               $bytes Amount of bytes processed by transfer
     *
     * @return void
     */
    protected function handleTransferCounter($name, RequestDescriptor $descriptor, $bytes)
    {
        $meta = $descriptor->getMetadata();

        $info = $this->getTransferCounterMeta($name, $meta);

        $descriptor->resetCounter($info['resetCounter']);

        $meta = $descriptor->getMetadata();
        $descriptor->setMetadata($info['bytesCounter'], $meta[$info['bytesCounter']] + $bytes);

        $counter = $descriptor->getCounter($name);
        try {
            if (!$counter) {
                $counter = $this->createRateCounter($name, $info['minSpeed'], $info['duration'], $descriptor);
            }

            $counter->advance(microtime(true), $bytes);
            $descriptor->setMetadata(
                $info['speedCounter'],
                $counter->getCurrentSpeed() ?: $meta[$info['speedCounter']]
            );
        } catch (\OverflowException $e) {
            $callable = $info['exception'];

            throw $callable(
                $descriptor->getSocket(),
                $counter->getCurrentSpeed(),
                $counter->getCurrentDuration()
            );
        }
    }

    /**
     * Return metadata for transfer rate counter
     *
     * @param string $name Counter name
     * @param array  $meta Socket metadata
     *
     * @return array
     */
    private function getTransferCounterMeta($name, array $meta)
    {
        $map = [
            RequestDescriptor::COUNTER_RECV_MIN_RATE => [
                'resetCounter' => RequestDescriptor::COUNTER_SEND_MIN_RATE,
                'minSpeed'     => $meta[ RequestExecutorInterface::META_MIN_RECEIVE_SPEED ],
                'duration'     => $meta[ RequestExecutorInterface::META_MIN_RECEIVE_SPEED_DURATION ],
                'bytesCounter' => RequestExecutorInterface::META_BYTES_RECEIVED,
                'speedCounter' => RequestExecutorInterface::META_RECEIVE_SPEED,
                'exception'    => [ 'AsyncSockets\Exception\SlowSpeedTransferException', 'tooSlowDataReceiving' ],
            ],
            RequestDescriptor::COUNTER_SEND_MIN_RATE => [
                'resetCounter' => RequestDescriptor::COUNTER_RECV_MIN_RATE,
                'minSpeed'     => $meta[ RequestExecutorInterface::META_MIN_SEND_SPEED ],
                'duration'     => $meta[ RequestExecutorInterface::META_MIN_SEND_SPEED_DURATION ],
                'bytesCounter' => RequestExecutorInterface::META_BYTES_SENT,
                'speedCounter' => RequestExecutorInterface::META_SEND_SPEED,
                'exception'    => [ 'AsyncSockets\Exception\SlowSpeedTransferException', 'tooSlowDataSending' ],
            ],
        ];

        if (!isset($map[ $name ])) {
            throw new \LogicException('Can not process counter ' . $name . ' in transfer operation');
        }

        return $map[ $name ];
    }
}


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		namespace AsyncSockets\RequestExecutor\Pipeline;
11
12		use AsyncSockets\Event\ReadEvent;
13		use AsyncSockets\Frame\RawFramePicker;
14		use AsyncSockets\Operation\OperationInterface;
15		use AsyncSockets\RequestExecutor\EventHandlerInterface;
16		use AsyncSockets\RequestExecutor\IoHandlerInterface;
17		use AsyncSockets\RequestExecutor\Metadata\RequestDescriptor;
18		use AsyncSockets\RequestExecutor\Metadata\SpeedRateCounter;
19		use AsyncSockets\RequestExecutor\RequestExecutorInterface;
20
21		/**
22		* Class AbstractOobHandler
23		*/
24		abstract class AbstractOobHandler implements IoHandlerInterface
25		{
26		/**
27		* Operation to descriptor state map
28		*
29		* @var array
30		*/
31		private static $stateMap = [
32		OperationInterface::OPERATION_READ => RequestDescriptor::RDS_READ,
33		OperationInterface::OPERATION_WRITE => RequestDescriptor::RDS_WRITE,
34		];
35
36		/**
37		* {@inheritdoc}
38		*/
39	70	final public function handle(
40		RequestDescriptor $descriptor,
41		RequestExecutorInterface $executor,
42		EventHandlerInterface $eventHandler
43		) {
44	70	$result = $this->handleOobData($descriptor, $executor, $eventHandler);
45	70	if ($result) {
46	5	return $result;
47		}
48
49	65	$state = $this->getInvokeState($descriptor->getOperation());
50	65	if ($descriptor->hasState($state)) {
51	60	$descriptor->clearState($state);
52
53	60	$result = $this->handleOperation($descriptor, $executor, $eventHandler);
54		}
55
56	41	return $result;
57		}
58
59		/**
60		* Process given operation
61		*
62		* @param RequestDescriptor $descriptor Request descriptor
63		* @param RequestExecutorInterface $executor Executor, processing operation
64		* @param EventHandlerInterface $eventHandler Event handler for this operation
65		*
66		* @return OperationInterface\|null Next operation to pass in socket. Return null,
67		* if next operation is not required. Return $operation parameter, if operation is not completed yet
68		*/
69		abstract protected function handleOperation(
70		RequestDescriptor $descriptor,
71		RequestExecutorInterface $executor,
72		EventHandlerInterface $eventHandler
73		);
74
75		/**
76		* Return state which should invoke this operation
77		*
78		* @param OperationInterface $operation Operation object
79		*
80		* @return int RequestDescriptor::RDS_* constant
81		*/
82	65	private function getInvokeState(OperationInterface $operation)
83		{
84	65	$type = $operation->getType();
85
86	65	return isset(self::$stateMap[$type]) ? self::$stateMap[$type] : 0;
87		}
88
89		/**
90		* Handle OOB data
91		*
92		* @param RequestDescriptor $descriptor Request descriptor
93		* @param RequestExecutorInterface $executor Executor, processing operation
94		* @param EventHandlerInterface $eventHandler Event handler for this operation
95		*
96		* @return OperationInterface\|null Operation to return to user or null to continue normal processing
97		*/
98	70	private function handleOobData(
99		RequestDescriptor $descriptor,
100		RequestExecutorInterface $executor,
101		EventHandlerInterface $eventHandler
102		) {
103	70	if (!$descriptor->hasState(RequestDescriptor::RDS_OOB)) {
104	60	return null;
105		}
106
107	10	$descriptor->clearState(RequestDescriptor::RDS_OOB);
108
109	10	$picker = new RawFramePicker();
110	10	$socket = $descriptor->getSocket();
111	10	$meta = $descriptor->getMetadata();
112	10	$frame = $socket->read($picker, true);
113	10	$event = new ReadEvent(
114		$executor,
115		$socket,
116	10	$meta[ RequestExecutorInterface::META_USER_CONTEXT ],
117		$frame,
118	10	true
119		);
120
121	10	$eventHandler->invokeEvent($event);
122
123	10	return $event->getNextOperation();
124		}
125
126		/**
127		* Creates transfer rate counter with given parameters
128		*
129		* @param string $name Counter name to create
130		* @param int $minRate Minimum speed setting for counter
131		* @param int $duration Max duration of low speed
132		* @param RequestDescriptor $descriptor Request descriptor for counter
133		*
134		* @return SpeedRateCounter
135		*/
136	46	private function createRateCounter($name, $minRate, $duration, RequestDescriptor $descriptor)
137		{
138	46	$counter = $descriptor->getCounter($name);
139
140	46	if (!$counter) {
141	46	$meta = $descriptor->getMetadata();
142	46	$counter = new SpeedRateCounter($minRate, $duration);
143	46	$time = $meta[RequestExecutorInterface::META_LAST_IO_START_TIME] ?:
144	46	$meta[RequestExecutorInterface::META_CONNECTION_FINISH_TIME];
145	46	$counter->advance($time, 0);
146	46	$descriptor->registerCounter($name, $counter);
147		}
148
149	46	return $counter;
150		}
151
152		/**
153		* Process transfer bytes counter
154		*
155		* @param string $name Counter name operating this transfer
156		* @param RequestDescriptor $descriptor Socket descriptor
157		* @param int $bytes Amount of bytes processed by transfer
158		*
159		* @return void
160		*/
161	46	protected function handleTransferCounter($name, RequestDescriptor $descriptor, $bytes)
162		{
163	46	$meta = $descriptor->getMetadata();
164
165		$info = $this->getTransferCounterMeta($name, $meta);
166
167		$descriptor->resetCounter($info['resetCounter']);
168
169		$meta = $descriptor->getMetadata();
170		$descriptor->setMetadata($info['bytesCounter'], $meta[$info['bytesCounter']] + $bytes);
171
172		$counter = $descriptor->getCounter($name);
173	46	try {
174		if (!$counter) {
175		$counter = $this->createRateCounter($name, $info['minSpeed'], $info['duration'], $descriptor);
176		}
177
178		$counter->advance(microtime(true), $bytes);
179		$descriptor->setMetadata(
180		$info['speedCounter'],
181		$counter->getCurrentSpeed() ?: $meta[$info['speedCounter']]
182		);
183		} catch (\OverflowException $e) {
184	46	$callable = $info['exception'];
185
186		throw $callable(
187		$descriptor->getSocket(),
188	46	$counter->getCurrentSpeed(),
189		$counter->getCurrentDuration()
190	46	);
191		}
192	46	}
193	46
194		/**
195	46	* Return metadata for transfer rate counter
196		*
197	46	* @param string $name Counter name
198	46	* @param array $meta Socket metadata
199		*
200		* @return array
201	46	*/
202	45	private function getTransferCounterMeta($name, array $meta)
203	45	{
204	45	$map = [
205		RequestDescriptor::COUNTER_RECV_MIN_RATE => [
206	1	'resetCounter' => RequestDescriptor::COUNTER_SEND_MIN_RATE,
207	1	'minSpeed' => $meta[ RequestExecutorInterface::META_MIN_RECEIVE_SPEED ],
208		'duration' => $meta[ RequestExecutorInterface::META_MIN_RECEIVE_SPEED_DURATION ],
209	1	'bytesCounter' => RequestExecutorInterface::META_BYTES_RECEIVED,
210	1	'speedCounter' => RequestExecutorInterface::META_RECEIVE_SPEED,
211	1	'exception' => [ 'AsyncSockets\Exception\SlowSpeedTransferException', 'tooSlowDataReceiving' ],
212	1	],
213		RequestDescriptor::COUNTER_SEND_MIN_RATE => [
214		'resetCounter' => RequestDescriptor::COUNTER_RECV_MIN_RATE,
215	45	'minSpeed' => $meta[ RequestExecutorInterface::META_MIN_SEND_SPEED ],
216		'duration' => $meta[ RequestExecutorInterface::META_MIN_SEND_SPEED_DURATION ],
217		'bytesCounter' => RequestExecutorInterface::META_BYTES_SENT,
218		'speedCounter' => RequestExecutorInterface::META_SEND_SPEED,
219		'exception' => [ 'AsyncSockets\Exception\SlowSpeedTransferException', 'tooSlowDataSending' ],
220		],
221		];
222
223		if (!isset($map[ $name ])) {
224		throw new \LogicException('Can not process counter ' . $name . ' in transfer operation');
225		}
226
227		return $map[ $name ];
228		}
229		}
230

edefimov / async-sockets

Branch — 0.4-dev (25e57a)

AbstractOobHandler::handleTransferCounter() B

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like