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