1
|
|
|
<?php |
2
|
|
|
|
3
|
|
|
namespace Cmobi\RabbitmqBundle\Transport\Rpc; |
4
|
|
|
|
5
|
|
|
use Cmobi\RabbitmqBundle\Connection\CmobiAMQPChannel; |
6
|
|
|
use Cmobi\RabbitmqBundle\Connection\ConnectionManager; |
7
|
|
|
use Cmobi\RabbitmqBundle\Queue\CmobiAMQPMessage; |
8
|
|
|
use Cmobi\RabbitmqBundle\Queue\QueueProducerInterface; |
9
|
|
|
use Cmobi\RabbitmqBundle\Transport\Exception\QueueNotFoundException; |
10
|
|
|
use PhpAmqpLib\Message\AMQPMessage; |
11
|
|
|
|
12
|
|
|
class RpcClient implements QueueProducerInterface |
13
|
|
|
{ |
14
|
|
|
private $connectionManager; |
15
|
|
|
private $channel; |
16
|
|
|
private $fromName; |
17
|
|
|
private $queueName; |
18
|
|
|
private $response; |
19
|
|
|
private $correlationId; |
20
|
|
|
private $callbackQueue; |
21
|
|
|
|
22
|
|
|
public function __construct($queueName, ConnectionManager $manager, $fromName = '') |
23
|
|
|
{ |
24
|
|
|
$this->queueName = $queueName; |
25
|
|
|
$this->fromName = $fromName; |
26
|
|
|
$this->connectionManager = $manager; |
27
|
|
|
} |
28
|
|
|
|
29
|
|
|
/** |
30
|
|
|
* @param AMQPMessage $rep |
31
|
|
|
*/ |
32
|
|
|
public function onResponse(AMQPMessage $rep) |
33
|
|
|
{ |
34
|
|
|
if ($rep->get('correlation_id') === $this->correlationId) { |
35
|
|
|
$this->response = $rep->getBody(); |
36
|
|
|
} |
37
|
|
|
} |
38
|
|
|
|
39
|
|
|
/** |
40
|
|
|
* @return \PhpAmqpLib\Channel\AMQPChannel |
41
|
|
|
* |
42
|
|
|
* @throws \Cmobi\RabbitmqBundle\Connection\Exception\NotFoundAMQPConnectionFactoryException |
43
|
|
|
*/ |
44
|
|
View Code Duplication |
public function refreshChannel() |
|
|
|
|
45
|
|
|
{ |
46
|
|
|
$connection = $this->connectionManager->getConnection(); |
47
|
|
|
|
48
|
|
|
if (!$connection->isConnected()) { |
49
|
|
|
$connection->reconnect(); |
50
|
|
|
} |
51
|
|
|
$this->channel = $connection->channel(); |
52
|
|
|
|
53
|
|
|
return $this->channel; |
54
|
|
|
} |
55
|
|
|
|
56
|
|
|
/** |
57
|
|
|
* @param $data |
58
|
|
|
* @param int $expire |
59
|
|
|
* @param int $priority |
60
|
|
|
* @throws QueueNotFoundException |
61
|
|
|
* @throws \Cmobi\RabbitmqBundle\Connection\Exception\NotFoundAMQPConnectionFactoryException |
62
|
|
|
*/ |
63
|
|
|
public function publish($data, $expire = self::DEFAULT_TTL, $priority = self::PRIORITY_LOW) |
64
|
|
|
{ |
65
|
|
|
$this->refreshChannel(); |
66
|
|
|
|
67
|
|
|
if (! $this->queueHasExists()) { |
68
|
|
|
throw new QueueNotFoundException("Queue $this->queueName not declared."); |
69
|
|
|
} |
70
|
|
|
$this->correlationId = $this->generateCorrelationId(); |
71
|
|
|
$queueBag = new RpcQueueBag( |
72
|
|
|
sprintf('callback_to_%s_from_%s_%s', $this->getQueueName(), $this->getFromName(), microtime()) |
73
|
|
|
); |
74
|
|
|
$queueBag->setArguments([ |
75
|
|
|
'x-expires' => ['I', $expire], |
76
|
|
|
]); |
77
|
|
|
list($callbackQueue) = $this->getChannel()->queueDeclare($queueBag->getQueueDeclare()); |
78
|
|
|
$this->callbackQueue = $callbackQueue; |
79
|
|
|
$consumeQueueBag = new RpcQueueBag($callbackQueue); |
80
|
|
|
|
81
|
|
|
$this->getChannel()->basicConsume( |
82
|
|
|
$consumeQueueBag->getQueueConsume(), |
83
|
|
|
[$this, 'onResponse'] |
84
|
|
|
); |
85
|
|
|
$msg = new CmobiAMQPMessage( |
86
|
|
|
(string) $data, |
87
|
|
|
[ |
88
|
|
|
'correlation_id' => $this->correlationId, |
89
|
|
|
'reply_to' => $this->callbackQueue, |
90
|
|
|
'priority' => $priority, |
91
|
|
|
] |
92
|
|
|
); |
93
|
|
|
$this->getChannel()->basic_publish($msg, '', $this->getQueueName()); |
94
|
|
|
|
95
|
|
|
while (! $this->response) { |
96
|
|
|
$this->getChannel()->wait(null, 0, ($expire / 1000)); |
|
|
|
|
97
|
|
|
} |
98
|
|
|
$this->getChannel()->close(); |
99
|
|
|
$this->connectionManager->getConnection()->close(); |
100
|
|
|
} |
101
|
|
|
|
102
|
|
|
/** |
103
|
|
|
* @return bool |
104
|
|
|
*/ |
105
|
|
View Code Duplication |
public function queueHasExists() |
|
|
|
|
106
|
|
|
{ |
107
|
|
|
try { |
108
|
|
|
$this->getChannel()->queue_declare($this->queueName, true); |
109
|
|
|
} catch (\Exception $e) { |
110
|
|
|
return false; |
111
|
|
|
} |
112
|
|
|
|
113
|
|
|
return true; |
114
|
|
|
} |
115
|
|
|
|
116
|
|
|
/** |
117
|
|
|
* @return string |
118
|
|
|
*/ |
119
|
|
|
public function getQueueName() |
120
|
|
|
{ |
121
|
|
|
return $this->queueName; |
122
|
|
|
} |
123
|
|
|
|
124
|
|
|
/** |
125
|
|
|
* @return CmobiAMQPChannel |
126
|
|
|
*/ |
127
|
|
|
public function getChannel() |
128
|
|
|
{ |
129
|
|
|
return $this->channel; |
130
|
|
|
} |
131
|
|
|
|
132
|
|
|
/** |
133
|
|
|
* @return string |
134
|
|
|
*/ |
135
|
|
|
public function getFromName() |
136
|
|
|
{ |
137
|
|
|
return $this->fromName; |
138
|
|
|
} |
139
|
|
|
|
140
|
|
|
/** |
141
|
|
|
* @todo unecessary method set, its only exists to run tests whitout stay jailed in infinite while waiting response. |
142
|
|
|
* |
143
|
|
|
* @param $content |
144
|
|
|
*/ |
145
|
|
|
public function setResponse($content) |
146
|
|
|
{ |
147
|
|
|
$this->response = $content; |
148
|
|
|
} |
149
|
|
|
|
150
|
|
|
/** |
151
|
|
|
* @return string |
152
|
|
|
*/ |
153
|
|
|
public function getResponse() |
154
|
|
|
{ |
155
|
|
|
return $this->response; |
156
|
|
|
} |
157
|
|
|
|
158
|
|
|
/** @return string */ |
159
|
|
|
public function generateCorrelationId() |
160
|
|
|
{ |
161
|
|
|
return uniqid($this->getQueueName()).microtime(); |
162
|
|
|
} |
163
|
|
|
|
164
|
|
|
/** |
165
|
|
|
* @return string |
166
|
|
|
*/ |
167
|
|
|
public function getCurrentCorrelationId() |
168
|
|
|
{ |
169
|
|
|
return $this->correlationId; |
170
|
|
|
} |
171
|
|
|
|
172
|
|
|
/** |
173
|
|
|
* @return string |
174
|
|
|
*/ |
175
|
|
|
public function getExchange() |
176
|
|
|
{ |
177
|
|
|
return false; |
|
|
|
|
178
|
|
|
} |
179
|
|
|
|
180
|
|
|
/** |
181
|
|
|
* @return string |
182
|
|
|
*/ |
183
|
|
|
public function getExchangeType() |
184
|
|
|
{ |
185
|
|
|
return false; |
|
|
|
|
186
|
|
|
} |
187
|
|
|
} |
188
|
|
|
|
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.