BeanstalkDriver.php

<?php

namespace Bernard\Driver;

class BeanstalkDriver implements \Bernard\Driver
{
    private
        $_beanstalk,
        $_config
        ;

    public function __construct($beanstalk, $config=array())
    {
        $this->_beanstalk = $beanstalk;
        $this->_config = array_merge(array(
            'default_ttr' => \Pheanstalk_PheanstalkInterface::DEFAULT_TTR,
            'default_priority' => \Pheanstalk_PheanstalkInterface::DEFAULT_PRIORITY,
            'default_delay' => \Pheanstalk_PheanstalkInterface::DEFAULT_DELAY
        ), $config);
    }

    /**
     * {@inheritDoc}
     */
    public function countMessages($queueName)
    {
        if(!in_array($queueName, $this->listQueues()))
            return 0;

        $stats = $this->_beanstalk->statsTube($queueName);
        return $stats['current-jobs-ready'];
    }

    /**
     * {@inheritDoc}
     */
    public function createQueue($queueName)
    {
        // NOOP - all methods specify a tube to use
    }

    /**
     * {@inheritDoc}
     */
    public function removeQueue($queueName)
    {
        throw new \BadMethodCallException(__METHOD__.' not implemented');
    }

    /**
     * {@inheritDoc}
     */
    public function listQueues()
    {
        return $this->_beanstalk->listTubes();
    }

    /**
     * {@inheritDoc}
     */
    public function pushMessage($queueName, $message)
    {
        $this->_beanstalk->putInTube($queueName, $message,
            $this->_config['default_priority'], $this->_config['default_delay'],
            $this->_config['default_ttr']);
    }

    /**
     * {@inheritDoc}
     */
    public function popMessage($queueName, $interval = 5)
    {
        if($job = $this->_beanstalk->reserveFromTube($queueName, $interval)) {
            return array($job->getData(), serialize($job));
        }

        return array(null, null);
    }

    /**
     * {@inheritDoc}
     */
    public function peekQueue($queueName, $index = 0, $limit = 20)
    {
        throw new \BadMethodCallException(__METHOD__.' not implemented');
    }

    /**
     * {@inheritDoc}
     */
    public function acknowledgeMessage($queueName, $receipt)
    {
        $this->_beanstalk->delete(unserialize($receipt));
    }

    /**
     * {@inheritDoc}
     */
    public function info()
    {
        return $this->_beanstalk->stats();
    }
}