Initial commit to support Azure Service Bus

Brighter.sln

@@ -187,6 +187,8 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Paramore.Brighter.Extension
 EndProject
 Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Paramore.Brighter.Inbox.Postgres", "src\Paramore.Brighter.Inbox.Postgres\Paramore.Brighter.Inbox.Postgres.csproj", "{DE191819-35B1-40C8-9118-E169633C5041}"
 EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Paramore.Brighter.AzureServiceBus.Tests", "tests\Paramore.Brighter.AzureServiceBus.Tests\Paramore.Brighter.AzureServiceBus.Tests.csproj", "{48F584DF-0BA1-4485-A612-14FD4F6A4CF7}"
+EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
 		Debug|Any CPU = Debug|Any CPU
@@ -965,6 +967,18 @@ Global
 		{DE191819-35B1-40C8-9118-E169633C5041}.Release|Mixed Platforms.Build.0 = Release|Any CPU
 		{DE191819-35B1-40C8-9118-E169633C5041}.Release|x86.ActiveCfg = Release|Any CPU
 		{DE191819-35B1-40C8-9118-E169633C5041}.Release|x86.Build.0 = Release|Any CPU
+		{48F584DF-0BA1-4485-A612-14FD4F6A4CF7}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{48F584DF-0BA1-4485-A612-14FD4F6A4CF7}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{48F584DF-0BA1-4485-A612-14FD4F6A4CF7}.Debug|Mixed Platforms.ActiveCfg = Debug|Any CPU
+		{48F584DF-0BA1-4485-A612-14FD4F6A4CF7}.Debug|Mixed Platforms.Build.0 = Debug|Any CPU
+		{48F584DF-0BA1-4485-A612-14FD4F6A4CF7}.Debug|x86.ActiveCfg = Debug|Any CPU
+		{48F584DF-0BA1-4485-A612-14FD4F6A4CF7}.Debug|x86.Build.0 = Debug|Any CPU
+		{48F584DF-0BA1-4485-A612-14FD4F6A4CF7}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{48F584DF-0BA1-4485-A612-14FD4F6A4CF7}.Release|Any CPU.Build.0 = Release|Any CPU
+		{48F584DF-0BA1-4485-A612-14FD4F6A4CF7}.Release|Mixed Platforms.ActiveCfg = Release|Any CPU
+		{48F584DF-0BA1-4485-A612-14FD4F6A4CF7}.Release|Mixed Platforms.Build.0 = Release|Any CPU
+		{48F584DF-0BA1-4485-A612-14FD4F6A4CF7}.Release|x86.ActiveCfg = Release|Any CPU
+		{48F584DF-0BA1-4485-A612-14FD4F6A4CF7}.Release|x86.Build.0 = Release|Any CPU
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE
@@ -1021,6 +1035,7 @@ Global
 		{B853A06F-5DFD-4093-979B-B08D1BBDD2D6} = {329736D2-BF92-4D06-A7BF-19F4B6B64EDD}
 		{C2355F42-792D-4964-B068-79186F887C6D} = {329736D2-BF92-4D06-A7BF-19F4B6B64EDD}
 		{D2B52F80-5106-4981-9031-A02C5EB479B4} = {329736D2-BF92-4D06-A7BF-19F4B6B64EDD}
+		{48F584DF-0BA1-4485-A612-14FD4F6A4CF7} = {329736D2-BF92-4D06-A7BF-19F4B6B64EDD}
 	EndGlobalSection
 	GlobalSection(ExtensibilityGlobals) = postSolution
 		SolutionGuid = {8B7C7E31-2E32-4E0D-9426-BC9AF22E9F4C}

src/Paramore.Brighter.MessagingGateway.AzureServiceBus/AzureServiceBusChannelFactory.cs

@@ -0,0 +1,35 @@
+using System;
+
+namespace Paramore.Brighter.MessagingGateway.AzureServiceBus
+{
+    public class AzureServiceBusChannelFactory : IAmAChannelFactory
+    {
+        private readonly AzureServiceBusConsumerFactory _azureServiceBusConsumerFactory;
+
+        public AzureServiceBusChannelFactory(AzureServiceBusConsumerFactory azureServiceBusConsumerFactory)
+        {
+            _azureServiceBusConsumerFactory = azureServiceBusConsumerFactory;
+        }
+
+        public IAmAChannel CreateChannel(Subscription subscription)
+        {
+            if (!(subscription is AzureServiceBusSubscription azureServiceBusSubscription))
+            {
+                throw new ConfigurationException("We expect an AzureServiceBusSubscription or AzureServiceBusSubscription<T> as a parameter");
+            }
+
+            if (subscription.TimeoutInMiliseconds < 400)
+            {
+                throw new ArgumentException("The minimum allowed timeout is 400 milliseconds");
+            }
+
+            IAmAMessageConsumer messageConsumer = _azureServiceBusConsumerFactory.Create(azureServiceBusSubscription);
+
+            return new Channel(
+                channelName: subscription.ChannelName,
+                messageConsumer: messageConsumer,
+                maxQueueLength: subscription.BufferSize
+            );
+        }
+    }
+}

