8000 [Messenger] Added X-Ray trace header support to the SQS transport · symfony/symfony@e04e70f · GitHub
[go: up one dir, main page]
Skip to content

Commit e04e70f

Browse files
WaylandAceWaylandAce
committed
[Messenger] Added X-Ray trace header support to the SQS transport
1 parent 3ca3de5 commit e04e70f

File tree

5 files changed

+68
-2
lines changed

5 files changed

+68
-2
lines changed

‎src/Symfony/Component/Messenger/Bridge/AmazonSqs/CHANGELOG.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@ CHANGELOG
66

77
* Added new `debug` option to log HTTP requests and responses.
88
* Allowed for receiver & sender injection into AmazonSqsTransport
9+
* Added X-Ray trace header support to the SQS transport
910

1011
5.2.0
1112
-----

‎src/Symfony/Component/Messenger/Bridge/AmazonSqs/Tests/Transport/AmazonSqsSenderTest.php‎

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,7 @@
1515
use Symfony\Component\Messenger\Bridge\AmazonSqs\Tests\Fixtures\DummyMessage;
1616
use Symfony\Component\Messenger\Bridge\AmazonSqs\Transport\AmazonSqsFifoStamp;
1717
use Symfony\Component\Messenger\Bridge\AmazonSqs\Transport\AmazonSqsSender;
18+
use Symfony\Component\Messenger\Bridge\AmazonSqs\Transport\AmazonSqsXrayTraceHeaderStamp;
1819
use Symfony\Component\Messenger\Bridge\AmazonSqs\Transport\Connection;
1920
use Symfony\Component\Messenger\Envelope;
2021
use Symfony\Component\Messenger\Transport\Serialization\SerializerInterface;
@@ -53,4 +54,22 @@ public function testSendWithAmazonSqsFifoStamp()
5354
$sender = new AmazonSqsSender($connection, $serializer);
5455
$sender->send($envelope);
5556
}
57+
58+
public function testSendWithAmazonSqsXrayTraceHeaderStamp()
59+
{
60+
$envelope = (new Envelope(new DummyMessage('Oy')))
61+
->with($stamp = new AmazonSqsXrayTraceHeaderStamp('traceHeader'));
62+
63+
$encoded = ['body' => '...', 'headers' => ['type' => DummyMessage::class]];
64+
65+
$connection = $this->createMock(Connection::class);
66+
$connection->expects($this->once())->method('send')
67+
->with($encoded['body'], $encoded['headers'], 0, null, null, $stamp->getTraceId());
68+
69+
$serializer = $this->createMock(SerializerInterface::class);
70+
$serializer->method('encode')->with($envelope)->willReturnOnConsecutiveCalls($encoded);
71+
72+
$sender = new AmazonSqsSender($connection, $serializer);
73+
$sender->send($envelope);
74+
}
5675
}

‎src/Symfony/Component/Messenger/Bridge/AmazonSqs/Transport/AmazonSqsSender.php‎

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -45,6 +45,7 @@ public function send(Envelope $envelope): Envelope
4545

4646
$messageGroupId = null;
4747
$messageDeduplicationId = null;
48+
$xrayTraceId = null;
4849

4950
/** @var AmazonSqsFifoStamp|null $amazonSqsFifoStamp */
5051
$amazonSqsFifoStamp = $envelope->last(AmazonSqsFifoStamp::class);
@@ -53,13 +54,20 @@ public function send(Envelope $envelope): Envelope
5354
$messageDeduplicationId = $amazonSqsFifoStamp->getMessageDeduplicationId();
5455
}
5556

57+
/** @var AmazonSqsXrayTraceHeaderStamp|null $amazonSqsXrayTraceHeaderStamp */
58+
$amazonSqsXrayTraceHeaderStamp = $envelope->last(AmazonSqsXrayTraceHeaderStamp::class);
59+
if (null !== $amazonSqsXrayTraceHeaderStamp) {
60+
$xrayTraceId = $amazonSqsXrayTraceHeaderStamp->getTraceId();
61+
}
62+
5663
try {
5764
$this->connection->send(
5865
$encodedMessage['body'],
5966
$encodedMessage['headers'] ?? [],
6067
$delay,
6168
$messageGroupId,
62-
$messageDeduplicationId
69+
$messageDeduplicationId,
70+
$xrayTraceId
6371
);
6472
} catch (HttpException $e) {
6573
throw new TransportException($e->getMessage(), 0, $e);

‎src/Symfony/Component/Messenger/Bridge/AmazonSqs/Transport/AmazonSqsXrayTraceHeaderStamp.php‎

Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,29 @@
1+
<?php
2+
3+
/*
4+
* This file is part of the Symfony package.
5+
*
6+
* (c) Fabien Potencier <fabien@symfony.com>
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 Symfony\Component\Messenger\Bridge\AmazonSqs\Transport;
13+
14+
use Symfony\Component\Messenger\Stamp\NonSendableStampInterface;
15+
16+
final class AmazonSqsXrayTraceHeaderStamp implements NonSendableStampInterface
17+
{
18+
private $traceId;
19+
20+
public function __construct(string $traceId)
21+
{
22+
$this->traceId = $traceId;
23+
}
24+
25+
public function getTraceId(): string
26+
{
27+
return $this->traceId;
28+
}
29+
}

‎src/Symfony/Component/Messenger/Bridge/AmazonSqs/Transport/Connection.php‎

Lines changed: 10 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,7 @@
3131
class Connection
3232
{
3333
private const AWS_SQS_FIFO_SUFFIX = '.fifo';
34+
private const AWS_XRAY_TRACEHEADER_NAME = 'AWSTraceHeader';
3435
private const MESSAGE_ATTRIBUTE_NAME = 'X-Symfony-Messenger';
3536

3637
private const DEFAULT_OPTIONS = [
@@ -312,7 +313,7 @@ public function getMessageCount(): int
312313
return (int) ($attributes[QueueAttributeName::APPROXIMATE_NUMBER_OF_MESSAGES] ?? 0);
313314
}
314315

315-
public function send(string $body, array $headers, int $delay = 0, ?string $messageGroupId = null, ?string $messageDeduplicationId = null): void
316+
public function send(string $body, array $headers, int $delay = 0, ?string $messageGroupId = null, ?string $messageDeduplicationId = null, ?string $xrayTraceId = null): void
316317
{
317318
if ($this->configuration['auto_setup']) {
318319
$this->setup();
@@ -323,6 +324,7 @@ public function send(string $body, array $headers, int $delay = 0, ?string $mess
323324
'MessageBody' => $body,
324325
'DelaySeconds' => $delay,
325326
'MessageAttributes' => [],
327+
'MessageSystemAttributes' => [],
326328
];
327329

328330
$specialHeaders = [];
@@ -346,6 +348,13 @@ public function send(string $body, array $headers, int $delay = 0, ?string $mess
346348
]);
347349
}
348350

351+
if (null !== $xrayTraceId) {
352+
$parameters['MessageSystemAttributes'][self::AWS_XRAY_TRACEHEADER_NAME] = new MessageAttributeValue([
353+
'DataType' => 'String',
354+
'StringValue' => $xrayTraceId,
355+
]);
356+
}
357+
349358
if (self::isFifoQueue($this->configuration['queue_name'])) {
350359
$parameters['MessageGroupId'] = null !== $messageGroupId ? $messageGroupId : __METHOD__;
351360
$parameters['MessageDeduplicationId'] = null !== $messageDeduplicationId ? $messageDeduplicationId : sha1(json_encode(['body' => $body, 'headers' => $headers]));

0 commit comments

Comments
 (0)
0