Connection::onRead() - Code Metrics - Inspection of "The AMQP Protocol implementation" - kakserpom/phpdaemon - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#286)

unknown

created 2018-01-21 18:20 UTC

Connection::onRead() D

↳ Parent: Connection

Complexity

Conditions	31
Paths	1

Size

Total Lines	175
Code Lines	109

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	31
eloc	109
nc	1
nop	0
dl	0
loc	175
rs	4.3983
c	0
b	0
f	0

How to fix Long Method Complexity

<?php

namespace PHPDaemon\Clients\AMQP;

use PHPDaemon\Clients\AMQP\Driver\ConnectionOptions;
use PHPDaemon\Clients\AMQP\Driver\Exception\AMQPConnectionException;
use PHPDaemon\Clients\AMQP\Driver\Features;
use PHPDaemon\Clients\AMQP\Driver\PackageInfo;
use PHPDaemon\Clients\AMQP\Driver\Protocol\CommandInterface;
use PHPDaemon\Clients\AMQP\Driver\Protocol\v091\Parser\Frame;
use PHPDaemon\Clients\AMQP\Driver\Protocol\v091\Parser\Frame as FrameParser;
use PHPDaemon\Clients\AMQP\Driver\Protocol\v091\Parser\Table as TableParser;
use PHPDaemon\Clients\AMQP\Driver\Protocol\v091\Protocol\Connection\ConnectionCloseFrame;
use PHPDaemon\Clients\AMQP\Driver\Protocol\v091\Protocol\Connection\ConnectionOpenFrame;
use PHPDaemon\Clients\AMQP\Driver\Protocol\v091\Protocol\Connection\ConnectionOpenOkFrame;
use PHPDaemon\Clients\AMQP\Driver\Protocol\v091\Protocol\Connection\ConnectionStartFrame;
use PHPDaemon\Clients\AMQP\Driver\Protocol\v091\Protocol\Connection\ConnectionStartOkFrame;
use PHPDaemon\Clients\AMQP\Driver\Protocol\v091\Protocol\Connection\ConnectionTuneFrame;
use PHPDaemon\Clients\AMQP\Driver\Protocol\v091\Protocol\Connection\ConnectionTuneOkFrame;
use PHPDaemon\Clients\AMQP\Driver\Protocol\v091\Protocol\HeartbeatFrame;
use PHPDaemon\Clients\AMQP\Driver\Protocol\v091\Protocol\OutgoingFrame;
use PHPDaemon\Clients\AMQP\Driver\Protocol\v091\Serializer\Frame as FrameSerializer;
use PHPDaemon\Clients\AMQP\Driver\Protocol\v091\Serializer\Table as TableSerializer;
use PHPDaemon\Network\ClientConnection;
use PHPDaemon\Utils\Binary;

/**
 * Class Connection
 * @author Aleksey I. Kuleshov YOU GLOBAL LIMITED
 * @package PHPDaemon\Clients\AMQP
 */
class Connection extends ClientConnection implements CommandInterface
{

    /**
     * The AMQP protocol header is sent by the client before any frame-based
     * communication takes place.  It is the only data transferred that is not
     * a frame.
     */
    const PROTOCOL_HEADER = "AMQP\x00\x00\x09\x01";

    /**
     * The maximum number of channels.
     *
     * AMQP channel ID is 2 bytes, but zero is reserved for connection-level
     * communication.
     */
    const MAXIMUM_CHANNELS = 0xffff - 1;

    /**
     * The maximum frame size the client supports.
     *
     * Note: RabbitMQ's default is 0x20000 (128 KB), our limit is higher to
     * allow for some server-side configurability.
     */
    const MAXIMUM_FRAME_SIZE = 0x80000; // 512 KB

    /**
     * The broker sends channelMax of zero in the tune frame if it does not
     * impose a channel limit.
     */
    const UNLIMITED_CHANNELS = 0;

    /**
     * The broker sends frameMax of zero in the tune frame if it does not impose
     * a frame size limit.
     */
    const UNLIMITED_FRAME_SIZE = 0;

    /**
     * The broker sends a heartbeat of zero in the tune frame if it does not use
     * heartbeats.
     */
    const HEARTBEAT_DISABLED = 0;

    /**
     * Event raised when protocol handshake ready
     */
    const EVENT_ON_HANDSHAKE = 'event.amqp.connection.handshake';

