Wire::setHeartbeat() - Code Metrics - skolodyazhnyy/butter-amqplib - Measure and Improve Code Quality continuously with Scrutinizer

Wire::setHeartbeat() A
last analyzed 2018-06-14 19:28 UTC

↳ Parent: Wire

Complexity

Conditions	1
Paths	1

Size

Total Lines	6
Code Lines	3

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	3
CRAP Score	1

Importance

Changes

Metric	Value
dl	0
loc	6
ccs	3
cts	3
cp	1
rs	9.4285
c	0
b	0
f	0
cc	1
eloc	3
nc	1
nop	1
crap	1

<?php

namespace ButterAMQP\AMQP091;

use ButterAMQP\Buffer;
use ButterAMQP\Exception\InvalidFrameEndingException;
use ButterAMQP\AMQP091\Framing\Content;
use ButterAMQP\AMQP091\Framing\Frame;
use ButterAMQP\AMQP091\Framing\Heartbeat;
use ButterAMQP\Heartbeat\NullHeartbeat;
use ButterAMQP\HeartbeatInterface;
use ButterAMQP\IOInterface;
use ButterAMQP\Url;
use Psr\Log\LoggerAwareInterface;
use Psr\Log\LoggerAwareTrait;
use Psr\Log\NullLogger;

class Wire implements WireInterface, LoggerAwareInterface
{
    const PROTOCOL_HEADER = "AMQP\x00\x00\x09\x01";
    const FRAME_ENDING = "\xCE";

    use LoggerAwareTrait;

    /**
     * @var IOInterface
     */
    private $io;

    /**
     * @var WireSubscriberInterface[]
     */
    private $subscribers = [];

    /**
     * @var HeartbeatInterface
     */
    private $heartbeat;

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

    /**
     * @var string
     */
    private $buffer;

    /**
     * @param IOInterface $io
     */
    public function __construct(IOInterface $io)
    {
        $this->io = $io;
        $this->logger = new NullLogger();
        $this->heartbeat = new NullHeartbeat();
    }

    /**
     * {@inheritdoc}
     */
    public function open(Url $url)
    {
        $this->subscribers = [];
        $this->buffer = '';

        $this->io->open(
            $this->getProtocolForScheme($url),
            $url->getHost(),
            $url->getPort(),
            $url->getQuery()
        );

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

        // @todo: peek next 8 bytes and check if its a frame or "wrong protocol" reply

        return $this;
    }

    /**
     * {@inheritdoc}
     */
    public function next($blocking = true)
    {
        if ($this->heartbeat->shouldSendHeartbeat()) {
            $this->send(new Heartbeat(0));
        }

        if ($frame = $this->tryNext()) {
            $this->heartbeat->serverBeat();
            $this->dispatch($frame);
        } else {
            $this->read($blocking);
        }

        return $frame;
    }

    /**
     * Read next chunk of the data from the buffer.
     *
     * @param bool $blocking
     */
    private function read($blocking)
    {
        $this->buffer .= $this->io->read($this->frameMax ?: 8192, $blocking);
    }

    /**
     * Tries to fetch next frame from reading buffer.
     * Will return a Frame or null if there is not enough data in the buffer.
     *
     * @return Frame|null
     *
     * @throws InvalidFrameEndingException
     */
    private function tryNext()
    {
        $length = strlen($this->buffer);

        if ($length < 7) {
            return null;
        }

        $header = unpack('Ctype/nchannel/Nsize', $this->buffer);
        $size = $header['size'] + 8;

        if ($length < $size) {
            return null;
        }

        if ($length == $size) {
            $buffer = $this->buffer;
            $this->buffer = '';
        } else {
            $buffer = substr($this->buffer, 0, $size);
            $this->buffer = substr($this->buffer, $size, $length - $size);
        }

        $end = $buffer[$size - 1];

        if ($end != self::FRAME_ENDING) {
            throw new InvalidFrameEndingException(sprintf('Invalid frame ending (%d)', unpack('c', $end)[1]));
        }

        return Frame::decode($header, new Buffer($buffer, 7));
    }

    /**
     * @param Frame $frame
     */
    private function dispatch(Frame $frame)
    {
        if ($subscriber = $this->getSubscriber($frame->getChannel())) {
            $subscriber->dispatch($frame);
        }
    }

    /**
     * {@inheritdoc}
     */
    public function send(Frame $frame)
    {
        $this->heartbeat->clientBeat();

        foreach ($this->chop($frame) as $piece) {
            $this->io->write($piece->encode());
        }

        return $this;
    }

    /**
     * @param Frame $frame
     *
     * @return array
     */
    private function chop(Frame $frame)
    {
        if (!$this->frameMax || !$frame instanceof Content) {
            return [$frame];
        }

        $frames = [];
        $data = $frame->getData();
        $size = $this->frameMax - 8;
        $chunks = ceil(strlen($data) / $size);

        for ($c = 0; $c < $chunks; ++$c) {
            $frames[] = new Content($frame->getChannel(), substr($data, $c * $size, $size));
        }

        return $frames;
    }

