diff --git a/src/Paramore.Brighter.MessagingGateway.AWSSQS/AWSMessagingGateway.cs b/src/Paramore.Brighter.MessagingGateway.AWSSQS/AWSMessagingGateway.cs index 5b237a578..05b045a12 100644 --- a/src/Paramore.Brighter.MessagingGateway.AWSSQS/AWSMessagingGateway.cs +++ b/src/Paramore.Brighter.MessagingGateway.AWSSQS/AWSMessagingGateway.cs @@ -24,7 +24,6 @@ THE SOFTWARE. */ using System; using System.Collections.Generic; -using Amazon.SimpleNotificationService; using Amazon.SimpleNotificationService.Model; using Microsoft.Extensions.Logging; using Paramore.Brighter.Logging; diff --git a/src/Paramore.Brighter.MessagingGateway.AWSSQS/SqsMessageProducer.cs b/src/Paramore.Brighter.MessagingGateway.AWSSQS/SqsMessageProducer.cs index 40d581b20..9bb7043d0 100644 --- a/src/Paramore.Brighter.MessagingGateway.AWSSQS/SqsMessageProducer.cs +++ b/src/Paramore.Brighter.MessagingGateway.AWSSQS/SqsMessageProducer.cs @@ -23,6 +23,7 @@ THE SOFTWARE. */ using System; using System.Collections.Generic; +using System.Threading.Tasks; using Microsoft.Extensions.Logging; namespace Paramore.Brighter.MessagingGateway.AWSSQS @@ -30,7 +31,7 @@ namespace Paramore.Brighter.MessagingGateway.AWSSQS /// /// Class SqsMessageProducer. /// - public class SqsMessageProducer : AWSMessagingGateway, IAmAMessageProducerSync + public class SqsMessageProducer : AWSMessagingGateway, IAmAMessageProducerSync, IAmAMessageProducerAsync { /// /// How many outstanding messages may the outbox have before we terminate the programme with an OutboxLimitReached exception? @@ -92,33 +93,42 @@ public bool ConfirmTopicExists(string topic = null) return !string.IsNullOrEmpty(ChannelTopicArn); } - + /// + /// Sends the specified message. + /// + /// The message. + public async Task SendAsync(Message message) + { + s_logger.LogDebug("SQSMessageProducer: Publishing message with topic {Topic} and id {Id} and message: {Request}", + message.Header.Topic, message.Id, message.Body); + + ConfirmTopicExists(message.Header.Topic); + + using (var client = _clientFactory.CreateSnsClient()) + { + var publisher = new SqsMessagePublisher(ChannelTopicArn, client); + var messageId = await publisher.PublishAsync(message); + if (messageId != null) + { + s_logger.LogDebug( + "SQSMessageProducer: Published message with topic {Topic}, Brighter messageId {MessageId} and SNS messageId {SNSMessageId}", + message.Header.Topic, message.Id, messageId); + return; + } + } + + throw new InvalidOperationException( + string.Format($"Failed to publish message with topic {message.Header.Topic} and id {message.Id} and message: {message.Body}")); + } + + /// /// Sends the specified message. /// /// The message. public void Send(Message message) { - s_logger.LogDebug("SQSMessageProducer: Publishing message with topic {Topic} and id {Id} and message: {Request}", - message.Header.Topic, message.Id, message.Body); - - ConfirmTopicExists(message.Header.Topic); - - using (var client = _clientFactory.CreateSnsClient()) - { - var publisher = new SqsMessagePublisher(ChannelTopicArn, client); - var messageId = publisher.Publish(message); - if (messageId != null) - { - s_logger.LogDebug( - "SQSMessageProducer: Published message with topic {Topic}, Brighter messageId {MessageId} and SNS messageId {SNSMessageId}", - message.Header.Topic, message.Id, messageId); - return; - } - } - - throw new InvalidOperationException( - string.Format($"Failed to publish message with topic {message.Header.Topic} and id {message.Id} and message: {message.Body}")); + SendAsync(message).Wait(); } /// diff --git a/src/Paramore.Brighter.MessagingGateway.AWSSQS/SqsMessagePublisher.cs b/src/Paramore.Brighter.MessagingGateway.AWSSQS/SqsMessagePublisher.cs index 8f07c032f..fd80fa9f5 100644 --- a/src/Paramore.Brighter.MessagingGateway.AWSSQS/SqsMessagePublisher.cs +++ b/src/Paramore.Brighter.MessagingGateway.AWSSQS/SqsMessagePublisher.cs @@ -23,11 +23,10 @@ THE SOFTWARE. */ using System; using System.Collections.Generic; -using System.Net.Mime; using System.Text.Json; +using System.Threading.Tasks; using Amazon.SimpleNotificationService; using Amazon.SimpleNotificationService.Model; -using Paramore.Brighter.Transforms.Transformers; namespace Paramore.Brighter.MessagingGateway.AWSSQS { @@ -42,7 +41,7 @@ public SqsMessagePublisher(string topicArn, AmazonSimpleNotificationServiceClien _client = client; } - public string Publish(Message message) + public async Task PublishAsync(Message message) { var messageString = message.Body.Value; var publishRequest = new PublishRequest(_topicArn, messageString); @@ -65,7 +64,7 @@ public string Publish(Message message) publishRequest.MessageAttributes = messageAttributes; - var response = _client.PublishAsync(publishRequest).GetAwaiter().GetResult(); + var response = await _client.PublishAsync(publishRequest); if (response.HttpStatusCode == System.Net.HttpStatusCode.OK || response.HttpStatusCode == System.Net.HttpStatusCode.Created || response.HttpStatusCode == System.Net.HttpStatusCode.Accepted) { return response.MessageId; @@ -73,5 +72,10 @@ public string Publish(Message message) return null; } + + public string Publish(Message message) + { + return PublishAsync(message).GetAwaiter().GetResult(); + } } } diff --git a/tests/Paramore.Brighter.AWS.Tests/MessagingGateway/When_posting_a_message_via_the_messaging_gateway.cs b/tests/Paramore.Brighter.AWS.Tests/MessagingGateway/When_posting_a_message_via_the_messaging_gateway.cs index 514020956..128948271 100644 --- a/tests/Paramore.Brighter.AWS.Tests/MessagingGateway/When_posting_a_message_via_the_messaging_gateway.cs +++ b/tests/Paramore.Brighter.AWS.Tests/MessagingGateway/When_posting_a_message_via_the_messaging_gateway.cs @@ -57,11 +57,20 @@ public SqsMessageProducerSendTests() - [Fact] - public async Task When_posting_a_message_via_the_producer() + [Theory] + [InlineData(true)] + [InlineData(false)] + public async Task When_posting_a_message_via_the_producer(bool sendAsync) { //arrange - _messageProducer.Send(_message); + if (sendAsync) + { + _messageProducer.SendAsync(_message).Wait(); + } + else + { + _messageProducer.Send(_message); + } await Task.Delay(1000);