8000 [Messenger] Add message timestamp to amqp connection · symfony/symfony@f4be339 · GitHub
[go: up one dir, main page]

Skip to content

Commit f4be339

Browse files
author
Bartłomiej Zając
committed
[Messenger] Add message timestamp to amqp connection
1 parent e1cfbd2 commit f4be339

File tree

2 files changed

+10
-9
lines changed

2 files changed

+10
-9
lines changed

src/Symfony/Component/Messenger/Bridge/Amqp/Tests/Transport/ConnectionTest.php

Lines changed: 9 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -266,7 +266,7 @@ public function testItSetupsTheConnectionWithDefaults()
266266
);
267267

268268
$amqpExchange->expects($this->once())->method('declareExchange');
269-
$amqpExchange->expects($this->once())->method('publish')->with('body', null, AMQP_NOPARAM, ['headers' => [], 'delivery_mode' => 2]);
269+
$amqpExchange->expects($this->once())->method('publish')->with('body', null, AMQP_NOPARAM, ['headers' => [], 'delivery_mode' => 2, 'timestamp' => time()]);
270270
$amqpQueue->expects($this->once())->method('declareQueue');
271271
$amqpQueue->expects($this->once())->method('bind')->with(self::DEFAULT_EXCHANGE_NAME, null);
272272

@@ -289,7 +289,7 @@ public function testItSetupsTheConnection()
289289
$factory->method('createQueue')->will($this->onConsecutiveCalls($amqpQueue0, $amqpQueue1));
290290

291291
$amqpExchange->expects($this->once())->method('declareExchange');
292-
$amqpExchange->expects($this->once())->method('publish')->with('body', 'routing_key', AMQP_NOPARAM, ['headers' => [], 'delivery_mode' => 2]);
292+
$amqpExchange->expects($this->once())->method('publish')->with('body', 'routing_key', AMQP_NOPARAM, ['headers' => [], 'delivery_mode' => 2, 'timestamp' => time()]);
293293
$amqpQueue0->expects($this->once())->method('declareQueue');
294294
$amqpQueue0->expects($this->exactly(2))->method('bind')->withConsecutive(
295295
[self::DEFAULT_EXCHANGE_NAME, 'binding_key0'],
@@ -326,7 +326,7 @@ public function testBindingArguments()
326326
$factory->method('createQueue')->willReturn($amqpQueue);
327327

328328
$amqpExchange->expects($this->once())->method('declareExchange');
329-
$amqpExchange->expects($this->once())->method('publish')->with('body', null, AMQP_NOPARAM, ['headers' => [], 'delivery_mode' => 2]);
329+
$amqpExchange->expects($this->once())->method('publish')->with('body', null, AMQP_NOPARAM, ['headers' => [], 'delivery_mode' => 2, 'timestamp' => time()]);
330330
$amqpQueue->expects($this->once())->method('declareQueue');
331331
$amqpQueue->expects($this->exactly(1))->method('bind')->withConsecutive(
332332
[self::DEFAULT_EXCHANGE_NAME, null, ['x-match' => 'all']]
@@ -439,7 +439,7 @@ public function testItDelaysTheMessage()
439439
$delayQueue->expects($this->once())->method('declareQueue');
440440
$delayQueue->expects($this->once())->method('bind')->with('delays', 'delay_messages__5000');
441441

442-
$delayExchange->expects($this->once())->method('publish')->with('{}', 'delay_messages__5000', AMQP_NOPARAM, ['headers' => ['x-some-headers' => 'foo'], 'delivery_mode' => 2]);
442+
$delayExchange->expects($this->once())->method('publish')->with('{}', 'delay_messages__5000', AMQP_NOPARAM, ['headers' => ['x-some-headers' => 'foo'], 'delivery_mode' => 2, 'timestamp' => time()]);
443443

444444
$connection = Connection::fromDsn('amqp://localhost', [], $factory);
445445
$connection->publish('{}', ['x-some-headers' => 'foo'], 5000);
@@ -481,7 +481,7 @@ public function testItDelaysTheMessageWithADifferentRoutingKeyAndTTLs()
481481
$delayQueue->expects($this->once())->method('declareQueue');
482482
$delayQueue->expects($this->once())->method('bind')->with('delays', 'delay_messages__120000');
483483

484-
$delayExchange->expects($this->once())->method('publish')->with('{}', 'delay_messages__120000', AMQP_NOPARAM, ['headers' => [], 'delivery_mode' => 2]);
484+
$delayExchange->expects($this->once())->method('publish')->with('{}', 'delay_messages__120000', AMQP_NOPARAM, ['headers' => [], 'delivery_mode' => 2, 'timestamp' => time()]);
485485
$connection->publish('{}', [], 120000);
486486
}
487487

@@ -513,7 +513,7 @@ public function testAmqpStampHeadersAreUsed()
513513
$amqpExchange = $this->createMock(\AMQPExchange::class)
514514
);
515515