    /**
     * {@inheritdoc}
     */
    public function wait($channel, $types)
    {
        if (!is_array($types)) {
            $types = [$types];
        }

        do {
            $frame = $this->next(true);
        } while (!$this->isFrameMatch($frame, $channel, $types));

        return $frame;
    }

    /**
     * @param Frame $frame
     * @param int   $channel
     * @param array $types
     *
     * @return bool
     */
    private function isFrameMatch(Frame $frame = null, $channel = 0, array $types = [])
    {
        return $frame && $frame->getChannel() == $channel && in_array(get_class($frame), $types);
    }

    /**
     * {@inheritdoc}
     */
    public function subscribe($channel, WireSubscriberInterface $handler)
    {
        $this->subscribers[$channel] = $handler;

        return $this;
    }

    /**
     * {@inheritdoc}
     */
    public function close()
    {
        $this->io->close();

        return $this;
    }

    /**
     * @param HeartbeatInterface $heartbeat
     *
     * @return $this
     */
    public function setHeartbeat(HeartbeatInterface $heartbeat)
    {
        $this->heartbeat = $heartbeat;

        return $this;
    }

    /**
     * @param int $frameMax
     *
     * @return $this
     */
    public function setFrameMax($frameMax)
    {
        $this->frameMax = $frameMax;

        return $this;
    }

    /**
     * @param int $channel
     *
     * @return WireSubscriberInterface|null
     */
    private function getSubscriber($channel)
    {
        return isset($this->subscribers[$channel]) ? $this->subscribers[$channel] : null;
    }

    /**
     * @param Url $url
     *
     * @return string
     */
    private function getProtocolForScheme(Url $url)
    {
        return strcasecmp($url->getScheme(), 'amqps') == 0 ? 'ssl' : 'tcp';
    }
}


