1
|
|
|
<?php |
2
|
|
|
|
3
|
|
|
namespace Cmobi\RabbitmqBundle\Transport\Worker; |
4
|
|
|
|
5
|
|
|
use Cmobi\RabbitmqBundle\Connection\CmobiAMQPChannel; |
6
|
|
|
use Cmobi\RabbitmqBundle\Connection\CmobiAMQPConnectionInterface; |
7
|
|
|
use Cmobi\RabbitmqBundle\Connection\ConnectionManager; |
8
|
|
|
use Cmobi\RabbitmqBundle\Queue\CmobiAMQPMessage; |
9
|
|
|
use Cmobi\RabbitmqBundle\Queue\QueueProducerInterface; |
10
|
|
|
use Cmobi\RabbitmqBundle\Transport\Exception\QueueNotFoundException; |
11
|
|
|
|
12
|
|
|
class Task implements QueueProducerInterface |
13
|
|
|
{ |
14
|
|
|
private $connectionManager; |
15
|
|
|
private $fromName; |
16
|
|
|
private $queueName; |
17
|
|
|
|
18
|
|
|
public function __construct($queueName, ConnectionManager $manager, $fromName) |
19
|
|
|
{ |
20
|
|
|
$this->queueName = $queueName; |
21
|
|
|
$this->fromName = $fromName; |
22
|
|
|
$this->connectionManager = $manager; |
23
|
|
|
} |
24
|
|
|
|
25
|
|
|
/** |
26
|
|
|
* @param $data |
27
|
|
|
* @param int $expire |
28
|
|
|
* @param int $priority |
29
|
|
|
* @throws QueueNotFoundException |
30
|
|
|
* @throws \Cmobi\RabbitmqBundle\Connection\Exception\NotFoundAMQPConnectionFactoryException |
31
|
|
|
*/ |
32
|
|
|
public function publish($data, $expire = self::DEFAULT_TTL, $priority = self::PRIORITY_LOW) |
33
|
|
|
{ |
34
|
|
|
/** @var CmobiAMQPConnectionInterface $connection */ |
35
|
|
|
$connection = $this->connectionManager->getConnection(); |
36
|
|
|
$channel = $connection->channel(); |
37
|
|
|
|
38
|
|
|
if (! $this->queueHasExists($channel)) { |
39
|
|
|
throw new QueueNotFoundException("Queue $this->queueName not declared."); |
40
|
|
|
} |
41
|
|
|
$queueBag = new WorkerQueueBag($this->getQueueName()); |
42
|
|
|
$channel->queueDeclare($queueBag->getQueueDeclare()); |
43
|
|
|
$msg = new CmobiAMQPMessage( |
44
|
|
|
(string) $data, |
45
|
|
|
[ |
46
|
|
|
'delivery_mode' => 2, // make message persistent |
47
|
|
|
'priority' => $priority, |
48
|
|
|
] |
49
|
|
|
); |
50
|
|
|
$channel->basic_publish($msg, '', $this->getQueueName()); |
51
|
|
|
|
52
|
|
|
$channel->close(); |
53
|
|
|
$connection->close(); |
54
|
|
|
} |
55
|
|
|
|
56
|
|
|
/** |
57
|
|
|
* @param CmobiAMQPChannel $channel |
58
|
|
|
* @return bool |
59
|
|
|
*/ |
60
|
|
View Code Duplication |
public function queueHasExists(CmobiAMQPChannel $channel) |
|
|
|
|
61
|
|
|
{ |
62
|
|
|
try { |
63
|
|
|
$channel->queue_declare($this->queueName, true); |
64
|
|
|
} catch (\Exception $e) { |
65
|
|
|
return false; |
66
|
|
|
} |
67
|
|
|
|
68
|
|
|
return true; |
69
|
|
|
} |
70
|
|
|
|
71
|
|
|
/** |
72
|
|
|
* @return string |
73
|
|
|
*/ |
74
|
|
|
public function getQueueName() |
75
|
|
|
{ |
76
|
|
|
return $this->queueName; |
77
|
|
|
} |
78
|
|
|
|
79
|
|
|
/** |
80
|
|
|
* @return string |
81
|
|
|
*/ |
82
|
|
|
public function getFromName() |
83
|
|
|
{ |
84
|
|
|
return $this->fromName; |
85
|
|
|
} |
86
|
|
|
|
87
|
|
|
/** |
88
|
|
|
* @return string |
89
|
|
|
*/ |
90
|
|
|
public function getExchange() |
91
|
|
|
{ |
92
|
|
|
return false; |
|
|
|
|
93
|
|
|
} |
94
|
|
|
|
95
|
|
|
/** |
96
|
|
|
* @return string |
97
|
|
|
*/ |
98
|
|
|
public function getExchangeType() |
99
|
|
|
{ |
100
|
|
|
return false; |
|
|
|
|
101
|
|
|
} |
102
|
|
|
} |
103
|
|
|
|
Duplicated code is one of the most pungent code smells. If you need to duplicate the same code in three or more different places, we strongly encourage you to look into extracting the code into a single class or operation.
You can also find more detailed suggestions in the “Code” section of your repository.