Skip to content

Commit 3cc20ea

Browse files
weaverryanweaverryan
committed
fixing a bug where the delay queue name did not contain the final routing key
1 parent e8a37f5 commit 3cc20ea

File tree

2 files changed

+80
-26
lines changed

2 files changed

+80
-26
lines changed

‎Tests/Transport/AmqpExt/AmqpExtIntegrationTest.php‎

Lines changed: 54 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,9 @@
1919
use Symfony\Component\Messenger\Transport\AmqpExt\AmqpReceivedStamp;
2020
use Symfony\Component\Messenger\Transport\AmqpExt\AmqpReceiver;
2121
use Symfony\Component\Messenger\Transport\AmqpExt\AmqpSender;
22+
use Symfony\Component\Messenger\Transport\AmqpExt\AmqpStamp;
2223
use Symfony\Component\Messenger\Transport\AmqpExt\Connection;
24+
use Symfony\Component\Messenger\Transport\Receiver\ReceiverInterface;
2325
use Symfony\Component\Messenger\Transport\Serialization\Serializer;
2426
use Symfony\Component\Messenger\Transport\Serialization\SerializerInterface;
2527
use Symfony\Component\Process\PhpProcess;
@@ -84,8 +86,10 @@ public function testRetryAndDelay()
8486
$sender = new AmqpSender($connection, $serializer);
8587
$receiver = new AmqpReceiver($connection, $serializer);
8688

89+
// send a first message
8790
$sender->send($first = new Envelope(new DummyMessage('First')));
8891

92+
// receive it immediately and imitate a redeliver with 2 second delay
8993
$envelopes = iterator_to_array($receiver->get());
9094
/** @var Envelope $envelope */
9195
$envelope = $envelopes[0];
@@ -95,25 +99,36 @@ public function testRetryAndDelay()
9599
$sender->send($newEnvelope);
96100
$receiver->ack($envelope);
97101

98-
$envelopes = [];
99-
$startTime = time();
100-
// wait for next message, but only for max 3 seconds
101-
while (0 === \count($envelopes) && $startTime + 3 > time()) {
102-
$envelopes = iterator_to_array($receiver->get());
103-
}
102+
// send a 2nd message with a shorter delay and custom routing key
103+
$customRoutingKeyMessage = new DummyMessage('custom routing key');
104+
$envelopeCustomRoutingKey = new Envelope($customRoutingKeyMessage, [
105+
new DelayStamp(1000),
106+
new AmqpStamp('my_custom_routing_key'),
107+
]);
108+
$sender->send($envelopeCustomRoutingKey);
109+
110+
// wait for next message (but max at 3 seconds)
111+
$startTime = microtime(true);
112+
$envelopes = $this->receiveEnvelopes($receiver, 3);
113+
114+
// duration should be about 1 second
115+
$this->assertApproximateDuration($startTime, 1);
104116

117+
// this should be the custom routing key message first
105118
$this->assertCount(1, $envelopes);
106-
/** @var Envelope $envelope */
107-
$envelope = $envelopes[0];
119+
/* @var Envelope $envelope */
120+
$receiver->ack($envelopes[0]);
121+
$this->assertEquals($customRoutingKeyMessage, $envelopes[0]->getMessage());
108122

109-
// should have a 2 second delay
110-
$this->assertGreaterThanOrEqual($startTime + 2, time());
111-
// but only a 2 second delay
112-
$this->assertLessThan($startTime + 4, time());
123+
// wait for final message (but max at 3 seconds)
124+
$envelopes = $this->receiveEnvelopes($receiver, 3);
125+
// duration should be about 2 seconds
126+
$this->assertApproximateDuration($startTime, 2);
113127

114-
/** @var RedeliveryStamp|null $retryStamp */
128+
/* @var RedeliveryStamp|null $retryStamp */
115129
// verify the stamp still exists from the last send
116-
$retryStamp = $envelope->last(RedeliveryStamp::class);
130+
$this->assertCount(1, $envelopes);
131+
$retryStamp = $envelopes[0]->last(RedeliveryStamp::class);
117132
$this->assertNotNull($retryStamp);
118133
$this->assertSame(1, $retryStamp->getRetryCount());
119134

@@ -206,4 +221,29 @@ private function createSerializer(): SerializerInterface
206221
new SerializerComponent\Serializer([new ObjectNormalizer(), new ArrayDenormalizer()], ['json' => new JsonEncoder()])
207222
);
208223
}
224+
225+
private function assertApproximateDuration($startTime, int $expectedDuration)
226+
{
227+
$actualDuration = microtime(true) - $startTime;
228+
229+
if (\method_exists([$this, 'assertEqualsWithDelta'])) {
230+
$this->assertEqualsWithDelta($expectedDuration, $actualDuration, 'Duration was not within expected range', .5);
231+
} else {
232+
$this->assertEquals($expectedDuration, $actualDuration, 'Duration was not within expected range', .5);
233+
}
234+
}
235+
236+
/**
237+
* @return Envelope[]
238+
*/
239+
private function receiveEnvelopes(ReceiverInterface $receiver, int $timeout): array
240+
{
241+
$envelopes = [];
242+
$startTime = microtime(true);
243+
while (0 === \count($envelopes) && $startTime + $timeout > time()) {
244+
$envelopes = iterator_to_array($receiver->get());
245+
}
246+
247+
return $envelopes;
248+
}
209249
}

‎Transport/AmqpExt/Connection.php‎