src/Paramore.Brighter.MessagingGateway.AzureServiceBus/AzureServiceBusConfiguration.cs

@@ -0,0 +1,12 @@
+namespace Paramore.Brighter.MessagingGateway.AzureServiceBus
+{
+    public class AzureServiceBusConfiguration
+    {
+        public AzureServiceBusConfiguration(string connectionString)
+        {
+            ConnectionString = connectionString;
+        }
+
+        public string ConnectionString { get; }
+    }
+}

src/Paramore.Brighter.MessagingGateway.AzureServiceBus/AzureServiceBusConsumer.cs

@@ -0,0 +1,193 @@
+using System;
+using System.Collections.Generic;
+using Microsoft.Azure.ServiceBus;
+using Microsoft.Azure.ServiceBus.Management;
+using Paramore.Brighter.Logging;
+using Paramore.Brighter.MessagingGateway.AzureServiceBus.AzureServiceBusWrappers;
+
+namespace Paramore.Brighter.MessagingGateway.AzureServiceBus
+{
+    public class AzureServiceBusConsumer : IAmAMessageConsumer
+    {
+        private readonly string _topicName;
+        private readonly IAmAMessageProducer _messageProducer;
+        private readonly IManagementClientWrapper _managementClientWrapper;
+        private readonly IMessageReceiverProvider _messageReceiverProvider;
+        private readonly int _batchSize;
+        private IMessageReceiverWrapper _messageReceiver;
+        private readonly string _subscriptionName;
+        private bool _subscriptionCreated;
+        private static readonly Lazy<ILog> _logger = new Lazy<ILog>(LogProvider.For<AzureServiceBusConsumer>);
+
+        public AzureServiceBusConsumer(string topicName, string subscriptionName, IAmAMessageProducer messageProducer, IManagementClientWrapper managementClientWrapper, 
+            IMessageReceiverProvider messageReceiverProvider, int batchSize = 10)
+        {
+            _subscriptionName = subscriptionName;
+            _topicName = topicName;
+            _messageProducer = messageProducer;
+            _managementClientWrapper = managementClientWrapper;
+            _messageReceiverProvider = messageReceiverProvider;
+            _batchSize = batchSize;
+
+            GetMessageReceiverProvider();
+        }
+
+        private void GetMessageReceiverProvider()
+        {
+            _logger.Value.Info($"Getting message receiver provider for topic {_topicName} and subscription {_subscriptionName}...");
+            try
+            {
+                _messageReceiver = _messageReceiverProvider.Get(_topicName, _subscriptionName, ReceiveMode.ReceiveAndDelete);
+            }
+            catch (Exception e)
+            {
+                _logger.Value.ErrorException($"Failed to get message receiver provider for topic {_topicName} and subscription {_subscriptionName}.", e);
+            }
+        }
+
+        public Message[] Receive(int timeoutInMilliseconds)
+        {
+            _logger.Value.Debug($"Preparing to retrieve next message(s) from topic {_topicName} via subscription {_subscriptionName} with timeout {timeoutInMilliseconds} and batch size {_batchSize}.");
+
+            IEnumerable<IBrokeredMessageWrapper> messages;
+            EnsureSubscription();
+
+            var messagesToReturn = new List<Message>();
+
+            try
+            {
+                messages = _messageReceiver.Receive(_batchSize, TimeSpan.FromMilliseconds(timeoutInMilliseconds)).GetAwaiter().GetResult();
+            }
+            catch (Exception e)
+            {
+                if (_messageReceiver.IsClosedOrClosing)
+                {
+                    _logger.Value.Debug("Message Receiver is closing...");
+                    var message = new Message(new MessageHeader(Guid.NewGuid(), _topicName, MessageType.MT_QUIT), new MessageBody(string.Empty));
+                    messagesToReturn.Add(message);
+                    return messagesToReturn.ToArray();
+                }
+
+                _logger.Value.ErrorException("Failing to receive messages.", e);
+
+                //The connection to Azure Service bus may have failed so we re-establish the connection.
+                GetMessageReceiverProvider();
+
+                throw new ChannelFailureException("Failing to receive messages.", e);
+            }
+
+            foreach (IBrokeredMessageWrapper azureServiceBusMessage in messages)
+            {
+                Message message = MapToBrighterMessage(azureServiceBusMessage);
+                messagesToReturn.Add(message);
+            }
+
+            return messagesToReturn.ToArray();
+        }
+
+        private Message MapToBrighterMessage(IBrokeredMessageWrapper azureServiceBusMessage)
+        {
+            if (azureServiceBusMessage.MessageBodyValue == null)
+            {
+                _logger.Value.Warn($"Null message body received from topic {_topicName} via subscription {_subscriptionName}.");
+            }
+
+            var messageBody = System.Text.Encoding.Default.GetString(azureServiceBusMessage.MessageBodyValue ?? Array.Empty<byte>());
+            _logger.Value.Debug($"Received message from topic {_topicName} via subscription {_subscriptionName} with body {messageBody}.");
+            MessageType messageType = GetMessageType(azureServiceBusMessage);
+            var message = new Message(new MessageHeader(Guid.NewGuid(), _topicName, messageType), new MessageBody(messageBody));
+            return message;
+        }
+
+        private static MessageType GetMessageType(IBrokeredMessageWrapper azureServiceBusMessage)
+        {
+            if (azureServiceBusMessage.UserProperties == null ||!azureServiceBusMessage.UserProperties.ContainsKey("MessageType")) 
+                return MessageType.MT_EVENT;
+
+            if (Enum.TryParse(azureServiceBusMessage.UserProperties["MessageType"].ToString(), true, out MessageType messageType))
+                return messageType;
+
+            return MessageType.MT_EVENT;
+        }
+
+        private void EnsureSubscription()
+        {
+            const int maxDeliveryCount = 2000;
+
+            if (_subscriptionCreated)
+                return;
+
+            try
+            {
+                if (_managementClientWrapper.SubscriptionExists(_topicName, _subscriptionName))
+                {
+                    _subscriptionCreated = true;
+                    return;
+                }
+
+                _managementClientWrapper.CreateSubscription(_topicName, _subscriptionName, maxDeliveryCount);
+                _subscriptionCreated = true;
+            }
+            catch (MessagingEntityAlreadyExistsException)
+            {
+                _logger.Value.Warn($"Message entity already exists with topic {_topicName} and subscription {_subscriptionName}.");
+                _subscriptionCreated = true;
+            }
+            catch (Exception e)
+            {
+                _logger.Value.ErrorException("Failing to check or create subscription.", e);
+
+                //The connection to Azure Service bus may have failed so we re-establish the connection.
+                _managementClientWrapper.Reset();
+
+                throw new ChannelFailureException("Failing to check or create subscription", e);
+            }
+        }
+
+        public void Requeue(Message message, int delayMilliseconds)
+        {
+            var topic = message.Header.Topic;
+
+            _logger.Value.Info($"Requeuing message with topic {topic} and id {message.Id}.");
+
+            if (delayMilliseconds > 0)
+            {
+                _messageProducer.SendWithDelay(message, delayMilliseconds);
+            }
+            else
+            {
+                _messageProducer.Send(message);
+            }
+        }
+
+        public void Acknowledge(Message message)
+        {
+            //Not implemented as we use ReceiveMode.ReceiveAndDelete (Brighter will call this method anyway)
+        }
+
+        public void Reject(Message message)
+        {
+            _logger.Value.Warn("Reject method NOT IMPLEMENTED.");
+        }
+
+        public void Reject(Message message, bool requeue)
+        {
+            if (requeue)
+            {
+                Requeue(message, 0);
+            }
+        }
+
+        public void Purge()
+        {
+            _logger.Value.Warn("Purge method NOT IMPLEMENTED.");
+        }
+
+        public void Dispose()
+        {
+            _logger.Value.Info("Disposing the consumer...");
+            _messageReceiver.Close();
+            _logger.Value.Info("Consumer disposed.");
+        }
+    }
+}

src/Paramore.Brighter.MessagingGateway.AzureServiceBus/AzureServiceBusConsumerFactory.cs

@@ -0,0 +1,24 @@
+using Paramore.Brighter.MessagingGateway.AzureServiceBus.AzureServiceBusWrappers;
+
+namespace Paramore.Brighter.MessagingGateway.AzureServiceBus
+{
+    public class AzureServiceBusConsumerFactory : IAmAMessageConsumerFactory
+    {
+        private readonly AzureServiceBusConfiguration _configuration;
+
+        public AzureServiceBusConsumerFactory(AzureServiceBusConfiguration configuration)
+        {
+            _configuration = configuration;
+        }
+
+        public IAmAMessageConsumer Create(Subscription subscription)
+        {
+            var nameSpaceManagerWrapper = new ManagementClientWrapper(_configuration);
+
+            return new AzureServiceBusConsumer(subscription.RoutingKey, subscription.ChannelName,
+                new AzureServiceBusMessageProducer(nameSpaceManagerWrapper,
+                    new TopicClientProvider(_configuration)), nameSpaceManagerWrapper,
+                new MessageReceiverProvider(_configuration));
+        }
+    }
+}