JFIFxxC      C  " }!1AQa"q2#BR$3br %&'()*456789:CDEFGHIJSTUVWXYZcdefghijstuvwxyz w!1AQaq"2B #3Rbrdefault = $default; $this->blockFor = $blockFor; $this->timeToRun = $timeToRun; $this->pheanstalk = $pheanstalk; $this->dispatchAfterCommit = $dispatchAfterCommit; } /** * Get the size of the queue. * * @param string|null $queue * @return int */ public function size($queue = null) { return (int) $this->pheanstalk->statsTube(new TubeName($this->getQueue($queue)))->currentJobsReady; } /** * Push a new job onto the queue. * * @param string $job * @param mixed $data * @param string|null $queue * @return mixed */ public function push($job, $data = '', $queue = null) { return $this->enqueueUsing( $job, $this->createPayload($job, $this->getQueue($queue), $data), $queue, null, function ($payload, $queue) { return $this->pushRaw($payload, $queue); } ); } /** * Push a raw payload onto the queue. * * @param string $payload * @param string|null $queue * @param array $options * @return mixed */ public function pushRaw($payload, $queue = null, array $options = []) { $this->pheanstalk->useTube(new TubeName($this->getQueue($queue))); return $this->pheanstalk->put( $payload, Pheanstalk::DEFAULT_PRIORITY, Pheanstalk::DEFAULT_DELAY, $this->timeToRun ); } /** * Push a new job onto the queue after (n) seconds. * * @param \DateTimeInterface|\DateInterval|int $delay * @param string $job * @param mixed $data * @param string|null $queue * @return mixed */ public function later($delay, $job, $data = '', $queue = null) { return $this->enqueueUsing( $job, $this->createPayload($job, $this->getQueue($queue), $data), $queue, $delay, function ($payload, $queue, $delay) { $this->pheanstalk->useTube(new TubeName($this->getQueue($queue))); return $this->pheanstalk->put( $payload, Pheanstalk::DEFAULT_PRIORITY, $this->secondsUntil($delay), $this->timeToRun ); } ); } /** * Push an array of jobs onto the queue. * * @param array $jobs * @param mixed $data * @param string|null $queue * @return void */ public function bulk($jobs, $data = '', $queue = null) { foreach ((array) $jobs as $job) { if (isset($job->delay)) { $this->later($job->delay, $job, $data, $queue); } else { $this->push($job, $data, $queue); } } } /** * Pop the next job off of the queue. * * @param string|null $queue * @return \Illuminate\Contracts\Queue\Job|null */ public function pop($queue = null) { $this->pheanstalk->watch( $tube = new TubeName($queue = $this->getQueue($queue)) ); foreach ($this->pheanstalk->listTubesWatched() as $watched) { if ($watched->value !== $tube->value) { $this->pheanstalk->ignore($watched); } } $job = $this->pheanstalk->reserveWithTimeout($this->blockFor); if ($job instanceof JobIdInterface) { return new BeanstalkdJob( $this->container, $this->pheanstalk, $job, $this->connectionName, $queue ); } } /** * Delete a message from the Beanstalk queue. * * @param string $queue * @param string|int $id * @return void */ public function deleteMessage($queue, $id) { $this->pheanstalk->useTube(new TubeName($this->getQueue($queue))); $this->pheanstalk->delete(new Job(new JobId($id), '')); } /** * 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\Contract\PheanstalkManagerInterface&\Pheanstalk\Contract\PheanstalkPublisherInterface&\Pheanstalk\Contract\PheanstalkSubscriberInterface */ public function getPheanstalk() { return $this->pheanstalk; } }