Lines changed: 26 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -79,8 +79,8 @@ class Connection
7979
* * flags: Exchange flags (Default: AMQP_DURABLE)
8080
* * arguments: Extra arguments
8181
* * delay:
82-
* * routing_key_pattern: The pattern of the routing key (Default: "delay_%delay%")
83-
* * queue_name_pattern: Pattern to use to create the queues (Default: "delay_queue_%delay%")
82+
* * routing_key_pattern: The pattern of the routing key (Default: "delay_%routing_key%_%delay%")
83+
* * queue_name_pattern: Pattern to use to create the queues (Default: "delay_queue_%routing_key%_%delay%")
8484
* * exchange_name: Name of the exchange to be used for the retried messages (Default: "retry")
8585
* * auto_setup: Enable or not the auto-setup of queues and exchanges (Default: true)
8686
* * loop_sleep: Amount of micro-seconds to wait if no message are available (Default: 200000)
@@ -90,9 +90,9 @@ public function __construct(array $connectionOptions, array $exchangeOptions, ar
9090
{
9191
$this->connectionOptions = array_replace_recursive([
9292
'delay' => [
93-
'routing_key_pattern' => 'delay_%delay%',
93+
'routing_key_pattern' => 'delay_%routing_key%_%delay%',
9494
'exchange_name' => 'delay',
95-
'queue_name_pattern' => 'delay_queue_%delay%',
95+
'queue_name_pattern' => 'delay_queue_%routing_key%_%delay%',
9696
],
9797
], $connectionOptions);
9898
$this->exchangeOptions = $exchangeOptions;
@@ -186,7 +186,7 @@ public function publish(string $body, array $headers = [], int $delay = 0, AmqpS
186186
$this->publishOnExchange(
187187
$this->exchange(),
188188
$body,
189-
(null !== $amqpStamp ? $amqpStamp->getRoutingKey() : null) ?? $this->getDefaultPublishRoutingKey(),
189+
$this->getRoutingKeyForMessage($amqpStamp),
190190
[
191191
'headers' => $headers,
192192
],
@@ -209,14 +209,16 @@ public function countMessagesInQueues(): int
209209
*/
210210
private function publishWithDelay(string $body, array $headers, int $delay, AmqpStamp $amqpStamp = null)
211211
{
212+
$routingKey = $this->getRoutingKeyForMessage($amqpStamp);
213+
212214
if ($this->shouldSetup()) {
213-
$this->setupDelay($delay, null !== $amqpStamp ? $amqpStamp->getRoutingKey() : null);
215+
$this->setupDelay($delay, $routingKey);
214216
}
215217

216218
$this->publishOnExchange(
217219
$this->getDelayExchange(),
218220
$body,
219-
$this->getRoutingKeyForDelay($delay),
221+
$this->getRoutingKeyForDelay($delay, $routingKey),
220222
[
221223
'headers' => $headers,
222224
],
@@ -245,7 +247,7 @@ private function setupDelay(int $delay, ?string $routingKey)
245247

246248
$queue = $this->createDelayQueue($delay, $routingKey);
247249
$queue->declareQueue();
248-
$queue->bind($exchange->getName(), $this->getRoutingKeyForDelay($delay));
250+
$queue->bind($exchange->getName(), $this->getRoutingKeyForDelay($delay, $routingKey));
249251
}
250252

251253
private function getDelayExchange(): \AMQPExchange
@@ -271,13 +273,16 @@ private function getDelayExchange(): \AMQPExchange
271273
private function createDelayQueue(int $delay, ?string $routingKey)
272274
{
273275
$queue = $this->amqpFactory->createQueue($this->channel());
274-
$queue->setName(str_replace('%delay%', $delay, $this->connectionOptions['delay']['queue_name_pattern']));
276+
$queue->setName(str_replace(
277+
['%delay%', '%routing_key%'],
278+
[$delay, $routingKey ?: ''],
279+
$this->connectionOptions['delay']['queue_name_pattern']
280+
));
275281
$queue->setArguments([
276282
'x-message-ttl' => $delay,
277283
'x-dead-letter-exchange' => $this->exchange()->getName(),
278284
]);
279285

280-
$routingKey = $routingKey ?? $this->getDefaultPublishRoutingKey();
281286
if (null !== $routingKey) {
282287
// after being released from to DLX, this routing key will be used
283288
$queue->setArgument('x-dead-letter-routing-key', $routingKey);
@@ -286,9 +291,13 @@ private function createDelayQueue(int $delay, ?string $routingKey)
286291
return $queue;
287292
}
288293

289-
private function getRoutingKeyForDelay(int $delay): string
294+
private function getRoutingKeyForDelay(int $delay, ?string $finalRoutingKey): string
290295
{
291-
return str_replace('%delay%', $delay, $this->connectionOptions['delay']['routing_key_pattern']);
296+
return str_replace(
297+
['%delay%', '%routing_key%'],
298+
[$delay, $finalRoutingKey ?: ''],
299+
$this->connectionOptions['delay']['routing_key_pattern']
300+
);
292301
}
293302

294303
/**
@@ -444,4 +453,9 @@ public function purgeQueues()
444453
$this->queue($queueName)->purge();
445454
}
446455
}
456+
457+
private function getRoutingKeyForMessage(?AmqpStamp $amqpStamp): ?string
458+
{
459+
return (null !== $amqpStamp ? $amqpStamp->getRoutingKey() : null) ?? $this->getDefaultPublishRoutingKey();
460+
}
447461
}

0 commit comments

Comments
 (0)