Skip to content

[Messenger] Allow auto creation of sqs queue #58575

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 2 commits into
base: 7.4
Choose a base branch
from
Open

[Messenger] Allow auto creation of sqs queue #58575

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
488fe38
Allow auto creation of sqs queue
kevinrudde Oct 15, 2024
479b18b
Add AWS_ENDPOINT_URL to integration-tests which points to localstack
kevinrudde Oct 16, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions .github/workflows/integration-tests.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -243,6 +243,7 @@ jobs:
MESSENGER_AMQP_DSN: amqp://localhost/%2f/messages
MESSENGER_SQS_DSN: "sqs://localhost:4566/messages?sslmode=disable&poll_timeout=0.01"
MESSENGER_SQS_FIFO_QUEUE_DSN: "sqs://localhost:4566/messages.fifo?sslmode=disable&poll_timeout=0.01"
AWS_ENDPOINT_URL: "http://localhost:4566"
KAFKA_BROKER: 127.0.0.1:9092
POSTGRES_HOST: localhost
PGBOUNCER_HOST: localhost:6432
Expand Down
11 changes: 9 additions & 2 deletions src/Symfony/Component/Messenger/Bridge/AmazonSqs/Transport/Connection.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@

namespace Symfony\Component\Messenger\Bridge\AmazonSqs\Transport;

use AsyncAws\Core\Sts\StsClient;
use AsyncAws\Sqs\Enum\MessageSystemAttributeName;
use AsyncAws\Sqs\Enum\QueueAttributeName;
use AsyncAws\Sqs\Result\ReceiveMessageResult;
Expand Down Expand Up @@ -53,6 +54,7 @@ class Connection

private array $configuration;
private SqsClient $client;
private StsClient $stsClient;
private ?ReceiveMessageResult $currentResponse = null;
/** @var array[] */
private array $buffer = [];
Expand All @@ -61,9 +63,11 @@ public function __construct(
array $configuration,
?SqsClient $client = null,
private ?string $queueUrl = null,
?StsClient $stsClient = null,
) {
$this->configuration = array_replace_recursive(self::DEFAULT_OPTIONS, $configuration);
$this->client = $client ?? new SqsClient([]);
$this->stsClient = $stsClient ?? new StsClient([]);
}

public function __sleep(): array
Expand Down Expand Up @@ -274,8 +278,11 @@ public function setup(): void
return;
}

if (null !== $this->configuration['account']) {
throw new InvalidArgumentException(\sprintf('The Amazon SQS queue "%s" does not exist (or you don\'t have permissions on it), and can\'t be created when an account is provided.', $this->configuration['queue_name']));
$accountId = $this->stsClient->getCallerIdentity()->getAccount();

// If the queue does not exist, we can create it only if the account is the same
if (null !== $this->configuration['account'] && $accountId !== $this->configuration['account']) {
throw new InvalidArgumentException(\sprintf('The Amazon SQS queue "%s" does not exist (or you don\'t have permissions on it), and can\'t be created when an other account is provided.', $this->configuration['queue_name']));
}

$parameters = ['QueueName' => $this->configuration['queue_name']];
Expand Down