protected function _callback(callable $callable, AMQPEnvelope $envelope, AMQPQueue $queue)
    {
        if ($this->stop) {
            return false;
        }

        if ($envelope === false) {
            return false;
        }

        if ($this->stop) {
            $this->log('-> Putting job back into the queue', 'warning');
            $queue->nack($envelope->getDeliveryTag(), AMQP_REQUEUE);
            return false;
        }

        $result = false;

        try {
            $this->_working = true;

            $body = $envelope->getBody();
            $compressed = $envelope->getContentEncoding() === 'gzip';

            if ($compressed) {
                $body = gzuncompress($body);
            }

            switch ($envelope->getContentType()) {
                case 'application/json':
                    $result = $callable(json_decode($body, true), $envelope, $queue);
                    break;

                case 'application/text':
                    $result = $callable($body, $envelope, $queue);
                    break;

                default:
                    throw new RuntimeException('Unknown serializer: ' . $envelope->getContentType());
            }

        } catch (Exception $e) {
            $queue->nack($envelope->getDeliveryTag(), AMQP_REQUEUE);
            throw $e;
        } finally {
            $this->_working = false;
        }

        if ($result !== false) {
            return $queue->ack($envelope->getDeliveryTag());
        }

        return $queue->nack($envelope->getDeliveryTag(), AMQP_REQUEUE);
    }
protected function _callback(callable $callable, AMQPEnvelope $envelope, AMQPQueue $queue)
    {
        if ($this->stop) {
            return false;
        }

        if ($envelope === false) {
            return false;
        }

        if ($this->stop) {
            $this->log('-> Putting job back into the queue', 'warning');
            $queue->nack($envelope->getDeliveryTag(), AMQP_REQUEUE);
            return false;
        }

        $result = false;

        try {
            $this->_working = true;

            $body = $envelope->getBody();
            $compressed = $envelope->getContentEncoding() === 'gzip';

            if ($compressed) {
                $body = gzuncompress($body);
            }

            switch ($envelope->getContentType()) {
                case 'application/json':
                    $result = $callable(json_decode($body, true), $envelope, $queue);
                    break;

                case 'application/text':
                    $result = $callable($body, $envelope, $queue);
                    break;

                default:
                    throw new RuntimeException('Unknown serializer: ' . $envelope->getContentType());
            }

        } catch (Exception $e) {
            $queue->nack($envelope->getDeliveryTag(), AMQP_REQUEUE);
            throw $e;
        } finally {
            $this->_working = false;
        }

        if ($result !== false) {
            return $queue->ack($envelope->getDeliveryTag());
        }

        return $queue->nack($envelope->getDeliveryTag(), AMQP_REQUEUE);
    }
protected function _callback ( callable $ callable , AMQPEnvelope $ envelope , AMQPQueue $ queue ) { if ( $ this -> stop ) { return false ; } if ( $ envelope === false ) { return false ; } if ( $ this -> stop ) { $ this -> log ( '-> Putting job back into the queue' , 'warning' ) ; $ queue -> nack ( $ envelope -> getDeliveryTag ( ) , AMQP_REQUEUE ) ; return false ; } $ result = false ; try { $ this -> _working = true ; $ body = $ envelope -> getBody ( ) ; $ compressed = $ envelope -> getContentEncoding ( ) === 'gzip' ; if ( $ compressed ) { $ body = gzuncompress ( $ body ) ; } switch ( $ envelope -> getContentType ( ) ) { case 'application/json' : $ result = $ callable ( json_decode ( $ body , true ) , $ envelope , $ queue ) ; break ; case 'application/text' : $ result = $ callable ( $ body , $ envelope , $ queue ) ; break ; default : throw new RuntimeException ( 'Unknown serializer: ' . $ envelope -> getContentType ( ) ) ; } } catch ( Exception $ e ) { $ queue -> nack ( $ envelope -> getDeliveryTag ( ) , AMQP_REQUEUE ) ; throw $ e ; } finally { $ this -> _working = false ; } if ( $ result !== false ) { return $ queue -> ack ( $ envelope -> getDeliveryTag ( ) ) ; } return $ queue -> nack ( $ envelope -> getDeliveryTag ( ) , AMQP_REQUEUE ) ; }
Actual callback method

@param  callable $callable
@param \AMQPEnvelope $envelope
@param \AMQPQueue $queue
@return boolean
@throws \Exception
@throws \RuntimeException
Actual callback method