ButterAMQP\Wire

Complexity

Total Complexity

29

Size/Duplication

Total Lines	245
Duplicated Lines	0 %

Coupling/Cohesion

Components	1
Dependencies	13

Test Coverage

Coverage

98.84%

Importance

Changes	5
Bugs	0	Features	2

12 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	6	1
A	open()	0	17	1
A	subscribe()	0	6	1
A	close()	0	6	1
A	setHeartbeat()	0	6	1
A	setFrameMax()	0	6	1
A	getSubscriber()	0	4	2
A	getProtocolForScheme()	0	4	2
B	next()	0	38	6
A	send()	0	23	2
A	chop()	0	17	4
C	wait()	0	27	7

<?php

namespace ButterAMQP;

use ButterAMQP\Exception\InvalidFrameEndingException;
use ButterAMQP\Framing\Content;
use ButterAMQP\Framing\Frame;
use ButterAMQP\Framing\Heartbeat;
use ButterAMQP\Heartbeat\NullHeartbeat;
use ButterAMQP\Value\LongValue;
use ButterAMQP\Value\ShortValue;
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;

    /**
     * @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->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(0, new Heartbeat());
        }

        if (($buffer = $this->io->peek(7, $blocking)) === null) {
            return null;
        }

        $header = unpack('Ctype/nchannel/Nsize', $buffer);

        if (($buffer = $this->io->read($header['size'] + 8, $blocking)) === null) {
            return null;
        }

        $payload = substr($buffer, 7, strlen($buffer) - 8);
        $end = $buffer[strlen($buffer) - 1];

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

        $frame = Frame::create($header['type'], $header['channel'], $payload);

        //$this->logger->debug(sprintf('Receive "%s" at channel #%d', get_class($frame), $frame->getChannel()), [

70% of this comment could be valid code. Did you maybe forget this after debugging?

        //    'channel' => $frame->getChannel(),

64% of this comment could be valid code. Did you maybe forget this after debugging?

        //    'frame' => get_class($frame),

60% of this comment could be valid code. Did you maybe forget this after debugging?

        //]);

        if ($subscriber = $this->getSubscriber($frame->getChannel())) {
            $subscriber->dispatch($frame);
        }

        $this->heartbeat->serverBeat();

        return $frame;
    }

    /**
     * {@inheritdoc}
     */
    public function send($channel, Frame $frame)
    {
        //$this->logger->debug(sprintf('Sending "%s" to channel #%d', get_class($frame), $channel), [

69% of this comment could be valid code. Did you maybe forget this after debugging?

        //    'channel' => $channel,

58% of this comment could be valid code. Did you maybe forget this after debugging?

        //    'frame' => get_class($frame),

60% of this comment could be valid code. Did you maybe forget this after debugging?

        //]);

        $this->heartbeat->clientBeat();

        foreach ($this->chop($frame) as $piece) {
            $data = $piece->encode();

            $this->io->write(
                $piece->getFrameType().
                ShortValue::encode($channel).
                LongValue::encode(strlen($data)).
                $data.
                self::FRAME_ENDING
            );
        }

        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(substr($data, $c * $size, $size));
        }

        return $frames;
    }

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

        //$this->logger->debug(sprintf('Waiting "%s" at channel #%d', implode('", "', $types), $channel), [

68% of this comment could be valid code. Did you maybe forget this after debugging?

        //    'channel' => $channel,

58% of this comment could be valid code. Did you maybe forget this after debugging?

        //    'frame' => $types,

58% of this comment could be valid code. Did you maybe forget this after debugging?

        //]);

        do {
            $frame = $this->next(true);

            if (!$frame || $frame->getChannel() != $channel) {
                continue;
            }

            foreach ($types as $type) {
                if (is_a($frame, $type)) {
                    return $frame;
                }
            }
        } while (true);

        return $frame;
    }

    /**
     * {@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';
    }
}