yii2-queue/BeanstalkdQueue.php at master · talview/yii2-queue · GitHub

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
<?php
/**
 * @link http://www.yiiframework.com/
 * @copyright Copyright (c) 2008 Yii Software LLC
 * @license http://www.yiiframework.com/license/
 */

namespace yii\queue;
use Pheanstalk\Job;
use Pheanstalk\PheanstalkInterface;
use Pheanstalk\Pheanstalk;
use Yii;
use yii\base\Component;
use yii\base\InvalidConfigException;
use yii\helpers\Json;

/**
 * Beanstalkd Queue
 *
 * @author Mani Ka <subramanian.kailash@gmail.com>
 */
class Beanstalkd extends Component implements QueueInterface
{
   /**
     * @var \Pheanstalk\Pheanstalk $beanstalkd
     */
    public $beanstalkd;

    /**
     * Loads the Pheanstalk client for beanstalkd message broker / queue
     * from configuration
     *
     * @throws \yii\base\InvalidConfigException
     */
    public function init()
    {
        parent::init();
        if($this->beanstalkd == null ){
            throw new InvalidConfigException('The "beanstalkd" property must be set.');
        }
        if(!$this->beanstalkd instanceof Pheanstalk){
            $this->beanstalkd = new Pheanstalk($this->beanstalkd);
        }
    }

    /**
     * @param array $message
     * Deletes a job from
     * @return $this
     * @throws \yii\base\InvalidConfigException
     */
    public function delete(Array $message)
    {
        $this->validateMessage($message);
        $this->beanstalkd->useTube($message['queue']);
        $job = new Job($message['id'], $message['body']);
        return $this->beanstalkd->delete($job);
    }

    public function push($payload, $queue,$delay = PheanstalkInterface::DEFAULT_DELAY)
    {
        return $this->beanstalkd->putInTube(
            $queue,
            is_string($payload) ? $payload : Json::encode($payload),
            PheanstalkInterface::DEFAULT_PRIORITY,
            $delay,
            PheanstalkInterface::DEFAULT_TTR);
    }

    public function pop($queue)
    {
        $job = $this->beanstalkd->reserveFromTube($queue);
        return [
            'id'    => $job['id'],
            'queue' => $queue,
            'body'  => $job['data']
        ];
    }

    public function release(Array $message,$delay=PheanstalkInterface::DEFAULT_DELAY)
    {
        $this->validateMessage($message);
        $this->beanstalkd->useTube($message['queue']);
        $job = new Job($message['id'],$message['body']);
        return $this->beanstalkd->release($job,PheanstalkInterface::DEFAULT_PRIORITY,$delay);
    }

    public function purge($queue)
    {
        while ($job = $this->beanstalkd->watch($queue)->ignore("default")->reserve(0)) {
            $this->beanstalkd->delete($job);
        }
    }

    private function validateMessage(Array $message)
    {
        if(isset($message['id']) || ~isset($message['body']) || !isset($message['queue'])){
            throw  new InvalidConfigException("Invalid message configuration");
        }
    }
}