BufferedStompClient.php
public function readMessageFrame(string $subscriptionId, int $timeout): ?Frame
    {
        // pop up frame from the buffer
        if (isset($this->buffer[$subscriptionId]) && ($frame = array_shift($this->buffer[$subscriptionId]))) {
            --$this->currentBufferSize;

            return $frame;
        }

        // do nothing when buffer is full
        if ($this->currentBufferSize >= $this->bufferSize) {
            return null;
        }

        $startTime = microtime(true);
        $remainingTimeout = $timeout * 1000;

        while (true) {
            $this->getConnection()->setReadTimeout(0, $remainingTimeout);

            // there is nothing to read
            if (false === $frame = $this->readFrame()) {
                return null;
            }

            if ('MESSAGE' !== $frame->getCommand()) {
                throw new \LogicException(sprintf('Unexpected frame was received: "%s"', $frame->getCommand()));
            }

            $headers = $frame->getHeaders();

            if (false == isset($headers['subscription'])) {
                throw new \LogicException('Got message frame with missing subscription header');
            }

            // frame belongs to another subscription
            if ($headers['subscription'] !== $subscriptionId) {
                $this->buffer[$headers['subscription']][] = $frame;
                ++$this->currentBufferSize;

                $remainingTimeout -= (microtime(true) - $startTime) * 1000000;

                if ($remainingTimeout <= 0) {
                    return null;
                }

                continue;
            }

            return $frame;
        }
    }
public function readMessageFrame(string $subscriptionId, int $timeout): ?Frame
    {
        // pop up frame from the buffer
        if (isset($this->buffer[$subscriptionId]) && ($frame = array_shift($this->buffer[$subscriptionId]))) {
            --$this->currentBufferSize;

            return $frame;
        }

        // do nothing when buffer is full
        if ($this->currentBufferSize >= $this->bufferSize) {
            return null;
        }

        $startTime = microtime(true);
        $remainingTimeout = $timeout * 1000;

        while (true) {
            $this->getConnection()->setReadTimeout(0, $remainingTimeout);

            // there is nothing to read
            if (false === $frame = $this->readFrame()) {
                return null;
            }

            if ('MESSAGE' !== $frame->getCommand()) {
                throw new \LogicException(sprintf('Unexpected frame was received: "%s"', $frame->getCommand()));
            }

            $headers = $frame->getHeaders();

            if (false == isset($headers['subscription'])) {
                throw new \LogicException('Got message frame with missing subscription header');
            }

            // frame belongs to another subscription
            if ($headers['subscription'] !== $subscriptionId) {
                $this->buffer[$headers['subscription']][] = $frame;
                ++$this->currentBufferSize;

                $remainingTimeout -= (microtime(true) - $startTime) * 1000000;

                if ($remainingTimeout <= 0) {
                    return null;
                }

                continue;
            }

            return $frame;
        }
    }
public function readMessageFrame ( string $ subscriptionId , int $ timeout ) : ? Frame { // pop up frame from the buffer if ( isset ( $ this -> buffer [ $ subscriptionId ] ) && ( $ frame = array_shift ( $ this -> buffer [ $ subscriptionId ] ) ) ) { -- $ this -> currentBufferSize ; return $ frame ; } // do nothing when buffer is full if ( $ this -> currentBufferSize >= $ this -> bufferSize ) { return null ; } $ startTime = microtime ( true ) ; $ remainingTimeout = $ timeout * 1000 ; while ( true ) { $ this -> getConnection ( ) -> setReadTimeout ( 0 , $ remainingTimeout ) ; // there is nothing to read if ( false === $ frame = $ this -> readFrame ( ) ) { return null ; } if ( 'MESSAGE' !== $ frame -> getCommand ( ) ) { throw new \ LogicException ( sprintf ( 'Unexpected frame was received: "%s"' , $ frame -> getCommand ( ) ) ) ; } $ headers = $ frame -> getHeaders ( ) ; if ( false == isset ( $ headers [ 'subscription' ] ) ) { throw new \ LogicException ( 'Got message frame with missing subscription header' ) ; } // frame belongs to another subscription if ( $ headers [ 'subscription' ] !== $ subscriptionId ) { $ this -> buffer [ $ headers [ 'subscription' ] ] [ ] = $ frame ; ++ $ this -> currentBufferSize ; $ remainingTimeout -= ( microtime ( true ) - $ startTime ) * 1000000 ; if ( $ remainingTimeout <= 0 ) { return null ; } continue ; } return $ frame ; } }
Timeout is in milliseconds.
Timeout is in milliseconds .