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\$
2
\$\begingroup\$

Cancellation Token Source

You create a token source, and then immediately overwrite it with a new instance, disregarding the previously created instance. Surely, you did not mean to instantiate it twice?

_runningSubscriberCancellationTokenSource = new CancellationTokenSource();
_runningSubscriberCancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);

Thread-Safety

StartAsync and StopAsync are not thread-safe. Concurrent access to these methods may result in a partially running and inconsistent state of the instance. You should use a SemaphoreSlim to block concurrent access to these methods.


Task.FromException

StartAsync and StopAsync are async wrappers of synchronous code. You return return Task.CompletedTask;, however if some exception occurs before the return statement, the exception does not get wrapped as an async task. You should catch exceptions and return return Task.FromException(exception);.


Q&A

Using the while (IsRunning) loop makes me feel to question the code design.

If IsRunning would have been declared as private volatile bool IsRunning ,this code would not be such a bad idea. Then at least you are certain the value of the variable is read atomically on each cycle in the loop. Since you did not declare it as such, it is possible that the value gets cached on the first read and never re-read by the same thread (Volatile keyword in C#).

An important note on the volatile keyword (see link above):

The compiler, the runtime system, and even hardware may rearrange reads and writes to memory locations for performance reasons. Fields that are declared volatile are not subject to these optimizations.

As an alternative, you could use while (!cancellationToken.IsCancellationRequested) and perhaps one last call to cancellationToken.ThrowIfCancellationRequested(); after the loop.

\$\endgroup\$

Your Answer

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

Not the answer you're looking for? Browse other questions tagged or ask your own question.