    /**
     * Event raised when connection close frame incoming
     */

    const EVENT_ON_CONNECTION_CLOSE = 'event.amqp.connection.close';

    /**
     * @var FrameParser
     */
    protected $parser;

    /**
     * @var FrameSerializer
     */
    protected $serializer;

    /**
     * @var ConnectionOptions
     */
    protected $connectionOptions;

    /**
     * @var bool
     */
    protected $isHandshaked = false;

    /**
     * The broker's supported features.
     * @var Features
     */
    private $features;

    /**
     * @var int
     */
    private $maximumChannelCount;

    /**
     * @var int
     */
    private $maximumFrameSize;

    /**
     * @var array
     */
    private $channels = [];

    /**
     * @var int
     */
    private $nextChannelId = 1;

    /**
     * @var bool
     */
    private $debug = false;

    /**
     *
     */
    protected function init()
    {
        $this->parser = new FrameParser(new TableParser());
        $this->serializer = new FrameSerializer(new TableSerializer());

        $this->connectionOptions = new ConnectionOptions(
            $this->pool->config->host->value,
            $this->pool->config->port->value,
            $this->pool->config->username->value,
            $this->pool->config->password->value,
            $this->pool->config->vhost->value
        );

        $this->debug = isset($this->pool->config->debug);
    }

    /**
     *
     */
    public function onReady()
    {
        if ($this->isHandshaked) {
            parent::onReady();
        }

        $this->write(self::PROTOCOL_HEADER);

        parent::onReady();
    }

