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