Fix invalid char in SQS Headers
This commit is contained in:
parent
e5b5d9ea14
commit
76a18b0b47
@ -45,7 +45,7 @@ class AmazonSqsIntegrationTest extends TestCase
|
||||
$connection->setup();
|
||||
$this->clearSqs($dsn);
|
||||
|
||||
$connection->send('{"message": "Hi"}', ['type' => DummyMessage::class]);
|
||||
$connection->send('{"message": "Hi"}', ['type' => DummyMessage::class, DummyMessage::class => 'special']);
|
||||
$this->assertSame(1, $connection->getMessageCount());
|
||||
|
||||
$wait = 0;
|
||||
@ -54,7 +54,7 @@ class AmazonSqsIntegrationTest extends TestCase
|
||||
}
|
||||
|
||||
$this->assertEquals('{"message": "Hi"}', $encoded['body']);
|
||||
$this->assertEquals(['type' => DummyMessage::class], $encoded['headers']);
|
||||
$this->assertEquals(['type' => DummyMessage::class, DummyMessage::class => 'special'], $encoded['headers']);
|
||||
}
|
||||
|
||||
private function clearSqs(string $dsn): void
|
||||
|
@ -30,6 +30,7 @@ use Symfony\Contracts\HttpClient\HttpClientInterface;
|
||||
class Connection
|
||||
{
|
||||
private const AWS_SQS_FIFO_SUFFIX = '.fifo';
|
||||
private const MESSAGE_ATTRIBUTE_NAME = 'X-Symfony-Messenger';
|
||||
|
||||
private const DEFAULT_OPTIONS = [
|
||||
'buffer_size' => 9,
|
||||
@ -200,7 +201,12 @@ class Connection
|
||||
|
||||
foreach ($this->currentResponse->getMessages() as $message) {
|
||||
$headers = [];
|
||||
foreach ($message->getMessageAttributes() as $name => $attribute) {
|
||||
$attributes = $message->getMessageAttributes();
|
||||
if (isset($attributes[self::MESSAGE_ATTRIBUTE_NAME]) && 'String' === $attributes[self::MESSAGE_ATTRIBUTE_NAME]->getDataType()) {
|
||||
$headers = json_decode($attributes[self::MESSAGE_ATTRIBUTE_NAME]->getStringValue(), true);
|
||||
unset($attributes[self::MESSAGE_ATTRIBUTE_NAME]);
|
||||
}
|
||||
foreach ($attributes as $name => $attribute) {
|
||||
if ('String' !== $attribute->getDataType()) {
|
||||
continue;
|
||||
}
|
||||
@ -284,13 +290,29 @@ class Connection
|
||||
'MessageAttributes' => [],
|
||||
];
|
||||
|
||||
$specialHeaders = [];
|
||||
foreach ($headers as $name => $value) {
|
||||
if ('.' === $name[0] || self::MESSAGE_ATTRIBUTE_NAME === $name || \strlen($name) > 256 || '.' === substr($name, -1) || 'AWS.' === substr($name, 0, \strlen('AWS.')) || 'Amazon.' === substr($name, 0, \strlen('Amazon.')) || preg_match('/([^a-zA-Z0-9_\.-]+|\.\.)/', $name)) {
|
||||
$specialHeaders[$name] = $value;
|
||||
|
||||
continue;
|
||||
}
|
||||
|
||||
$parameters['MessageAttributes'][$name] = new MessageAttributeValue([
|
||||
'DataType' => 'String',
|
||||
'StringValue' => $value,
|
||||
]);
|
||||
}
|
||||
|
||||
if (!empty($specialHeaders)) {
|
||||
$parameters['MessageAttributes'][self::MESSAGE_ATTRIBUTE_NAME] = new MessageAttributeValue([
|
||||
'DataType' => 'String',
|
||||
'StringValue' => json_encode($specialHeaders),
|
||||
]);
|
||||
}
|
||||
|
||||
dd($parameters);
|
||||
|
||||
if (self::isFifoQueue($this->configuration['queue_name'])) {
|
||||
$parameters['MessageGroupId'] = null !== $messageGroupId ? $messageGroupId : __METHOD__;
|
||||
$parameters['MessageDeduplicationId'] = null !== $messageDeduplicationId ? $messageDeduplicationId : sha1(json_encode(['body' => $body, 'headers' => $headers]));
|
||||
|
Reference in New Issue
Block a user