    /**
     *
     * @throws \PHPDaemon\Clients\AMQP\Driver\Protocol\Exception\AMQPProtocolException
     * @throws \PHPDaemon\Clients\AMQP\Driver\Exception\AMQPConnectionException
     * @throws \InvalidArgumentException
     */
    protected function onRead()
    {
        if ($this->getInputLength() <= 0) {
            return;
        }
        // set busy for connection
        $this->busy = true;

        /**
         * 1) read Frame::HEADER_SIZE
         * 2) get payload length from header
         * 3) concatenate header and payload + 1 byte (for Constants::FRAME_END) into buffer
         * 4) parse buffer
         * 5) do stuff...
         * 6) if bev contains more data go to 1
         */
        frame:
        $header = $this->readExact(Frame::HEADER_SIZE);
        $framePayloadSize = Binary::b2i(substr($header, Frame::HEADER_TYPE_SIZE + Frame::HEADER_CHANNEL_SIZE, Frame::HEADER_PAYLOAD_LENGTH_SIZE));


        $buffer = $header . $this->readExact($framePayloadSize + 1);

        $frame = $this->parser->feed($buffer);
        if ($frame === null) {
            return;
        }
        if ($this->debug) {
            $this->log(sprintf('[AMQP] %s packet received', get_class($frame)));
        }

        if (!$this->isHandshaked) {
            switch (true) {
                case $frame instanceof ConnectionStartFrame:
                    if ($frame->versionMajor !== 0 || $frame->versionMinor !== 9) {
                        throw AMQPConnectionException::handshakeFailed(
                            $this->connectionOptions,
                            sprintf(
                                'the broker reported an unexpected AMQP version (v%d.%d)',
                                $frame->versionMajor,
                                $frame->versionMinor
                            )
                        );
                    }
                    if (!preg_match('/\bAMQPLAIN\b/', $frame->mechanisms)) {
                        throw AMQPConnectionException::handshakeFailed(
                            $this->connectionOptions,
                            'the AMQPLAIN authentication mechanism is not supported'
                        );
                    }

                    $this->features = new Features();
                    $properties = $frame->serverProperties;
                    if (isset($properties['product']) && 'RabbitMQ' === $properties['product']) {
                        $this->features->qosSizeLimits = false;
                    }
                    if (array_key_exists('capabilities', $properties)) {
                        if (array_key_exists('per_consumer_qos', $properties['capabilities'])) {
                            $this->features->perConsumerQos = (bool)$properties['capabilities']['per_consumer_qos'];
                        }
                        if (array_key_exists('exchange_exchange_bindings', $properties['capabilities'])) {
                            $this->features->exchangeToExchangeBindings = (bool)$properties['capabilities']['exchange_exchange_bindings'];

                        }
                    }

                    // Serialize credentials in "AMQPLAIN" format, which is essentially an
                    // AMQP table without the 4-byte size header ...
                    $user = $this->connectionOptions->getUsername();
                    $pass = $this->connectionOptions->getPassword();

                    $credentials = "\x05LOGINS" . pack('N', strlen($user)) . $user
                        . "\x08PASSWORDS" . pack('N', strlen($pass)) . $pass;

                    $this->command(ConnectionStartOkFrame::create(
                        [
                            'product' => $this->connectionOptions->getProductName(),
                            'version' => $this->connectionOptions->getProductVersion(),
                            'platform' => PackageInfo::AMQP_PLATFORM,
                            'copyright' => PackageInfo::AMQP_COPYRIGHT,
                            'information' => PackageInfo::AMQP_INFORMATION,

                        ],
                        'AMQPLAIN',
                        $credentials
                    ));
                    break;
                case $frame instanceof ConnectionTuneFrame:
                    $this->maximumChannelCount = self::MAXIMUM_CHANNELS;
                    if ($frame->channelMax === self::UNLIMITED_CHANNELS) {
                        $this->maximumChannelCount = self::MAXIMUM_CHANNELS;
                    } elseif ($frame->channelMax < self::MAXIMUM_CHANNELS) {
                        $this->maximumChannelCount = $frame->channelMax;
                    }

                    $this->maximumFrameSize = self::MAXIMUM_FRAME_SIZE;
                    if ($frame->frameMax === self::UNLIMITED_FRAME_SIZE) {
                        $this->maximumFrameSize = self::MAXIMUM_FRAME_SIZE;
                    } elseif ($frame->frameMax < self::MAXIMUM_FRAME_SIZE) {
                        $this->maximumFrameSize = $frame->frameMax;
                    }

                    $heartbeatInterval = 0;
                    if (!self::HEARTBEAT_DISABLED) {
                        $heartbeatInterval = $this->connectionOptions->getHeartbeatInterval();
                        if (null === $heartbeatInterval) {
                            $heartbeatInterval = $frame->heartbeat;
                        } elseif ($frame->heartbeat < $heartbeatInterval) {
                            $heartbeatInterval = $frame->heartbeat;
                        }
                    }

                    $outputFrame = ConnectionTuneOkFrame::create(
                        $this->maximumChannelCount,
                        $this->maximumFrameSize,
                        $heartbeatInterval
                    );

                    if ($outputFrame->heartbeat > 0) {
                        /**
                         * We need to set timeout value = ConnectionTuneFrame::heartbeat + 5 sec
                         */
                        $timeout = $outputFrame->heartbeat + 5;
                        $this->setTimeout($timeout);
                        $this->connectionOptions->setHeartbeatInterval($outputFrame->heartbeat);
                        $this->connectionOptions->setConnectionTimeout($timeout);
                    }

                    $this->command($outputFrame);

                    $outputFrame = ConnectionOpenFrame::create(
                        $this->connectionOptions->getVhost()
                    );
                    $this->command($outputFrame);
                    break;

                case $frame instanceof ConnectionOpenOkFrame:
                    $this->isHandshaked = true;
                    $this->openChannel(function ($channel) {
                        $this->trigger(self::EVENT_ON_HANDSHAKE, $channel);
                    });
                    break;
            }

            $this->checkFree();
            return;
        }

        switch (true) {
            case $frame instanceof HeartbeatFrame:
                $this->command($frame);
                break;
            case $frame instanceof ConnectionCloseFrame:
                $this->trigger(self::EVENT_ON_CONNECTION_CLOSE, $frame->replyCode, $frame->replyText);
                $this->close();
                return;
                break;
switch ($x) {
    case 1:
        return 'foo';
        break; // This break is not necessary and can be left off.
}
            default:
                if (isset($frame->frameChannelId)
                    && $frame->frameChannelId > 0
                    && array_key_exists($frame->frameChannelId, $this->channels)) {
                    /** @var Channel $channel */
                    $channel = $this->channels[$frame->frameChannelId];
                    $channel->trigger(get_class($frame), $frame);
                    break; // exit
                }

                $this->trigger(get_class($frame), $frame);
                break;
        }

        if ($this->bev && $this->getInputLength() > 0) {
            goto frame;
        }

        $this->checkFree();
    }

