PhpAmqpLib/Connection/AbstractConnection.php
protected function wait_frame($timeout = 0)
    {
        if (is_null($this->input))
        {
            $this->setIsConnected(false);
            throw new AMQPConnectionClosedException('Broken pipe or closed connection');
        }

        $currentTimeout = $this->input->getTimeout();
        $this->input->setTimeout($timeout);

        try {
            // frame_type + channel_id + size
            $this->wait_frame_reader->reuse(
                $this->input->read(AMQPReader::OCTET + AMQPReader::SHORT + AMQPReader::LONG)
            );

            $frame_type = $this->wait_frame_reader->read_octet();
            $class = self::$PROTOCOL_CONSTANTS_CLASS;
            if (!array_key_exists($frame_type, $class::$FRAME_TYPES)) {
                throw new AMQPInvalidFrameException('Invalid frame type ' . $frame_type);
            }
            $channel = $this->wait_frame_reader->read_short();
            $size = $this->wait_frame_reader->read_long();

            // payload + ch
            $this->wait_frame_reader->reuse($this->input->read(AMQPReader::OCTET + (int) $size));

            $payload = $this->wait_frame_reader->read($size);
            $ch = $this->wait_frame_reader->read_octet();

        } catch (AMQPTimeoutException $e) {
            $this->input->setTimeout($currentTimeout);
            throw $e;
        } catch (AMQPNoDataException $e) {
            if ($this->input) {
                $this->input->setTimeout($currentTimeout);
            }
            throw $e;
        } catch (AMQPConnectionClosedException $exception) {
            $this->do_close();
            throw $exception;
        }

        $this->input->setTimeout($currentTimeout);

        if ($ch != 0xCE) {
            throw new AMQPInvalidFrameException(sprintf(
                'Framing error, unexpected byte: %x',
                $ch
            ));
        }

        return array($frame_type, $channel, $payload);
    }
protected function wait_frame($timeout = 0)
    {
        if (is_null($this->input))
        {
            $this->setIsConnected(false);
            throw new AMQPConnectionClosedException('Broken pipe or closed connection');
        }

        $currentTimeout = $this->input->getTimeout();
        $this->input->setTimeout($timeout);

        try {
            // frame_type + channel_id + size
            $this->wait_frame_reader->reuse(
                $this->input->read(AMQPReader::OCTET + AMQPReader::SHORT + AMQPReader::LONG)
            );

            $frame_type = $this->wait_frame_reader->read_octet();
            $class = self::$PROTOCOL_CONSTANTS_CLASS;
            if (!array_key_exists($frame_type, $class::$FRAME_TYPES)) {
                throw new AMQPInvalidFrameException('Invalid frame type ' . $frame_type);
            }
            $channel = $this->wait_frame_reader->read_short();
            $size = $this->wait_frame_reader->read_long();

            // payload + ch
            $this->wait_frame_reader->reuse($this->input->read(AMQPReader::OCTET + (int) $size));

            $payload = $this->wait_frame_reader->read($size);
            $ch = $this->wait_frame_reader->read_octet();

        } catch (AMQPTimeoutException $e) {
            $this->input->setTimeout($currentTimeout);
            throw $e;
        } catch (AMQPNoDataException $e) {
            if ($this->input) {
                $this->input->setTimeout($currentTimeout);
            }
            throw $e;
        } catch (AMQPConnectionClosedException $exception) {
            $this->do_close();
            throw $exception;
        }

        $this->input->setTimeout($currentTimeout);

        if ($ch != 0xCE) {
            throw new AMQPInvalidFrameException(sprintf(
                'Framing error, unexpected byte: %x',
                $ch
            ));
        }

        return array($frame_type, $channel, $payload);
    }
protected function wait_frame ( $ timeout = 0 ) { if ( is_null ( $ this -> input ) ) { $ this -> setIsConnected ( false ) ; throw new AMQPConnectionClosedException ( 'Broken pipe or closed connection' ) ; } $ currentTimeout = $ this -> input -> getTimeout ( ) ; $ this -> input -> setTimeout ( $ timeout ) ; try { // frame_type + channel_id + size $ this -> wait_frame_reader -> reuse ( $ this -> input -> read ( AMQPReader :: OCTET + AMQPReader :: SHORT + AMQPReader :: LONG ) ) ; $ frame_type = $ this -> wait_frame_reader -> read_octet ( ) ; $ class = self :: $ PROTOCOL_CONSTANTS_CLASS ; if ( ! array_key_exists ( $ frame_type , $ class :: $ FRAME_TYPES ) ) { throw new AMQPInvalidFrameException ( 'Invalid frame type ' . $ frame_type ) ; } $ channel = $ this -> wait_frame_reader -> read_short ( ) ; $ size = $ this -> wait_frame_reader -> read_long ( ) ; // payload + ch $ this -> wait_frame_reader -> reuse ( $ this -> input -> read ( AMQPReader :: OCTET + ( int ) $ size ) ) ; $ payload = $ this -> wait_frame_reader -> read ( $ size ) ; $ ch = $ this -> wait_frame_reader -> read_octet ( ) ; } catch ( AMQPTimeoutException $ e ) { $ this -> input -> setTimeout ( $ currentTimeout ) ; throw $ e ; } catch ( AMQPNoDataException $ e ) { if ( $ this -> input ) { $ this -> input -> setTimeout ( $ currentTimeout ) ; } throw $ e ; } catch ( AMQPConnectionClosedException $ exception ) { $ this -> do_close ( ) ; throw $ exception ; } $ this -> input -> setTimeout ( $ currentTimeout ) ; if ( $ ch != 0xCE ) { throw new AMQPInvalidFrameException ( sprintf ( 'Framing error, unexpected byte: %x' , $ ch ) ) ; } return array ( $ frame_type , $ channel , $ payload ) ; }
Waits for a frame from the server

@param int|float|null $timeout
@return array
@throws \Exception
@throws \PhpAmqpLib\Exception\AMQPTimeoutException
@throws \PhpAmqpLib\Exception\AMQPRuntimeException
Waits for a frame from the server