README

This lib allows a simpler integration of AWS's SQS queues through IAM authentication.

It will register a new Messenger Transport for any DSN matching ^https://sqs..

Installation

$ composer require new-immo-group/messenger-transport-sqs-iam

Configuration

Env variables

These env variables will be required:

AWS_REGION=                         # in my case, eu-west-3
AWS_VERSION=                        # generically latest
AWS_SDK_LOAD_NONDEFAULT_CONFIG=true # whether to load .aws/config
AWS_PROFILE=                        # profile in $HOME/.aws/config you want to load
AWS_ACCOUNT_ID=
MESSENGER_TRANSPORT_DSN=https://sqs.${AWS_REGION}.amazonaws.com/${AWS_ACCOUNT_ID}

AWS SDK

The files $HOME/.aws/config and $HOME/.aws/credentials must exists on the host.

Note that if the env variable AWS_SDK_LOAD_NONDEFAULT_CONFIG is set, when loading .aws/credentials the AWS-SDK will also load .aws/config.

If there is an active open_basedir() restriction, make sure to add $HOME/.aws to the allowed path list.

Symfony

Register the SQS transport

services:
  NewImmoGroup\AwsBroker\SqsTransportFactory:
    class: NewImmoGroup\AwsBroker\SqsTransportFactory
    tags: [messenger.transport_factory]
    arguments:
      $defaultOptions:
        auto_setup: true
        queue_prefix: 'some-prefix-'
        queue_tags:
          queue_env: "%env(APP_ENV)%"

We can provide the factory optional default option values:

auto_setup: It controls, at the Transport level, whether the queue is created automatically when a message is sent. Default to false.
queue_prefix: Define a prefix that will be added to the queue name.
queue_tags: A list of tags that will be set on all created queues.

Any of these options can be overridden at the queue level (see below).

Queue configuration example

framework:
  messenger:
    transports:
      async_fail:
        dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
        options:
          queue_name: &failed_queue_name 'async-failed'
          use_extended_configuration: true
          configure_receiver:
            idempotent_explicit_setup: true
            message_retention_period: 43200
            receive_message_wait_time_seconds: 20
            redrive_allow_policy: 'allowAll'
        retry_strategy:
          max_retries: 20

      async:
        dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
        failure_transport: async_fail
        options:
          queue_name: 'async'
          use_extended_configuration: true
          configure_receiver:
            idempotent_explicit_setup: true
            delay_seconds: 0
            message_retention_period: 43200
            receive_message_wait_time_seconds: 20
            redrive_policy:
              max_receive_count: 10
              dead_letter_target_name: *failed_queue_name

Here is the spec of the configuration of the options :

queue_prefix: facultative a queue prefix
queue_name: mandatory the queue name
auto_setup: facultative the option for a queue to be auto-created when it does not exist (mutually exclusive with configure_receiver.idempotent_explicit_setup)
aws_client_options: facultative Options to give to AWS's SDK
use_extended_configuration: true: facultative Opt in the second version of the implementation
configure_receiver: facultative A dictionnary to configure finely a queue from the point of view of the owner of this queue
configure_receiver.idempotent_explicit_setup: facultative When using messenger:setup, should we overwrite an existing queue config to match the description of the queue (mutually exclusive with auto_setup
configure_receiver.idempotent_explicit_setup: facultative When using messenger:setup, should we overwrite an existing queue config to match the description of the queue (mutually exclusive with auto_setup
configure_receiver.tags: facultative A dictionnary of key and value pairs to describe the queue tags. Watch out, true and false should be quoted as strings
configure_receiver.delay_seconds: facultativeAws's DelaySeconds setting
configure_receiver.message_retention_period: facultativeAws's MessageRetentionPeriod setting
configure_receiver.receive_message_wait_time_seconds: facultativeAws's ReceiveMessageWaitTimeSeconds setting
configure_receiver.visibility_timeout: facultativeAws's VisibilityTimeout setting
configure_receiver.maximum_message_size: facultativeAws's MaximumMessageSize setting
configure_receiver.redrive_policy: facultativeAws's RedrivePolicy setting, except the dictionnary it expects has dead_letter_target_name instead of arn
configure_receiver.redrive_allow_policy: facultativeAws's RedriveAllowPolicy setting. It does not accept per queue as setting

For reference, see https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_SetQueueAttributes.html

Prolonging a message's visibility timeout from a handler

Only available when use_extended_configuration: true (the second implementation, SqsTransportUsingFacade).

While a message is being processed, its handler may need more time than the queue's VisibilityTimeout allows before it's considered abandoned and redelivered. To extend it, register NewImmoGroup\AwsBroker\Middleware\MessageProlongationMiddleware on the bus:

framework:
  messenger:
    buses:
      messenger.bus.default:
        middleware:
          - NewImmoGroup\AwsBroker\Middleware\MessageProlongationMiddleware

This same service can then be autowired in a handler to call SQS's ChangeMessageVisibility for the message currently being handled:

final class MyHandler
{
    public function __construct(
        private MessageProlongationMiddleware $prolongation,
    ) {
    }

    public function __invoke(MyMessage $message): void
    {
        // ... long-running work ...
        $this->prolongation->resetMessageVisibilityTo(120); // message becomes visible again in 120s from now
        // ... more long-running work ...
    }
}

It works by attaching a NewImmoGroup\AwsBroker\Entity\MessageProlongationStamp to each received Envelope (a closure bound to the message's ReceiptHandle), and by having the middleware keep track of the Envelope currently being handled so resetMessageVisibilityTo() can retrieve that stamp. Calling resetMessageVisibilityTo() outside of message handling, or on a message received from a transport that isn't the "second implementation", throws a LogicException.

This requires the ChangeMessageVisibility permission (see AWS.SQS Permissions).

CI/Tests

PHP7.4-CLI and PHP8.1-CLI are required.

The end-to-end/functional test will require a valid AWS configuration (which usually expires after a few hours).

make test # Symfony5/PHP7.4
make test@php8 # Symfony5/PHP8.1
make test@sf6 # Symfony6/PHP8.1

AWS.SQS Permissions

Provided user should be able to call :

GetQueueUrl
CreateQueue
SendMessage
DeleteMessage
GetQueueAttribute
SetQueueAttribute
UntagQueue
TagQueue
ListQueueTags

new-immo-group / messenger-transport-sqs-iam

Maintainers

Package info

Statistics

Security