    /**
     * @param int $id
     * @param callable $callback
     * @throws \InvalidArgumentException
     * @throws \PHPDaemon\Clients\AMQP\Driver\Protocol\Exception\AMQPProtocolException
     * @throws \PHPDaemon\Clients\AMQP\Driver\Exception\AMQPConnectionException
     */
    public function getChannel(callable $callback, $id = 1)
    {
        if (count($this->channels) === 0) {
            $this->openChannel($callback);
            return;
        }

        if (is_callable($callback)) {
            $callback($this->channels[$id]);
        }
    }

    /**
     * @param callable|null $callback
     * @throws \InvalidArgumentException
     * @throws \PHPDaemon\Clients\AMQP\Driver\Protocol\Exception\AMQPProtocolException
     * @throws \PHPDaemon\Clients\AMQP\Driver\Exception\AMQPConnectionException
     */
    public function openChannel(callable $callback = null)
    {
        new Channel($this, $callback);
    }

    /**
     * @return int
     * @throws \PHPDaemon\Clients\AMQP\Driver\Exception\AMQPConnectionException
     */
    public function findChannelId()
    {
        // first check in range [next, max] ...
        for (

            $channelId = $this->nextChannelId;
            $channelId <= $this->maximumChannelCount;
            ++$channelId
        ) {
            if (!isset($this->channels[$channelId])) {
                $this->nextChannelId = $channelId + 1;

                return $channelId;
            }
        }

        // then check in range [min, next) ...
        for (

            $channelId = 1;
            $channelId < $this->nextChannelId;
            ++$channelId
        ) {
            if (!isset($this->channels[$channelId])) {
                $this->nextChannelId = $channelId + 1;

                return $channelId;
            }
        }

        throw new AMQPConnectionException('No available channels');
    }

    /**
     * @param OutgoingFrame $frame
     * @param callable|null $callback
     * @return bool
     * @throws \PHPDaemon\Clients\AMQP\Driver\Protocol\Exception\AMQPProtocolException
     * @throws \InvalidArgumentException
     */
    public function command(OutgoingFrame $frame, callable $callback = null)
    {
        if ($callback) {
            $this->onResponse($callback);
        }
        $serializedFrame = $this->serializer->serialize($frame);
        return $this->write($serializedFrame);
    }


    public function addChannel($id, Channel $channel)
    {
        $this->channels[$id] = $channel;
        $this->nextChannelId = max(array_keys($this->channels)) + 1;
        return $this;
    }

    /**
     * @return Features
     */
    public function getFeatures()
    {
        return $this->features;
    }

    /**
     * @return ConnectionOptions
     */
    public function getConnectionOptions()
    {
        return $this->connectionOptions;
    }

    /**
     * @return int
     */
    public function getMaximumChannelCount()
    {
        return $this->maximumChannelCount;
    }

    /**
     * @return int
     */
    public function getMaximumFrameSize()
    {
        return $this->maximumFrameSize;
    }

    /**
     * @return bool
     */
    public function isHandshaked()
    {
        return $this->isHandshaked;
    }
}


