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',
                mb_orig_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 = mb_orig_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 = mb_orig_substr($this->buffer, 0, $length);
            $this->buffer = mb_orig_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 = mb_orig_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;
    }
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',
                mb_orig_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 = mb_orig_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 = mb_orig_substr($this->buffer, 0, $length);
            $this->buffer = mb_orig_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 = mb_orig_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;
    }
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' , mb_orig_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 = mb_orig_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 = mb_orig_substr ( $ this -> buffer , 0 , $ length ) ; $ this -> buffer = mb_orig_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 = mb_orig_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 ; }
Retrieve the next frame from the internal buffer.

@param string $buffer Binary data to feed to the parser.
@param int &$requiredBytes The minimum number of bytes that must be
read to produce the next frame.

@return ProtocolFrameInterface|null        The frame parsed from the start of the buffer.
@throws AMQPProtocolException              The incoming data does not conform to the AMQP specification.
Retrieve the next frame from the internal buffer .