5
\$\begingroup\$

I have the below code to pull messages from a FIFO SQS queue. Using the while (IsRunning) loop makes me feel to question the code design. Is there a pattern I am missing or some SDK that will do this more efficiently with recovery on error and other features for a subscriber that won't need human intervention to restart?

From AWS SDK, I haven't seen any other way - like pub/sub - but the only way is polling the SQS queue.

namespace Messaging.Services
{
    public class AwsSimpleQueueSubscriber : IMessagingSubscriber
    {
        private readonly string _queueUrl;
        private readonly IAmazonSQS _amazonSqsClient;
        private readonly ILogger<AwsSimpleQueueSubscriber> _logger;
        private readonly int _receiveMessageWaitTime;
        private readonly int _maxNumberOfMessages;

        private Task _runningSubscriberTask;
        private CancellationTokenSource _runningSubscriberCancellationTokenSource;

        public bool IsRunning { get; private set; }

        public AwsSimpleQueueSubscriber(
            string queueUrl,
            IAmazonSQS amazonSqsClient,
            ILogger<AwsSimpleQueueSubscriber> logger,
            int receiveMessageWaitTime = 20,
            int maxNumberOfMessages = 1)
        {
            Guard.Argument(queueUrl, nameof(queueUrl)).NotNull().NotEmpty().NotWhiteSpace();
            Guard.Argument(amazonSqsClient, nameof(amazonSqsClient)).NotNull();
            Guard.Argument(logger, nameof(logger)).NotNull();

            _queueUrl = queueUrl;
            _amazonSqsClient = amazonSqsClient;
            _logger = logger;
            _receiveMessageWaitTime = receiveMessageWaitTime;
            _maxNumberOfMessages = maxNumberOfMessages;
        }

        public Task StartAsync(Func<MessagePayload, Task<bool>> func, CancellationToken cancellationToken = default)
        {
            if (!IsRunning)
            {
                _runningSubscriberCancellationTokenSource = new CancellationTokenSource();
                _runningSubscriberCancellationTokenSource =
                    CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
                _runningSubscriberTask = CheckForMessagesAsync(func, _runningSubscriberCancellationTokenSource.Token);
                IsRunning = true;
            }

            return Task.CompletedTask;
        }

        public Task StopAsync(CancellationToken cancellationToken = default)
        {
            _runningSubscriberCancellationTokenSource.Cancel();
            _runningSubscriberTask = null;
            IsRunning = false;
            return Task.CompletedTask;
        }

        private async Task CheckForMessagesAsync(
            Func<MessagePayload, Task<bool>> func,
            CancellationToken cancellationToken)
        {
            _logger.LogInformation("Checking for SQS messages");

            while (IsRunning)
            {
                try
                {
                    if (cancellationToken.IsCancellationRequested)
                    {
                        cancellationToken.ThrowIfCancellationRequested();
                    }

                    var receiveMessageRequest = new ReceiveMessageRequest
                    {
                        QueueUrl = _queueUrl,
                        WaitTimeSeconds = _receiveMessageWaitTime,
                        MaxNumberOfMessages = _maxNumberOfMessages
                    };

                    var response = await _amazonSqsClient.ReceiveMessageAsync(receiveMessageRequest, cancellationToken);

                    if (response.HttpStatusCode == HttpStatusCode.OK &&
                        response.Messages.Any())
                    {
                        _logger.LogDebug($"Process {response.Messages.Count} messages from {_queueUrl}");

                        foreach (var message in response.Messages)
                        {
                            var funcResult = await func(new MessagePayload {Body = message.Body});

                            if (funcResult)
                            {
                                var deleteMessageRequest = new DeleteMessageRequest
                                {
                                    QueueUrl = _queueUrl,
                                    ReceiptHandle = message.ReceiptHandle
                                };

                                var result =
                                    await _amazonSqsClient.DeleteMessageAsync(deleteMessageRequest, cancellationToken);

                                if (result.HttpStatusCode != HttpStatusCode.OK)
                                {
                                    _logger.LogError(
                                        $"Amazon SQS DELETE ERROR: {response.HttpStatusCode}\r\nQueueURL: {_queueUrl}\r\nReceiptHandle: {message.ReceiptHandle}");
                                }
                            }
                        }
                    }

                    if (response.HttpStatusCode != HttpStatusCode.OK)
                    {
                        _logger.LogError($"Amazon SQS ERROR: {response.HttpStatusCode}\r\n QueueURL: {_queueUrl}");
                    }
                }
                catch (Exception ex)
                {
                    _logger.LogError(ex, $"Error consuming SQS message from {_queueUrl}");

                    if (ex is OperationCanceledException)
                    {
                        IsRunning = false;
                        break;
                    }
                }
            }
        }
    }
}
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.