BeanstalkdQueue.php 3.8KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164
  1. <?php
  2. namespace Illuminate\Queue;
  3. use Pheanstalk\Pheanstalk;
  4. use Pheanstalk\Job as PheanstalkJob;
  5. use Illuminate\Queue\Jobs\BeanstalkdJob;
  6. use Illuminate\Contracts\Queue\Queue as QueueContract;
  7. class BeanstalkdQueue extends Queue implements QueueContract
  8. {
  9. /**
  10. * The Pheanstalk instance.
  11. *
  12. * @var \Pheanstalk\Pheanstalk
  13. */
  14. protected $pheanstalk;
  15. /**
  16. * The name of the default tube.
  17. *
  18. * @var string
  19. */
  20. protected $default;
  21. /**
  22. * The "time to run" for all pushed jobs.
  23. *
  24. * @var int
  25. */
  26. protected $timeToRun;
  27. /**
  28. * Create a new Beanstalkd queue instance.
  29. *
  30. * @param \Pheanstalk\Pheanstalk $pheanstalk
  31. * @param string $default
  32. * @param int $timeToRun
  33. * @return void
  34. */
  35. public function __construct(Pheanstalk $pheanstalk, $default, $timeToRun)
  36. {
  37. $this->default = $default;
  38. $this->timeToRun = $timeToRun;
  39. $this->pheanstalk = $pheanstalk;
  40. }
  41. /**
  42. * Get the size of the queue.
  43. *
  44. * @param string $queue
  45. * @return int
  46. */
  47. public function size($queue = null)
  48. {
  49. $queue = $this->getQueue($queue);
  50. return (int) $this->pheanstalk->statsTube($queue)->current_jobs_ready;
  51. }
  52. /**
  53. * Push a new job onto the queue.
  54. *
  55. * @param string $job
  56. * @param mixed $data
  57. * @param string $queue
  58. * @return mixed
  59. */
  60. public function push($job, $data = '', $queue = null)
  61. {
  62. return $this->pushRaw($this->createPayload($job, $data), $queue);
  63. }
  64. /**
  65. * Push a raw payload onto the queue.
  66. *
  67. * @param string $payload
  68. * @param string $queue
  69. * @param array $options
  70. * @return mixed
  71. */
  72. public function pushRaw($payload, $queue = null, array $options = [])
  73. {
  74. return $this->pheanstalk->useTube($this->getQueue($queue))->put(
  75. $payload, Pheanstalk::DEFAULT_PRIORITY, Pheanstalk::DEFAULT_DELAY, $this->timeToRun
  76. );
  77. }
  78. /**
  79. * Push a new job onto the queue after a delay.
  80. *
  81. * @param \DateTimeInterface|\DateInterval|int $delay
  82. * @param string $job
  83. * @param mixed $data
  84. * @param string $queue
  85. * @return mixed
  86. */
  87. public function later($delay, $job, $data = '', $queue = null)
  88. {
  89. $pheanstalk = $this->pheanstalk->useTube($this->getQueue($queue));
  90. return $pheanstalk->put(
  91. $this->createPayload($job, $data),
  92. Pheanstalk::DEFAULT_PRIORITY,
  93. $this->secondsUntil($delay),
  94. $this->timeToRun
  95. );
  96. }
  97. /**
  98. * Pop the next job off of the queue.
  99. *
  100. * @param string $queue
  101. * @return \Illuminate\Contracts\Queue\Job|null
  102. */
  103. public function pop($queue = null)
  104. {
  105. $queue = $this->getQueue($queue);
  106. $job = $this->pheanstalk->watchOnly($queue)->reserve(0);
  107. if ($job instanceof PheanstalkJob) {
  108. return new BeanstalkdJob(
  109. $this->container, $this->pheanstalk, $job, $this->connectionName, $queue
  110. );
  111. }
  112. }
  113. /**
  114. * Delete a message from the Beanstalk queue.
  115. *
  116. * @param string $queue
  117. * @param string $id
  118. * @return void
  119. */
  120. public function deleteMessage($queue, $id)
  121. {
  122. $queue = $this->getQueue($queue);
  123. $this->pheanstalk->useTube($queue)->delete(new PheanstalkJob($id, ''));
  124. }
  125. /**
  126. * Get the queue or return the default.
  127. *
  128. * @param string|null $queue
  129. * @return string
  130. */
  131. public function getQueue($queue)
  132. {
  133. return $queue ?: $this->default;
  134. }
  135. /**
  136. * Get the underlying Pheanstalk instance.
  137. *
  138. * @return \Pheanstalk\Pheanstalk
  139. */
  140. public function getPheanstalk()
  141. {
  142. return $this->pheanstalk;
  143. }
  144. }