SQS Send Message Batch

[brendonparker]

Describe the feature

Allow for sending a batch of messages (SendMessageBatch)

This was previously opened, but then closed due to inactivity: #264

Use Case

Without message batching you are limited to 300 operations per second (300 send, receive, or delete operations per second).
With batching (using SendMessageBatch): Up to 3,000 messages per second.

Anytime your enqueuing a list of things to do, batching is preferred.

Proposed Solution

• Add SendBatchAsync to ISQSPublisher
  • Accept a List<T> return a List<SQSSendResponse>
  • Note: Unsure how best to handle SQSOptions? as this can contain message-level properties (deduplication id, etc). So looking for guidance there.

Other Information

This was previously opened in SQSSendResponse

Acknowledgements

• I may be able to implement this feature request
• This feature might incur a breaking change

AWS.Messaging (or related) package versions

AWS.Messaging 1.0.1

Targeted .NET Platform

All

Operating System and version

All

[peterrsongg]

seems like a valid feature request. Will review with team on priority

[peterrsongg]

If you're willing to contribute a PR we'd be happy to take a look, but this prioritized in our backlog.

[brendonparker]

I have a somewhat hacky implementation going. There are some parts around how patterns/interfaces/contracts are setup that introduce some difficulty.

public async Task SendBatchAsync<T>(
    List<T> messages,
    SQSOptions? sqsOptions,
    CancellationToken token = default
)
{
    using var trace = _telemetryFactory.Trace("Publish to AWS SQS");
    try
    {
        trace.AddMetadata(TelemetryKeys.ObjectType, typeof(T).FullName!);

        _logger.LogDebug(
            "Publishing the message of type '{MessageType}' using the {PublisherType}.",
            typeof(T),
            nameof(CustomSQSPublisher)
        );

        var queueUrl = GetPublisherEndpoint(trace, typeof(T), sqsOptions);

        List<string> messageBodies = [];
        foreach (var message in messages)
        {
            var messageEnvelope = await _envelopeSerializer.CreateEnvelopeAsync(message);
            var messageBody = await _envelopeSerializer.SerializeAsync(messageEnvelope);
            messageBodies.Add(messageBody);
        }

        var client = InitClient(sqsOptions);

        _logger.LogDebug(
            "Sending the message of type '{MessageType}' to SQS. Publisher Endpoint: {Endpoint}",
            typeof(T),
            queueUrl
        );
        var sendMessageBatchRequest = CreateSendMessageBatchRequest(
            queueUrl,
            messageBodies,
            sqsOptions
        );
        var response = await client.SendMessageBatchAsync(sendMessageBatchRequest, token);
        var failCount = response.Failed?.Count ?? 0;
        _logger.LogDebug(
            "The message batch of type '{MessageType}' has been pushed to SQS. {ErrorCount} messages failed. {SuccessCount} messages succeeded.",
            typeof(T),
            failCount,
            response.Successful?.Count ?? 0
        );
        if (failCount > 0)
        {
            foreach (var failed in response.Failed!)
            {
                _logger.LogError(
                    "Failed to send message with Id '{MessageId}' to SQS. Error Code: {ErrorCode}, Error Message: {ErrorMessage}",
                    failed.Id,
                    failed.Code,
                    failed.Message
                );
            }
        }
    }
    catch (Exception ex)
    {
        trace.AddException(ex);
        if (ex is AmazonSQSException)
            throw new FailedToPublishException("Message failed to publish.", ex);
        throw;
    }
}

Note:
Because SendAsync takes a T message and SQSOptions? sqsOptions, I assume you'd want to follow the same pattern.
However, ideally SendBatchAsync could be use to send one batch of messages, with different messageGroupId's. Maybe that is just my uncommon usecase with fifo queues 🤷
The implementation above does not do that

Initial pass at CreateSendMessageBatchRequest:

private SendMessageBatchRequest CreateSendMessageBatchRequest(
    string queueUrl,
    List<string> messageBodies,
    SQSOptions? sqsOptions
)
{
    var request = new SendMessageBatchRequest { QueueUrl = queueUrl, Entries = [] };

    if (queueUrl.EndsWith(FIFO_SUFFIX) && string.IsNullOrEmpty(sqsOptions?.MessageGroupId))
    {
        var errorMessage =
            $"You are attempting to send to a FIFO SQS queue but the request does not include a message group ID. "
            + $"Please use {nameof(ISQSPublisher)} from the service collection to send to FIFO queues. "
            + $"It exposes a {nameof(SendAsync)} method that accepts {nameof(SQSOptions)} as a parameter. "
            + $"A message group ID must be specified via {nameof(SQSOptions.MessageGroupId)}. "
            + $"Additionally, {nameof(SQSOptions.MessageDeduplicationId)} must also be specified if content based de-duplication is not enabled on the queue.";

        _logger.LogError(errorMessage);
        throw new InvalidFifoPublishingRequestException(errorMessage);
    }

    foreach (var messageBody in messageBodies)
    {
        request.Entries.Add(
            new SendMessageBatchRequestEntry
            {
                Id = Guid.NewGuid().ToString(),
                MessageBody = messageBody,
            }
        );
        if (sqsOptions is null)
            continue;

        if (!string.IsNullOrEmpty(sqsOptions.MessageGroupId))
            request.Entries[^1].MessageGroupId = sqsOptions.MessageGroupId;

        if (sqsOptions.DelaySeconds.HasValue)
            request.Entries[^1].DelaySeconds = (int)sqsOptions.DelaySeconds;

        if (sqsOptions.MessageAttributes is not null)
            request.Entries[^1].MessageAttributes = sqsOptions.MessageAttributes;
    }

    return request;
}