Source of file BeanstalkdQueue.php
Size: 3,237 Bytes - Last Modified: 2021-12-23T10:38:27+00:00
/var/www/docs.ssmods.com/process/src/src/BeanstalkdQueue.php
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137 | <?php namespace StudioBonito\SilverStripe\Queue; use Pheanstalk_Job; use Pheanstalk_Pheanstalk as Pheanstalk; use StudioBonito\SilverStripe\Queue\Jobs\BeanstalkdJob; class BeanstalkdQueue extends AbstractQueue implements QueueInterface { /** * The Pheanstalk instance. * * @var Pheanstalk */ protected $pheanstalk; /** * The name of the default tube. * * @var string */ protected $default; /** * The "time to run" for all pushed jobs. * * @var int */ protected $timeToRun; /** * Create a new Beanstalkd queue instance. * * @param Pheanstalk $pheanstalk * @param string $default * @param int $timeToRun * * @return void */ public function __construct(Pheanstalk $pheanstalk, $default, $timeToRun) { $this->default = $default; $this->timeToRun = $timeToRun; $this->pheanstalk = $pheanstalk; } /** * Push a new job onto the queue. * * @param string $job * @param mixed $data * @param string $queue * * @return mixed */ public function push($job, $data = null, $queue = null) { return $this->pushRaw($this->createPayload($job, $data), $this->getQueue($queue)); } /** * Push a raw payload onto the queue. * * @param string $payload * @param string $queue * @param array $options * * @return mixed */ public function pushRaw($payload, $queue = null, array $options = array()) { return $this->pheanstalk->useTube($this->getQueue($queue))->put( $payload, Pheanstalk::DEFAULT_PRIORITY, Pheanstalk::DEFAULT_DELAY, $this->timeToRun ); } /** * Push a new job onto the queue after a delay. * * @param \DateTime|int $delay * @param string $job * @param mixed $data * @param string $queue * * @return mixed */ public function later($delay, $job, $data = '', $queue = null) { $payload = $this->createPayload($job, $data); $pheanstalk = $this->pheanstalk->useTube($this->getQueue($queue)); return $pheanstalk->put($payload, Pheanstalk::DEFAULT_PRIORITY, $this->getSeconds($delay)); } /** * Pop the next job off of the queue. * * @param string $queue * * @return \StudioBonito\SilverStripe\Queue\Jobs\AbstractJob|null */ public function pop($queue = null) { $queue = $this->getQueue($queue); $job = $this->pheanstalk->watchOnly($queue)->reserve(0); if ($job instanceof Pheanstalk_Job) { return new BeanstalkdJob($this->injector, $this->pheanstalk, $job, $queue); } } /** * Get the queue or return the default. * * @param string|null $queue * * @return string */ public function getQueue($queue) { return $queue ? : $this->default; } /** * Get the underlying Pheanstalk instance. * * @return Pheanstalk */ public function getPheanstalk() { return $this->pheanstalk; } } |