516-
$amqpExchange->expects($this->once())->method('publish')->with('body', null, AMQP_NOPARAM, ['headers' => ['Foo' => 'X', 'Bar' => 'Y'], 'delivery_mode' => 2]);
516+
$amqpExchange->expects($this->once())->method('publish')->with('body', null, AMQP_NOPARAM, ['headers' => ['Foo' => 'X', 'Bar' => 'Y'], 'delivery_mode' => 2, 'timestamp' => time()]);
517517

518518
$connection = Connection::fromDsn('amqp://localhost', [], $factory);
519519
$connection->publish('body', ['Foo' => 'X'], 0, new AmqpStamp(null, AMQP_NOPARAM, ['headers' => ['Bar' => 'Y']]));
@@ -528,7 +528,7 @@ public function testAmqpStampDelireryModeIsUsed()
528528
$amqpExchange = $this->createMock(\AMQPExchange::class)
529529
);
530530

531-
$amqpExchange->expects($this->once())->method('publish')->with('body', null, AMQP_NOPARAM, ['headers' => [], 'delivery_mode' => 1]);
531+
$amqpExchange->expects($this->once())->method('publish')->with('body', null, AMQP_NOPARAM, ['headers' => [], 'delivery_mode' => 1, 'timestamp' => time()]);
532532

533533
$connection = Connection::fromDsn('amqp://localhost', [], $factory);
534534
$connection->publish('body', [], 0, new AmqpStamp(null, AMQP_NOPARAM, ['delivery_mode' => 1]));
@@ -600,7 +600,7 @@ public function testItDelaysTheMessageWithTheInitialSuppliedRoutingKeyAsArgument
600600
$delayQueue->expects($this->once())->method('declareQueue');
601601
$delayQueue->expects($this->once())->method('bind')->with('delays', 'delay_messages_routing_key_120000');
602602

603-
$delayExchange->expects($this->once())->method('publish')->with('{}', 'delay_messages_routing_key_120000', AMQP_NOPARAM, ['headers' => [], 'delivery_mode' => 2]);
603+
$delayExchange->expects($this->once())->method('publish')->with('{}', 'delay_messages_routing_key_120000', AMQP_NOPARAM, ['headers' => [], 'delivery_mode' => 2, 'timestamp' => time()]);
604604
$connection->publish('{}', [], 120000, new AmqpStamp('routing_key'));
605605
}
606606

@@ -617,7 +617,7 @@ public function testItCanPublishWithCustomFlagsAndAttributes()
617617
'body',
618618
'routing_key',
619619
AMQP_IMMEDIATE,
620-
['delivery_mode' => 2, 'headers' => ['type' => DummyMessage::class]]
620+
['delivery_mode' => 2, 'headers' => ['type' => DummyMessage::class], 'timestamp' => time()]
621621
);
622622

623623
$connection = Connection::fromDsn('amqp://localhost', [], $factory);

src/Symfony/Component/Messenger/Bridge/Amqp/Transport/Connection.php

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -317,6 +317,7 @@ private function publishOnExchange(\AMQPExchange $exchange, string $body, string
317317
$attributes = $amqpStamp ? $amqpStamp->getAttributes() : [];
318318
$attributes['headers'] = array_merge($attributes['headers'] ?? [], $headers);
319319
$attributes['delivery_mode'] = $attributes['delivery_mode'] ?? 2;
320+
$attributes['timestamp'] = $attributes['timestamp'] ?? time();
320321

321322
$exchange->publish(
322323
$body,

0 commit comments

Comments
 (0)
0