File size: 2,648 Bytes
e4f4821 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 |
<?php
namespace SimpleQueue\Adapter;
use DateTime;
use Pheanstalk\Job as BeanstalkJob;
use Pheanstalk\Pheanstalk;
use Pheanstalk\PheanstalkInterface;
use SimpleQueue\Job;
use SimpleQueue\QueueAdapterInterface;
/**
* Class BeanstalkQueueAdapter
*
* @package SimpleQueue\Adapter
*/
class BeanstalkQueueAdapter implements QueueAdapterInterface
{
/**
* @var PheanstalkInterface
*/
protected $beanstalk;
/**
* @var string
*/
protected $queueName = '';
/**
* BeanstalkQueueAdapter constructor.
*
* @param PheanstalkInterface $beanstalk
* @param string $queueName
*/
public function __construct(PheanstalkInterface $beanstalk, $queueName)
{
$this->beanstalk = $beanstalk;
$this->queueName = $queueName;
}
/**
* Send a job
*
* @access public
* @param Job $job
* @return $this
*/
public function push(Job $job)
{
$this->beanstalk->putInTube($this->queueName, $job->serialize());
return $this;
}
/**
* Schedule a job in the future
*
* @access public
* @param Job $job
* @param DateTime $dateTime
* @return $this
*/
public function schedule(Job $job, DateTime $dateTime)
{
$now = new DateTime();
$when = clone($dateTime);
$delay = $when->getTimestamp() - $now->getTimestamp();
$this->beanstalk->putInTube($this->queueName, $job->serialize(), Pheanstalk::DEFAULT_PRIORITY, $delay);
return $this;
}
/**
* Wait and get job from a queue
*
* @access public
* @return Job|null
*/
public function pull()
{
$beanstalkJob = $this->beanstalk->reserveFromTube($this->queueName);
if ($beanstalkJob === false) {
return null;
}
$job = new Job();
$job->setId($beanstalkJob->getId());
$job->unserialize($beanstalkJob->getData());
return $job;
}
/**
* Acknowledge a job
*
* @access public
* @param Job $job
* @return $this
*/
public function completed(Job $job)
{
$beanstalkJob = new BeanstalkJob($job->getId(), $job->serialize());
$this->beanstalk->delete($beanstalkJob);
return $this;
}
/**
* Mark a job as failed
*
* @access public
* @param Job $job
* @return $this
*/
public function failed(Job $job)
{
$beanstalkJob = new BeanstalkJob($job->getId(), $job->serialize());
$this->beanstalk->bury($beanstalkJob);
return $this;
}
}
|