Subversion-Projekte lars-tiefland.laravel_shop

Revision

Details | Letzte Änderung | Log anzeigen | RSS feed

Revision Autor Zeilennr. Zeile
148 lars 1 
<?php
2 
 
3 
namespace Illuminate\Queue;
4 
 
5 
use Illuminate\Contracts\Queue\Queue as QueueContract;
6 
use Illuminate\Queue\Jobs\BeanstalkdJob;
7 
use Pheanstalk\Job as PheanstalkJob;
8 
use Pheanstalk\Pheanstalk;
9 
 
10 
class BeanstalkdQueue extends Queue implements QueueContract
11 
{
12 
    /**
13 
     * The Pheanstalk instance.
14 
     *
15 
     * @var \Pheanstalk\Pheanstalk
16 
     */
17 
    protected $pheanstalk;
18 
 
19 
    /**
20 
     * The name of the default tube.
21 
     *
22 
     * @var string
23 
     */
24 
    protected $default;
25 
 
26 
    /**
27 
     * The "time to run" for all pushed jobs.
28 
     *
29 
     * @var int
30 
     */
31 
    protected $timeToRun;
32 
 
33 
    /**
34 
     * The maximum number of seconds to block for a job.
35 
     *
36 
     * @var int
37 
     */
38 
    protected $blockFor;
39 
 
40 
    /**
41 
     * Create a new Beanstalkd queue instance.
42 
     *
43 
     * @param  \Pheanstalk\Pheanstalk  $pheanstalk
44 
     * @param  string  $default
45 
     * @param  int  $timeToRun
46 
     * @param  int  $blockFor
47 
     * @param  bool  $dispatchAfterCommit
48 
     * @return void
49 
     */
50 
    public function __construct(Pheanstalk $pheanstalk,
51 
                                $default,
52 
                                $timeToRun,
53 
                                $blockFor = 0,
54 
                                $dispatchAfterCommit = false)
55 
    {
56 
        $this->default = $default;
57 
        $this->blockFor = $blockFor;
58 
        $this->timeToRun = $timeToRun;
59 
        $this->pheanstalk = $pheanstalk;
60 
        $this->dispatchAfterCommit = $dispatchAfterCommit;
61 
    }
62 
 
63 
    /**
64 
     * Get the size of the queue.
65 
     *
66 
     * @param  string|null  $queue
67 
     * @return int
68 
     */
69 
    public function size($queue = null)
70 
    {
71 
        $queue = $this->getQueue($queue);
72 
 
73 
        return (int) $this->pheanstalk->statsTube($queue)->current_jobs_ready;
74 
    }
75 
 
76 
    /**
77 
     * Push a new job onto the queue.
78 
     *
79 
     * @param  string  $job
80 
     * @param  mixed  $data
81 
     * @param  string|null  $queue
82 
     * @return mixed
83 
     */
84 
    public function push($job, $data = '', $queue = null)
85 
    {
86 
        return $this->enqueueUsing(
87 
            $job,
88 
            $this->createPayload($job, $this->getQueue($queue), $data),
89 
            $queue,
90 
            null,
91 
            function ($payload, $queue) {
92 
                return $this->pushRaw($payload, $queue);
93 
            }
94 
        );
95 
    }
96 
 
97 
    /**
98 
     * Push a raw payload onto the queue.
99 
     *
100 
     * @param  string  $payload
101 
     * @param  string|null  $queue
102 
     * @param  array  $options
103 
     * @return mixed
104 
     */
105 
    public function pushRaw($payload, $queue = null, array $options = [])
106 
    {
107 
        return $this->pheanstalk->useTube($this->getQueue($queue))->put(
108 
            $payload, Pheanstalk::DEFAULT_PRIORITY, Pheanstalk::DEFAULT_DELAY, $this->timeToRun
109 
        );
110 
    }
111 
 
112 
    /**
113 
     * Push a new job onto the queue after (n) seconds.
114 
     *
115 
     * @param  \DateTimeInterface|\DateInterval|int  $delay
116 
     * @param  string  $job
117 
     * @param  mixed  $data
118 
     * @param  string|null  $queue
119 
     * @return mixed
120 
     */
121 
    public function later($delay, $job, $data = '', $queue = null)
122 
    {
123 
        return $this->enqueueUsing(
124 
            $job,
125 
            $this->createPayload($job, $this->getQueue($queue), $data),
126 
            $queue,
127 
            $delay,
128 
            function ($payload, $queue, $delay) {
129 
                return $this->pheanstalk->useTube($this->getQueue($queue))->put(
130 
                    $payload,
131 
                    Pheanstalk::DEFAULT_PRIORITY,
132 
                    $this->secondsUntil($delay),
133 
                    $this->timeToRun
134 
                );
135 
            }
136 
        );
137 
    }
138 
 
139 
    /**
140 
     * Push an array of jobs onto the queue.
141 
     *
142 
     * @param  array  $jobs
143 
     * @param  mixed  $data
144 
     * @param  string|null  $queue
145 
     * @return void
146 
     */
147 
    public function bulk($jobs, $data = '', $queue = null)
148 
    {
149 
        foreach ((array) $jobs as $job) {
150 
            if (isset($job->delay)) {
151 
                $this->later($job->delay, $job, $data, $queue);
152 
            } else {
153 
                $this->push($job, $data, $queue);
154 
            }
155 
        }
156 
    }
157 
 
158 
    /**
159 
     * Pop the next job off of the queue.
160 
     *
161 
     * @param  string|null  $queue
162 
     * @return \Illuminate\Contracts\Queue\Job|null
163 
     */
164 
    public function pop($queue = null)
165 
    {
166 
        $queue = $this->getQueue($queue);
167 
 
168 
        $job = $this->pheanstalk->watchOnly($queue)->reserveWithTimeout($this->blockFor);
169 
 
170 
        if ($job instanceof PheanstalkJob) {
171 
            return new BeanstalkdJob(
172 
                $this->container, $this->pheanstalk, $job, $this->connectionName, $queue
173 
            );
174 
        }
175 
    }
176 
 
177 
    /**
178 
     * Delete a message from the Beanstalk queue.
179 
     *
180 
     * @param  string  $queue
181 
     * @param  string|int  $id
182 
     * @return void
183 
     */
184 
    public function deleteMessage($queue, $id)
185 
    {
186 
        $queue = $this->getQueue($queue);
187 
 
188 
        $this->pheanstalk->useTube($queue)->delete(new PheanstalkJob($id, ''));
189 
    }
190 
 
191 
    /**
192 
     * Get the queue or return the default.
193 
     *
194 
     * @param  string|null  $queue
195 
     * @return string
196 
     */
197 
    public function getQueue($queue)
198 
    {
199 
        return $queue ?: $this->default;
200 
    }
201 
 
202 
    /**
203 
     * Get the underlying Pheanstalk instance.
204 
     *
205 
     * @return \Pheanstalk\Pheanstalk
206 
     */
207 
    public function getPheanstalk()
208 
    {
209 
        return $this->pheanstalk;
210 
    }
211 
}