PHPDaemon\Clients\AMQP\Driver\Protocol\v091\Parser\Frame::feed()

Complexity

Conditions	11
Paths	24

Size

Total Lines	100
Code Lines	57

Duplication

Lines	0
Ratio	0 %

Importance

Changes

0

<?php
    public function feed($buffer, &$requiredBytes = 0)
    {
        $this->buffer .= $buffer;
        $availableBytes = \strlen($this->buffer);

        // not enough bytes for a frame ...
        if ($availableBytes < $this->requiredBytes) {
            $requiredBytes = $this->requiredBytes;

            return null;

            // we're still looking for the header ...
        }
        if ($this->requiredBytes === self::MINIMUM_FRAME_SIZE) {
            // now that we know the payload size we can add that to the number
            // of required bytes ...
            $this->requiredBytes += \unpack(
                'N',
                \substr(
                    $this->buffer,
                    self::HEADER_TYPE_SIZE + self::HEADER_CHANNEL_SIZE,
                    self::HEADER_PAYLOAD_LENGTH_SIZE
                )
            )[1];

            // taking the payload into account we still don't have enough bytes
            // for the frame ...
            if ($availableBytes < $this->requiredBytes) {
                $requiredBytes = $this->requiredBytes;

                return null;
            }
        }

        // we've got enough bytes, check that the last byte is the end marker ...
        if (\ord($this->buffer[$this->requiredBytes - 1]) !== Constants::FRAME_END) {
            throw new AMQPProtocolException(
                sprintf(
                    'Frame end marker (0x%02x) is invalid.',
                    \ord($this->buffer[$this->requiredBytes - 1])
                )
            );
        }

        // read the (t)ype and (c)hannel then discard the header ...
        $fields = \unpack('Ct/nc', $this->buffer);
        $this->buffer = \substr($this->buffer, self::HEADER_SIZE);

        $type = $fields['t'];

        // read the frame ...
        if ($type === Constants::FRAME_METHOD) {
            $frame = $this->parseMethodFrame();
        } elseif ($type === Constants::FRAME_HEADER) {
            $frame = $this->parseHeaderFrame();
        } elseif ($type === Constants::FRAME_BODY) {
            $length = $this->requiredBytes - self::MINIMUM_FRAME_SIZE;
            $frame = new BodyFrame();
            $frame->content = \substr($this->buffer, 0, $length);
            $this->buffer = \substr($this->buffer, $length);
        } elseif ($type === Constants::FRAME_HEARTBEAT) {
            if (self::MINIMUM_FRAME_SIZE !== $this->requiredBytes) {
                throw new AMQPProtocolException(
                    sprintf(
                        'Heartbeat frame payload size (%d) is invalid, must be zero.',
                        $this->requiredBytes - self::MINIMUM_FRAME_SIZE
                    )
                );
            }
            $frame = new HeartbeatFrame();
        } else {
            throw new AMQPProtocolException(
                sprintf(
                    'Frame type (0x%02x) is invalid.',
                    $type
                )
            );
        }

        // discard the end marker ...
        $this->buffer = \substr($this->buffer, 1);

        $consumedBytes = $availableBytes - \strlen($this->buffer);

        // the frame lied about its payload size ...
        if ($consumedBytes !== $this->requiredBytes) {
            throw new AMQPProtocolException(
                sprintf(
                    'Mismatch between frame size (%s) and consumed bytes (%s).',
                    $this->requiredBytes,
                    $consumedBytes
                )
            );
        }

        $this->requiredBytes = $requiredBytes = self::MINIMUM_FRAME_SIZE;
        $frame->frameChannelId = $fields['c'];

        return $frame;
    }
}