1		<?php
2
3		namespace ButterAMQP\AMQP091;
4
5		use ButterAMQP\Buffer;
6		use ButterAMQP\Exception\InvalidFrameEndingException;
7		use ButterAMQP\AMQP091\Framing\Content;
8		use ButterAMQP\AMQP091\Framing\Frame;
9		use ButterAMQP\AMQP091\Framing\Heartbeat;
10		use ButterAMQP\Heartbeat\NullHeartbeat;
11		use ButterAMQP\HeartbeatInterface;
12		use ButterAMQP\IOInterface;
13		use ButterAMQP\Url;
14		use Psr\Log\LoggerAwareInterface;
15		use Psr\Log\LoggerAwareTrait;
16		use Psr\Log\NullLogger;
17
18		class Wire implements WireInterface, LoggerAwareInterface
19		{
20		const PROTOCOL_HEADER = "AMQP\x00\x00\x09\x01";
21		const FRAME_ENDING = "\xCE";
22
23		use LoggerAwareTrait;
24
25		/**
26		* @var IOInterface
27		*/
28		private $io;
29
30		/**
31		* @var WireSubscriberInterface[]
32		*/
33		private $subscribers = [];
34
35		/**
36		* @var HeartbeatInterface
37		*/
38		private $heartbeat;
39
40		/**
41		* @var int
42		*/
43		private $frameMax;
44
45		/**
46		* @var string
47		*/
48		private $buffer;
49
50		/**
51		* @param IOInterface $io
52		*/
53	33	public function __construct(IOInterface $io)
54		{
55	33	$this->io = $io;
56	33	$this->logger = new NullLogger();
57	33	$this->heartbeat = new NullHeartbeat();
58	33	}
59
60		/**
61		* {@inheritdoc}
62		*/
63	19	public function open(Url $url)
64		{
65	19	$this->subscribers = [];
66	19	$this->buffer = '';
67
68	19	$this->io->open(
69	19	$this->getProtocolForScheme($url),
70	19	$url->getHost(),
71	19	$url->getPort(),
72	19	$url->getQuery()
73	19	);
74
75	19	$this->io->write(self::PROTOCOL_HEADER);
76
77		// @todo: peek next 8 bytes and check if its a frame or "wrong protocol" reply
78
79	19	return $this;
80		}
81
82		/**
83		* {@inheritdoc}
84		*/
85	28	public function next($blocking = true)
86		{
87	28	if ($this->heartbeat->shouldSendHeartbeat()) {
88	1	$this->send(new Heartbeat(0));
89	1	}
90
91	28	if ($frame = $this->tryNext()) {
92	23	$this->heartbeat->serverBeat();
93	23	$this->dispatch($frame);
94	23	} else {
95	28	$this->read($blocking);
96		}
97
98	28	return $frame;
99		}
100
101		/**
102		* Read next chunk of the data from the buffer.
103		*
104		* @param bool $blocking
105		*/
106	28	private function read($blocking)
107		{
108	28	$this->buffer .= $this->io->read($this->frameMax ?: 8192, $blocking);
109	28	}
110
111		/**
112		* Tries to fetch next frame from reading buffer.
113		* Will return a Frame or null if there is not enough data in the buffer.
114		*
115		* @return Frame\|null
116		*
117		* @throws InvalidFrameEndingException
118		*/
119	28	private function tryNext()
120		{
121	28	$length = strlen($this->buffer);
122
123	28	if ($length < 7) {
124	28	return null;
125		}
126
127	24	$header = unpack('Ctype/nchannel/Nsize', $this->buffer);
128	24	$size = $header['size'] + 8;
129
130	24	if ($length < $size) {
131	1	return null;
132		}
133
134	24	if ($length == $size) {
135	23	$buffer = $this->buffer;
136	23	$this->buffer = '';
137	23	} else {
138	14	$buffer = substr($this->buffer, 0, $size);
139	14	$this->buffer = substr($this->buffer, $size, $length - $size);
140		}
141
142	24	$end = $buffer[$size - 1];
143
144	24	if ($end != self::FRAME_ENDING) {
145	1	throw new InvalidFrameEndingException(sprintf('Invalid frame ending (%d)', unpack('c', $end)[1]));
146		}
147
148	23	return Frame::decode($header, new Buffer($buffer, 7));
149		}
150
151		/**
152		* @param Frame $frame
153		*/
154	23	private function dispatch(Frame $frame)
155		{
156	23	if ($subscriber = $this->getSubscriber($frame->getChannel())) {
157	19	$subscriber->dispatch($frame);
158	19	}
159	23	}
160
161		/**
162		* {@inheritdoc}
163		*/
164	22	public function send(Frame $frame)
165		{
166	22	$this->heartbeat->clientBeat();
167
168	22	foreach ($this->chop($frame) as $piece) {
169	22	$this->io->write($piece->encode());
170	22	}
171
172	22	return $this;
173		}
174
175		/**
176		* @param Frame $frame
177		*
178		* @return array
179		*/
180	22	private function chop(Frame $frame)
181		{
182	22	if (!$this->frameMax \|\| !$frame instanceof Content) {
183	21	return [$frame];
184		}
185
186	15	$frames = [];
187	15	$data = $frame->getData();
188	15	$size = $this->frameMax - 8;
189	15	$chunks = ceil(strlen($data) / $size);
190
191	15	for ($c = 0; $c < $chunks; ++$c) {
192	8	$frames[] = new Content($frame->getChannel(), substr($data, $c * $size, $size));
193	8	}
194
195	15	return $frames;
196		}
197
198		/**
199		* {@inheritdoc}
200		*/
201	20	public function wait($channel, $types)
202		{
203	20	if (!is_array($types)) {
204	19	$types = [$types];
205	19	}
206
207		do {
208	20	$frame = $this->next(true);
209	20	} while (!$this->isFrameMatch($frame, $channel, $types));
210
211	19	return $frame;
212		}
213
214		/**
215		* @param Frame $frame
216		* @param int $channel
217		* @param array $types
218		*
219		* @return bool
220		*/
221	20	private function isFrameMatch(Frame $frame = null, $channel = 0, array $types = [])
222		{
223	20	return $frame && $frame->getChannel() == $channel && in_array(get_class($frame), $types);
224		}
225
226		/**
227		* {@inheritdoc}
228		*/
229	19	public function subscribe($channel, WireSubscriberInterface $handler)
230		{
231	19	$this->subscribers[$channel] = $handler;
232
233	19	return $this;
234		}
235
236		/**
237		* {@inheritdoc}
238		*/
239	19	public function close()
240		{
241	19	$this->io->close();
242
243	19	return $this;
244		}
245
246		/**
247		* @param HeartbeatInterface $heartbeat
248		*
249		* @return $this
250		*/
251	20	public function setHeartbeat(HeartbeatInterface $heartbeat)
252		{
253	20	$this->heartbeat = $heartbeat;
254
255	20	return $this;
256		}
257
258		/**
259		* @param int $frameMax
260		*
261		* @return $this
262		*/
263	18	public function setFrameMax($frameMax)
264		{
265	18	$this->frameMax = $frameMax;
266
267	18	return $this;
268		}
269
270		/**
271		* @param int $channel
272		*
273		* @return WireSubscriberInterface\|null
274		*/
275	23	private function getSubscriber($channel)
276		{
277	23	return isset($this->subscribers[$channel]) ? $this->subscribers[$channel] : null;
278		}
279
280		/**
281		* @param Url $url
282		*
283		* @return string
284		*/
285	19	private function getProtocolForScheme(Url $url)
286		{
287	19	return strcasecmp($url->getScheme(), 'amqps') == 0 ? 'ssl' : 'tcp';
288		}
289		}
290

skolodyazhnyy / butter-amqplib

Wire::setHeartbeat() A last analyzed 2018-06-14 19:28 UTC

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like

Wire::setHeartbeat() A
last analyzed 2018-06-14 19:28 UTC