1		<?php
2
3		namespace PHPDaemon\Clients\AMQP;
4
5		use PHPDaemon\Clients\AMQP\Driver\ConnectionOptions;
6		use PHPDaemon\Clients\AMQP\Driver\Exception\AMQPConnectionException;
7		use PHPDaemon\Clients\AMQP\Driver\Features;
8		use PHPDaemon\Clients\AMQP\Driver\PackageInfo;
9		use PHPDaemon\Clients\AMQP\Driver\Protocol\CommandInterface;
10		use PHPDaemon\Clients\AMQP\Driver\Protocol\v091\Parser\Frame;
11		use PHPDaemon\Clients\AMQP\Driver\Protocol\v091\Parser\Frame as FrameParser;
12		use PHPDaemon\Clients\AMQP\Driver\Protocol\v091\Parser\Table as TableParser;
13		use PHPDaemon\Clients\AMQP\Driver\Protocol\v091\Protocol\Connection\ConnectionCloseFrame;
14		use PHPDaemon\Clients\AMQP\Driver\Protocol\v091\Protocol\Connection\ConnectionOpenFrame;
15		use PHPDaemon\Clients\AMQP\Driver\Protocol\v091\Protocol\Connection\ConnectionOpenOkFrame;
16		use PHPDaemon\Clients\AMQP\Driver\Protocol\v091\Protocol\Connection\ConnectionStartFrame;
17		use PHPDaemon\Clients\AMQP\Driver\Protocol\v091\Protocol\Connection\ConnectionStartOkFrame;
18		use PHPDaemon\Clients\AMQP\Driver\Protocol\v091\Protocol\Connection\ConnectionTuneFrame;
19		use PHPDaemon\Clients\AMQP\Driver\Protocol\v091\Protocol\Connection\ConnectionTuneOkFrame;
20		use PHPDaemon\Clients\AMQP\Driver\Protocol\v091\Protocol\HeartbeatFrame;
21		use PHPDaemon\Clients\AMQP\Driver\Protocol\v091\Protocol\OutgoingFrame;
22		use PHPDaemon\Clients\AMQP\Driver\Protocol\v091\Serializer\Frame as FrameSerializer;
23		use PHPDaemon\Clients\AMQP\Driver\Protocol\v091\Serializer\Table as TableSerializer;
24		use PHPDaemon\Network\ClientConnection;
25		use PHPDaemon\Utils\Binary;
26
27		/**
28		* Class Connection
29		* @author Aleksey I. Kuleshov YOU GLOBAL LIMITED
30		* @package PHPDaemon\Clients\AMQP
31		*/
32		class Connection extends ClientConnection implements CommandInterface
33		{
34
35		/**
36		* The AMQP protocol header is sent by the client before any frame-based
37		* communication takes place. It is the only data transferred that is not
38		* a frame.
39		*/
40		const PROTOCOL_HEADER = "AMQP\x00\x00\x09\x01";
41
42		/**
43		* The maximum number of channels.
44		*
45		* AMQP channel ID is 2 bytes, but zero is reserved for connection-level
46		* communication.
47		*/
48		const MAXIMUM_CHANNELS = 0xffff - 1;
49
50		/**
51		* The maximum frame size the client supports.
52		*
53		* Note: RabbitMQ's default is 0x20000 (128 KB), our limit is higher to
54		* allow for some server-side configurability.
55		*/
56		const MAXIMUM_FRAME_SIZE = 0x80000; // 512 KB
57
58		/**
59		* The broker sends channelMax of zero in the tune frame if it does not
60		* impose a channel limit.
61		*/
62		const UNLIMITED_CHANNELS = 0;
63
64		/**
65		* The broker sends frameMax of zero in the tune frame if it does not impose
66		* a frame size limit.
67		*/
68		const UNLIMITED_FRAME_SIZE = 0;
69
70		/**
71		* The broker sends a heartbeat of zero in the tune frame if it does not use
72		* heartbeats.
73		*/
74		const HEARTBEAT_DISABLED = 0;
75
76		/**
77		* Event raised when protocol handshake ready
78		*/
79		const EVENT_ON_HANDSHAKE = 'event.amqp.connection.handshake';
80
81		/**
82		* Event raised when connection close frame incoming
83		*/
84
85		const EVENT_ON_CONNECTION_CLOSE = 'event.amqp.connection.close';
86
87		/**
88		* @var FrameParser
89		*/
90		protected $parser;
91
92		/**
93		* @var FrameSerializer
94		*/
95		protected $serializer;
96
97		/**
98		* @var ConnectionOptions
99		*/
100		protected $connectionOptions;
101
102		/**
103		* @var bool
104		*/
105		protected $isHandshaked = false;
106
107		/**
108		* The broker's supported features.
109		* @var Features
110		*/
111		private $features;
112
113		/**
114		* @var int
115		*/
116		private $maximumChannelCount;
117
118		/**
119		* @var int
120		*/
121		private $maximumFrameSize;
122
123		/**
124		* @var array
125		*/
126		private $channels = [];
127
128		/**
129		* @var int
130		*/
131		private $nextChannelId = 1;
132
133		/**
134		* @var bool
135		*/
136		private $debug = false;
137
138		/**
139		*
140		*/
141		protected function init()
142		{
143		$this->parser = new FrameParser(new TableParser());
144		$this->serializer = new FrameSerializer(new TableSerializer());
145
146		$this->connectionOptions = new ConnectionOptions(
147		$this->pool->config->host->value,
148		$this->pool->config->port->value,
149		$this->pool->config->username->value,
150		$this->pool->config->password->value,
151		$this->pool->config->vhost->value
152		);
153
154		$this->debug = isset($this->pool->config->debug);
155		}
156
157		/**
158		*
159		*/
160		public function onReady()
161		{
162		if ($this->isHandshaked) {
163		parent::onReady();
164		}
165
166		$this->write(self::PROTOCOL_HEADER);
167
168		parent::onReady();
169		}
170
171		/**
172		*
173		* @throws \PHPDaemon\Clients\AMQP\Driver\Protocol\Exception\AMQPProtocolException
174		* @throws \PHPDaemon\Clients\AMQP\Driver\Exception\AMQPConnectionException
175		* @throws \InvalidArgumentException
176		*/
177		protected function onRead()
178		{
179		if ($this->getInputLength() <= 0) {
180		return;
181		}
182		// set busy for connection
183		$this->busy = true;
184
185		/**
186		* 1) read Frame::HEADER_SIZE
187		* 2) get payload length from header
188		* 3) concatenate header and payload + 1 byte (for Constants::FRAME_END) into buffer
189		* 4) parse buffer
190		* 5) do stuff...
191		* 6) if bev contains more data go to 1
192		*/
193		frame:
194		$header = $this->readExact(Frame::HEADER_SIZE);
195		$framePayloadSize = Binary::b2i(substr($header, Frame::HEADER_TYPE_SIZE + Frame::HEADER_CHANNEL_SIZE, Frame::HEADER_PAYLOAD_LENGTH_SIZE));
		0 ignored issues – show Coding Style introduced 2018-01-21 18:23 UTC by Report Bug Copy Issue Report This line exceeds maximum limit of 120 characters; contains 146 characters Overly long lines are hard to read on any screen. Most code styles therefor impose a maximum limit on the number of characters in a line. Loading history...
196
197		$buffer = $header . $this->readExact($framePayloadSize + 1);
198
199		$frame = $this->parser->feed($buffer);
200		if ($frame === null) {
201		return;
202		}
203		if ($this->debug) {
204		$this->log(sprintf('[AMQP] %s packet received', get_class($frame)));
205		}
206
207		if (!$this->isHandshaked) {
208		switch (true) {
209		case $frame instanceof ConnectionStartFrame:
210		if ($frame->versionMajor !== 0 \|\| $frame->versionMinor !== 9) {
211		throw AMQPConnectionException::handshakeFailed(
212		$this->connectionOptions,
213		sprintf(
214		'the broker reported an unexpected AMQP version (v%d.%d)',
215		$frame->versionMajor,
216		$frame->versionMinor
217		)
218		);
219		}
220		if (!preg_match('/\bAMQPLAIN\b/', $frame->mechanisms)) {
221		throw AMQPConnectionException::handshakeFailed(
222		$this->connectionOptions,
223		'the AMQPLAIN authentication mechanism is not supported'
224		);
225		}
226
227		$this->features = new Features();
228		$properties = $frame->serverProperties;
229		if (isset($properties['product']) && 'RabbitMQ' === $properties['product']) {
230		$this->features->qosSizeLimits = false;
231		}
232		if (array_key_exists('capabilities', $properties)) {
233		if (array_key_exists('per_consumer_qos', $properties['capabilities'])) {
234		$this->features->perConsumerQos = (bool)$properties['capabilities']['per_consumer_qos'];
235		}
236		if (array_key_exists('exchange_exchange_bindings', $properties['capabilities'])) {
237		$this->features->exchangeToExchangeBindings = (bool)$properties['capabilities']['exchange_exchange_bindings'];
		0 ignored issues – show Coding Style introduced 2018-01-21 18:23 UTC by Report Bug Copy Issue Report This line exceeds maximum limit of 120 characters; contains 138 characters Overly long lines are hard to read on any screen. Most code styles therefor impose a maximum limit on the number of characters in a line. Loading history...
238		}
239		}
240
241		// Serialize credentials in "AMQPLAIN" format, which is essentially an
242		// AMQP table without the 4-byte size header ...
243		$user = $this->connectionOptions->getUsername();
244		$pass = $this->connectionOptions->getPassword();
245
246		$credentials = "\x05LOGINS" . pack('N', strlen($user)) . $user
247		. "\x08PASSWORDS" . pack('N', strlen($pass)) . $pass;
248
249		$this->command(ConnectionStartOkFrame::create(
250		[
251		'product' => $this->connectionOptions->getProductName(),
252		'version' => $this->connectionOptions->getProductVersion(),
253		'platform' => PackageInfo::AMQP_PLATFORM,
254		'copyright' => PackageInfo::AMQP_COPYRIGHT,
255		'information' => PackageInfo::AMQP_INFORMATION,
256
257		],
258		'AMQPLAIN',
259		$credentials
260		));
261		break;
262		case $frame instanceof ConnectionTuneFrame:
263		$this->maximumChannelCount = self::MAXIMUM_CHANNELS;
264		if ($frame->channelMax === self::UNLIMITED_CHANNELS) {
265		$this->maximumChannelCount = self::MAXIMUM_CHANNELS;
266		} elseif ($frame->channelMax < self::MAXIMUM_CHANNELS) {
267		$this->maximumChannelCount = $frame->channelMax;
268		}
269
270		$this->maximumFrameSize = self::MAXIMUM_FRAME_SIZE;
271		if ($frame->frameMax === self::UNLIMITED_FRAME_SIZE) {
272		$this->maximumFrameSize = self::MAXIMUM_FRAME_SIZE;
273		} elseif ($frame->frameMax < self::MAXIMUM_FRAME_SIZE) {
274		$this->maximumFrameSize = $frame->frameMax;
275		}
276
277		$heartbeatInterval = 0;
278		if (!self::HEARTBEAT_DISABLED) {
279		$heartbeatInterval = $this->connectionOptions->getHeartbeatInterval();
280		if (null === $heartbeatInterval) {
281		$heartbeatInterval = $frame->heartbeat;
282		} elseif ($frame->heartbeat < $heartbeatInterval) {
283		$heartbeatInterval = $frame->heartbeat;
284		}
285		}
286
287		$outputFrame = ConnectionTuneOkFrame::create(
288		$this->maximumChannelCount,
289		$this->maximumFrameSize,
290		$heartbeatInterval
291		);
292
293		if ($outputFrame->heartbeat > 0) {
294		/**
295		* We need to set timeout value = ConnectionTuneFrame::heartbeat + 5 sec
296		*/
297		$timeout = $outputFrame->heartbeat + 5;
298		$this->setTimeout($timeout);
299		$this->connectionOptions->setHeartbeatInterval($outputFrame->heartbeat);
300		$this->connectionOptions->setConnectionTimeout($timeout);
301		}
302
303		$this->command($outputFrame);
304
305		$outputFrame = ConnectionOpenFrame::create(
306		$this->connectionOptions->getVhost()
307		);
308		$this->command($outputFrame);
309		break;
310
311		case $frame instanceof ConnectionOpenOkFrame:
312		$this->isHandshaked = true;
313		$this->openChannel(function ($channel) {
314		$this->trigger(self::EVENT_ON_HANDSHAKE, $channel);
315		});
316		break;
317		}
318
319		$this->checkFree();
320		return;
321		}
322
323		switch (true) {
324		case $frame instanceof HeartbeatFrame:
325		$this->command($frame);
326		break;
327		case $frame instanceof ConnectionCloseFrame:
328		$this->trigger(self::EVENT_ON_CONNECTION_CLOSE, $frame->replyCode, $frame->replyText);
329		$this->close();
330		return;
331		break;
		0 ignored issues – show Unused Code introduced 2018-01-21 18:23 UTC by Report Bug Copy Issue Report `break` is not strictly necessary here and could be removed. The `break` statement is not necessary if it is preceded for example by a `return` statement: switch ($x) { case 1: return 'foo'; break; // This break is not necessary and can be left off. } If you would like to keep this construct to be consistent with other `case` statements, you can safely mark this issue as a false-positive. Loading history...
332		default:
333		if (isset($frame->frameChannelId)
334		&& $frame->frameChannelId > 0
335		&& array_key_exists($frame->frameChannelId, $this->channels)) {
336		/** @var Channel $channel */
337		$channel = $this->channels[$frame->frameChannelId];
338		$channel->trigger(get_class($frame), $frame);
339		break; // exit
340		}
341
342		$this->trigger(get_class($frame), $frame);
343		break;
344		}
345
346		if ($this->bev && $this->getInputLength() > 0) {
347		goto frame;
348		}
349
350		$this->checkFree();
351		}
352
353		/**
354		* @param int $id
355		* @param callable $callback
356		* @throws \InvalidArgumentException
357		* @throws \PHPDaemon\Clients\AMQP\Driver\Protocol\Exception\AMQPProtocolException
358		* @throws \PHPDaemon\Clients\AMQP\Driver\Exception\AMQPConnectionException
359		*/
360		public function getChannel(callable $callback, $id = 1)
361		{
362		if (count($this->channels) === 0) {
363		$this->openChannel($callback);
364		return;
365		}
366
367		if (is_callable($callback)) {
368		$callback($this->channels[$id]);
369		}
370		}
371
372		/**
373		* @param callable\|null $callback
374		* @throws \InvalidArgumentException
375		* @throws \PHPDaemon\Clients\AMQP\Driver\Protocol\Exception\AMQPProtocolException
376		* @throws \PHPDaemon\Clients\AMQP\Driver\Exception\AMQPConnectionException
377		*/
378		public function openChannel(callable $callback = null)
379		{
380		new Channel($this, $callback);
381		}
382
383		/**
384		* @return int
385		* @throws \PHPDaemon\Clients\AMQP\Driver\Exception\AMQPConnectionException
386		*/
387		public function findChannelId()
388		{
389		// first check in range [next, max] ...
390	View Code Duplication	for (
		0 ignored issues – show Duplication introduced 2018-01-21 18:23 UTC by Report Bug Copy Issue Report This code seems to be duplicated across your project. Duplicated code is one of the most pungent code smells. If you need to duplicate the same code in three or more different places, we strongly encourage you to look into extracting the code into a single class or operation. You can also find more detailed suggestions in the “Code” section of your repository. Loading history...
391		$channelId = $this->nextChannelId;
392		$channelId <= $this->maximumChannelCount;
393		++$channelId
394		) {
395		if (!isset($this->channels[$channelId])) {
396		$this->nextChannelId = $channelId + 1;
397
398		return $channelId;
399		}
400		}
401
402		// then check in range [min, next) ...
403	View Code Duplication	for (
		0 ignored issues – show Duplication introduced 2018-01-21 18:23 UTC by Report Bug Copy Issue Report This code seems to be duplicated across your project. Duplicated code is one of the most pungent code smells. If you need to duplicate the same code in three or more different places, we strongly encourage you to look into extracting the code into a single class or operation. You can also find more detailed suggestions in the “Code” section of your repository. Loading history...
404		$channelId = 1;
405		$channelId < $this->nextChannelId;
406		++$channelId
407		) {
408		if (!isset($this->channels[$channelId])) {
409		$this->nextChannelId = $channelId + 1;
410
411		return $channelId;
412		}
413		}
414
415		throw new AMQPConnectionException('No available channels');
416		}
417
418		/**
419		* @param OutgoingFrame $frame
420		* @param callable\|null $callback
421		* @return bool
422		* @throws \PHPDaemon\Clients\AMQP\Driver\Protocol\Exception\AMQPProtocolException
423		* @throws \InvalidArgumentException
424		*/
425		public function command(OutgoingFrame $frame, callable $callback = null)
426		{
427		if ($callback) {
428		$this->onResponse($callback);
429		}
430		$serializedFrame = $this->serializer->serialize($frame);
431		return $this->write($serializedFrame);
432		}
433
434
435		public function addChannel($id, Channel $channel)
436		{
437		$this->channels[$id] = $channel;
438		$this->nextChannelId = max(array_keys($this->channels)) + 1;
439		return $this;
440		}
441
442		/**
443		* @return Features
444		*/
445		public function getFeatures()
446		{
447		return $this->features;
448		}
449
450		/**
451		* @return ConnectionOptions
452		*/
453		public function getConnectionOptions()
454		{
455		return $this->connectionOptions;
456		}
457
458		/**
459		* @return int
460		*/
461		public function getMaximumChannelCount()
462		{
463		return $this->maximumChannelCount;
464		}
465
466		/**
467		* @return int
468		*/
469		public function getMaximumFrameSize()
470		{
471		return $this->maximumFrameSize;
472		}
473
474		/**
475		* @return bool
476		*/
477		public function isHandshaked()
478		{
479		return $this->isHandshaked;
480		}
481		}
482

kakserpom / phpdaemon

Pull Request — master (#286)

Connection::onRead() D

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Duplication Side-by-Side

Filter issues like