trjhtr

I have written the following class which is supposed to act as a consumer of messages that are passed in from producers and forward them onto amazon SQS, I have made a compromise by starting the operation at the constructor and waiting and completing on it in the dispose method, I made this compromise to make the clients job of using this class easier and because I thought since the processorTask and the SQS client have to be disposed anyway then it should be fine to use the dispose method to make sure they are finished with their duties.

I would like it to be reviewed for any sort of improvement.

public class BatchMessagePublisher : IMessagePublisher

 private const int PublishBatchSize = 10;

 private readonly IAmazonSQS _sqsClient;
 private readonly ILogger<BatchMessagePublisher> _logger;
 private readonly string _queueUrl;
 private readonly bool _isFifo;
 private readonly BatchBlock<Message> _messageBatcher;
 private readonly Task _processorTask;

 public BatchMessagePublisher(IAmazonSQS sqsClient, ILogger<BatchMessagePublisher> logger, string queueUrl, bool isFifo)
 
 _sqsClient = sqsClient;
 _logger = logger;
 _queueUrl = queueUrl;
 _isFifo = isFifo;

 _messageBatcher = new BatchBlock<Message>(PublishBatchSize, new GroupingDataflowBlockOptions
 
 BoundedCapacity = 1000
 );

 _processorTask = StartAsync();
 


 private SendMessageBatchRequestEntry CreateBatchRequestEntry(Message message)
 
 var request = new SendMessageBatchRequestEntry
 
 Id = message.MessageId,
 MessageBody = message.MessageBody,
 MessageAttributes =
 
 ["MessageType"] = new MessageAttributeValue
 
 DataType = "String",
 StringValue = message.MessageType
 
 
 ;

 if (_isFifo)
 
 request.MessageGroupId = message.GroupId;
 request.MessageDeduplicationId = message.MessageId;
 

 return request;
 

 private async Task StartAsync()
 
 while (!_messageBatcher.Completion.IsCompleted)
 
 var batch = (await _messageBatcher.ReceiveAsync().ConfigureAwait(false))
 .Select(CreateBatchRequestEntry)
 .ToList();

 _logger.LogDebug($"Publishing PublishBatchSize messages...");

 var messageIds = string.Join(", ", batch.Select(r => r.Id));

 try
 
 var batchResponse = await _sqsClient.SendMessageBatchAsync(_queueUrl,batch);

 if (batchResponse.Failed.Any())
 
 var errors = string.Join(", ", batchResponse.Failed.Select(f => f.Message));
 throw new MigrationException(
 $"Failed to publish batch of messages with IDs messageIds: errors");
 

 _logger.LogDebug($"Published batch of messages with IDs messageIds");
 
 catch (AmazonServiceException ex)
 
 throw new MigrationException(
 $"Failed to publish batch of messages with IDs messageIds.", ex);
 
 
 

 public void PublishMessages(params Message messages)
 
 foreach (var message in messages)
 
 _messageBatcher.Post(message);
 
 

 public void Dispose()
 
 _messageBatcher.TriggerBatch();
 _messageBatcher.Complete();
 _processorTask.Wait();
 _processorTask.Dispose();
 _sqsClient.Dispose();