|
1
|
|
|
<?php |
|
2
|
|
|
|
|
3
|
|
|
/* |
|
4
|
|
|
* This file is part of the Sonata Project package. |
|
5
|
|
|
* |
|
6
|
|
|
* (c) Thomas Rabaix <[email protected]> |
|
7
|
|
|
* |
|
8
|
|
|
* For the full copyright and license information, please view the LICENSE |
|
9
|
|
|
* file that was distributed with this source code. |
|
10
|
|
|
*/ |
|
11
|
|
|
|
|
12
|
|
|
namespace Sonata\NotificationBundle\Iterator; |
|
13
|
|
|
|
|
14
|
|
|
use Interop\Amqp\AmqpConsumer; |
|
15
|
|
|
use PhpAmqpLib\Channel\AMQPChannel; |
|
16
|
|
|
use PhpAmqpLib\Message\AMQPMessage; |
|
17
|
|
|
use Sonata\NotificationBundle\Model\Message; |
|
18
|
|
|
|
|
19
|
|
|
class AMQPMessageIterator implements MessageIteratorInterface |
|
20
|
|
|
{ |
|
21
|
|
|
/** |
|
22
|
|
|
* @deprecated since 3.2, will be removed in 4.x |
|
23
|
|
|
* |
|
24
|
|
|
* @var AMQPChannel |
|
25
|
|
|
*/ |
|
26
|
|
|
protected $channel; |
|
27
|
|
|
|
|
28
|
|
|
/** |
|
29
|
|
|
* @var mixed |
|
30
|
|
|
*/ |
|
31
|
|
|
protected $message; |
|
32
|
|
|
|
|
33
|
|
|
/** |
|
34
|
|
|
* @deprecated since 3.2, will be removed in 4.x |
|
35
|
|
|
* |
|
36
|
|
|
* @var AMQPMessage |
|
37
|
|
|
*/ |
|
38
|
|
|
protected $AMQMessage; |
|
39
|
|
|
|
|
40
|
|
|
/** |
|
41
|
|
|
* @deprecated since 3.2, will be removed in 4.x |
|
42
|
|
|
* |
|
43
|
|
|
* @var string |
|
44
|
|
|
*/ |
|
45
|
|
|
protected $queue; |
|
46
|
|
|
|
|
47
|
|
|
/** |
|
48
|
|
|
* @var int |
|
49
|
|
|
*/ |
|
50
|
|
|
protected $counter; |
|
51
|
|
|
|
|
52
|
|
|
/** |
|
53
|
|
|
* @var \Interop\Amqp\AmqpMessage |
|
54
|
|
|
*/ |
|
55
|
|
|
protected $interopMessage; |
|
56
|
|
|
|
|
57
|
|
|
/** |
|
58
|
|
|
* @var int |
|
59
|
|
|
*/ |
|
60
|
|
|
protected $timeout; |
|
61
|
|
|
|
|
62
|
|
|
/** |
|
63
|
|
|
* @var AmqpConsumer |
|
64
|
|
|
*/ |
|
65
|
|
|
protected $consumer; |
|
66
|
|
|
|
|
67
|
|
|
/** |
|
68
|
|
|
* @var bool |
|
69
|
|
|
*/ |
|
70
|
|
|
protected $isValid; |
|
71
|
|
|
|
|
72
|
|
|
/** |
|
73
|
|
|
* @param AMQPChannel $channel |
|
74
|
|
|
* @param AmqpConsumer $consumer |
|
75
|
|
|
*/ |
|
76
|
|
|
public function __construct(AMQPChannel $channel, AmqpConsumer $consumer) |
|
77
|
|
|
{ |
|
78
|
|
|
$this->consumer = $consumer; |
|
79
|
|
|
$this->counter = 0; |
|
80
|
|
|
$this->timeout = 0; |
|
81
|
|
|
$this->isValid = true; |
|
82
|
|
|
|
|
83
|
|
|
$this->channel = $channel; |
|
|
|
|
|
|
84
|
|
|
$this->queue = $consumer->getQueue()->getQueueName(); |
|
|
|
|
|
|
85
|
|
|
} |
|
86
|
|
|
|
|
87
|
|
|
/** |
|
88
|
|
|
* {@inheritdoc} |
|
89
|
|
|
*/ |
|
90
|
|
|
public function current() |
|
91
|
|
|
{ |
|
92
|
|
|
return $this->message; |
|
93
|
|
|
} |
|
94
|
|
|
|
|
95
|
|
|
/** |
|
96
|
|
|
* {@inheritdoc} |
|
97
|
|
|
*/ |
|
98
|
|
|
public function next() |
|
99
|
|
|
{ |
|
100
|
|
|
if ($amqpMessage = $this->consumer->receive($this->timeout)) { |
|
101
|
|
|
$this->AMQMessage = $this->convertToAmqpLibMessage($amqpMessage); |
|
|
|
|
|
|
102
|
|
|
|
|
103
|
|
|
$data = json_decode($amqpMessage->getBody(), true); |
|
104
|
|
|
$data['body']['interopMessage'] = $amqpMessage; |
|
105
|
|
|
|
|
106
|
|
|
// @deprecated |
|
107
|
|
|
$data['body']['AMQMessage'] = $this->AMQMessage; |
|
|
|
|
|
|
108
|
|
|
|
|
109
|
|
|
$message = new Message(); |
|
110
|
|
|
$message->setBody($data['body']); |
|
111
|
|
|
$message->setType($data['type']); |
|
112
|
|
|
$message->setState($data['state']); |
|
113
|
|
|
$this->message = $message; |
|
114
|
|
|
|
|
115
|
|
|
++$this->counter; |
|
116
|
|
|
$this->isValid = true; |
|
117
|
|
|
} else { |
|
118
|
|
|
$this->isValid = false; |
|
119
|
|
|
} |
|
120
|
|
|
} |
|
121
|
|
|
|
|
122
|
|
|
/** |
|
123
|
|
|
* {@inheritdoc} |
|
124
|
|
|
*/ |
|
125
|
|
|
public function key() |
|
126
|
|
|
{ |
|
127
|
|
|
$this->counter; |
|
128
|
|
|
} |
|
129
|
|
|
|
|
130
|
|
|
/** |
|
131
|
|
|
* {@inheritdoc} |
|
132
|
|
|
*/ |
|
133
|
|
|
public function valid() |
|
134
|
|
|
{ |
|
135
|
|
|
return $this->isValid; |
|
136
|
|
|
} |
|
137
|
|
|
|
|
138
|
|
|
/** |
|
139
|
|
|
* {@inheritdoc} |
|
140
|
|
|
*/ |
|
141
|
|
|
public function rewind() |
|
142
|
|
|
{ |
|
143
|
|
|
$this->next(); |
|
144
|
|
|
} |
|
145
|
|
|
|
|
146
|
|
|
/** |
|
147
|
|
|
* @deprecated since 3.2, will be removed in 4.x |
|
148
|
|
|
* |
|
149
|
|
|
* @param \Interop\Amqp\AmqpMessage $amqpMessage |
|
150
|
|
|
* |
|
151
|
|
|
* @return AMQPMessage |
|
152
|
|
|
*/ |
|
153
|
|
|
private function convertToAmqpLibMessage(\Interop\Amqp\AmqpMessage $amqpMessage) |
|
154
|
|
|
{ |
|
155
|
|
|
$amqpLibProperties = $amqpMessage->getHeaders(); |
|
156
|
|
|
$amqpLibProperties['application_headers'] = $amqpMessage->getProperties(); |
|
157
|
|
|
|
|
158
|
|
|
$amqpLibMessage = new AMQPMessage($amqpMessage->getBody(), $amqpLibProperties); |
|
159
|
|
|
$amqpLibMessage->delivery_info = [ |
|
160
|
|
|
'consumer_tag' => $this->consumer->getConsumerTag(), |
|
161
|
|
|
'delivery_tag' => $amqpMessage->getDeliveryTag(), |
|
162
|
|
|
'redelivered' => $amqpMessage->isRedelivered(), |
|
163
|
|
|
'routing_key' => $amqpMessage->getRoutingKey(), |
|
164
|
|
|
'channel' => $this->channel, |
|
|
|
|
|
|
165
|
|
|
]; |
|
166
|
|
|
|
|
167
|
|
|
return $amqpLibMessage; |
|
168
|
|
|
} |
|
169
|
|
|
} |
|
170
|
|
|
|
sonata-project /
SonataNotificationBundle
Loading history...
This property has been deprecated. The supplier of the class has supplied an explanatory message.
The explanatory message should give you some clue as to whether and when the property will be removed from the class and what other property to use instead.