feat: added simple test for ServiceBus

This is just a draft for testing the ServiceBus implementation.

Author: bruno-brant

src/Liquid.Activation/Worker/LightWorker.cs

@@ -25,7 +25,7 @@ public abstract class LightWorker : LightBackgroundTask, ILightWorker
         protected readonly static Dictionary<MethodInfo, QueueAttribute> _queues = new Dictionary<MethodInfo, QueueAttribute>();
         protected readonly static Dictionary<MethodInfo, TopicAttribute> _topics = new Dictionary<MethodInfo, TopicAttribute>();
         private readonly List<string> _inputValidationErrors = new List<string>();
-        protected ILightTelemetry Telemetry { get; } = Workbench.Instance.Telemetry != null ? (ILightTelemetry)Workbench.Instance.Telemetry.CloneService() : null;
+        protected ILightTelemetry Telemetry => Workbench.Instance.Telemetry;
         protected ILightCache Cache => Workbench.Instance.Cache;
         //Instance of CriticHandler to inject on the others classes
         private readonly CriticHandler _criticHandler = new CriticHandler();

src/Liquid.OnAzure/Liquid.OnAzure.csproj

@@ -15,6 +15,9 @@
     <DebugType>full</DebugType>
     <DebugSymbols>true</DebugSymbols>
   </PropertyGroup>
+  <ItemGroup>
+    <AdditionalFiles Include="..\..\stylecop.json" Link="stylecop.json" />
+  </ItemGroup>
   <ItemGroup>
     <Compile Remove="Telemetry\AppInsightsTelemetryMiddleware.cs" />
   </ItemGroup>
@@ -29,6 +32,10 @@
     <PackageReference Include="Microsoft.CodeAnalysis.FxCopAnalyzers" Version="2.9.3" />
     <PackageReference Include="Microsoft.Extensions.Caching.Abstractions" Version="2.2.0" />
     <PackageReference Include="Microsoft.Extensions.Caching.Redis" Version="2.2.0" />
+    <PackageReference Include="StyleCop.Analyzers" Version="1.2.0-beta.113">
+      <PrivateAssets>all</PrivateAssets>
+      <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
+    </PackageReference>
     <PackageReference Include="WindowsAzure.Storage" Version="9.3.3" />
   </ItemGroup>
   <ItemGroup>

src/Liquid.OnAzure/MessageBuses/ServiceBus.cs

@@ -1,23 +1,125 @@
-using Liquid.Base.Interfaces;
+// Copyright (c) Avanade Inc. All rights reserved.
+// Licensed under the MIT License. See LICENSE in the project root for license information.
+
+using System;
+using System.Collections.Generic;
+using System.Reflection;
+using System.Threading.Tasks;
+using Liquid.Activation;
 using Liquid.Domain;
 using Liquid.Domain.Base;
-using Liquid.Activation;
 using Liquid.Runtime.Configuration.Base;
 using Liquid.Runtime.Telemetry;
 using Microsoft.Azure.ServiceBus;
-using System;
-using System.Collections.Generic;
-using System.Reflection;
-using System.Text;
-using System.Threading.Tasks;
 
 namespace Liquid.OnAzure
 {
     /// <summary>
-    /// Implementation of the communication component between queues and topics of the Azure, this class is specific to azure
+    /// Defines an object capable of creating instances of <see cref="IQueueClient"/>.
+    /// </summary>
+    public interface IQueueClientFactory
+    {
+        /// <summary>
+        /// Creates a new instance of <see cref="IQueueClient"/>.
+        /// </summary>
+        /// <param name="connectionString">The connection string for the client.</param>
+        /// <param name="queueName">The name of the queue.</param>
+        /// <param name="receiveMode">The receive mode that the client will connect to the queue.</param>
+        /// <returns>A new instance of <see cref="IQueueClient"/>.</returns>
+        IQueueClient CreateClient(string connectionString, string queueName, ReceiveMode receiveMode);
+    }
+
+    /// <summary>
+    /// Defines an object capable of creating instances of <see cref="ISubscriptionClient"/>.
+    /// </summary>
+    public interface ISubscriptionClientFactory
+    {
+        /// <summary>
+        /// Creates a new instance of <see cref="ISubscriptionClient"/>.
+        /// </summary>
+        /// <param name="connectionString">The connection string for the client.</param>
+        /// <param name="topicName">The name of the topic to connect to.</param>
+        /// <param name="subscriptionName">Identifies the subscription to this topic.</param>
+        /// <param name="receiveMode">The receive mode that the client will connect to the queue.</param>
+        /// <returns>A new instance of <see cref="ISubscriptionClient"/>.</returns>
+        ISubscriptionClient CreateClient(string connectionString, string topicName, string subscriptionName, ReceiveMode receiveMode);
+    }
+
+    /// <summary>
+    /// Configuration source for <see cref="ServiceBus"/>.
+    /// </summary>
+    // TODO: should remove this class once we move to .NET configuration system
+    public interface IServiceBusConfigurationProvider
+    {
+        /// <summary>
+        /// Gets the configuration for a <see cref="ServiceBus"/>.
+        /// </summary>
+        /// <returns>
+        /// The current configuration for a service bus.
+        /// </returns>
+        ServiceBusConfiguration GetConfiguration();
+
+        /// <summary>
+        /// Gets the configuration for a <see cref="ServiceBus"/>.
+        /// </summary>
+        /// <param name="connectionName">
+        /// Identifies which connection should be retrieved from the file.
+        /// </param>
+        /// <returns>
+        /// The current configuration for a service bus.
+        /// </returns>
+        ServiceBusConfiguration GetConfiguration(string connectionName);
+    }
+
+    /// <summary>
+    /// Implementation of the communication component between queues and topics of the Azure, this class is specific to azure.
     /// </summary>
     public class ServiceBus : LightWorker, IWorkbenchService
     {
+        /// <summary>
+        /// Factory used to create a <see cref="IQueueClient"/>.
+        /// </summary>
+        private readonly IQueueClientFactory _queueClientFactory = new DefaultQueueClientFactory();
+
+        /// <summary>
+        /// Factory used to create a <see cref="ISubscriptionClient"/>.
+        /// </summary>
+        private readonly ISubscriptionClientFactory _subscriptionClientFactory = new DefaultSubscriptionClientFactory();
+
+        /// <summary>
+        /// Service that retrives a <see cref="ServiceBusConfiguration"/>.
+        /// </summary>
+        private readonly IServiceBusConfigurationProvider _configurationProvider = new DefaultServiceBusConfigurationProvider();
+
+        /// <summary>
+        /// Initializes a new instance of the <see cref="ServiceBus"/> class.
+        /// </summary>
+        public ServiceBus()
+        {
+        }
+
+        /// <summary>
+        /// Initializes a new instance of the <see cref="ServiceBus"/> class.
+        /// </summary>
+        /// <param name="queueClientFactory">
+        /// Dependency. Used to obtain new instances of a <see cref="IQueueClient"/>.
+        /// </param>
+        /// <param name="subscriptionClientFactory">
+        /// Dependency. Used to obtain new instances of a <see cref="ISubscriptionClient"/>.
+        /// </param>
+        /// <param name="configurationProvider">
+        /// Dependency. Used to retrieve a configuration for this class.
+        /// </param>
+        public ServiceBus(
+            IQueueClientFactory queueClientFactory,
+            ISubscriptionClientFactory subscriptionClientFactory,
+            IServiceBusConfigurationProvider configurationProvider)
+        {
+            _queueClientFactory = queueClientFactory ?? throw new ArgumentNullException(nameof(queueClientFactory));
+            _subscriptionClientFactory = subscriptionClientFactory ?? throw new ArgumentNullException(nameof(subscriptionClientFactory));
+            _configurationProvider = configurationProvider ?? throw new ArgumentNullException(nameof(configurationProvider));
+        }
+
         /// <summary>
         /// Implementation of the start process queue and process topic. It must be called  parent before start processes.
         /// </summary>
@@ -36,16 +138,17 @@ public override void Initialize()
         /// <returns>StringConnection of the ServiceBus</returns>
         private string GetConnection<T>(KeyValuePair<MethodInfo, T> item)
         {
-            MethodInfo method = item.Key;
-            string connectionKey = GetKeyConnection(method);
-            ServiceBusConfiguration config = null;
+            var method = item.Key;
+            var connectionKey = GetKeyConnection(method);
+
+            ServiceBusConfiguration config;
             if (string.IsNullOrEmpty(connectionKey)) // Load specific settings if provided
             {
-                config = LightConfigurator.Config<ServiceBusConfiguration>($"{nameof(ServiceBus)}");
+                config = _configurationProvider.GetConfiguration();//LightConfigurator.Config<ServiceBusConfiguration>($"{nameof(ServiceBus)}");
             }
             else
             {
-                config = LightConfigurator.Config<ServiceBusConfiguration>($"{nameof(ServiceBus)}_{connectionKey}");
+                config = _configurationProvider.GetConfiguration(connectionKey);//LightConfigurator.Config<ServiceBusConfiguration>($"{nameof(ServiceBus)}_{connectionKey}");
             }
 
             return config.ConnectionString;
@@ -80,7 +183,7 @@ public void ProcessQueue()
                     int takeQuantity = queue.Value.TakeQuantity;
 
                     //Register Trace on the telemetry 
-                    QueueClient queueReceiver = new QueueClient(GetConnection(queue), queueName, receiveMode);
+                    var queueReceiver = _queueClientFactory.CreateClient(GetConnection(queue), queueName, receiveMode);
 
                     //Register the method to process receive message
                     //The RegisterMessageHandler is validate for all register exist on the queue, without need loop for items
@@ -114,7 +217,7 @@ await queueReceiver.DeadLetterAsync(message.SystemProperties.LockToken,
             {
                 Exception moreInfo = new Exception($"Error setting up queue consumption from service bus. See inner exception for details. Message={exception.Message}", exception);
                 //Use the class instead of interface because tracking exceptions directly is not supposed to be done outside AMAW (i.e. by the business code)
-                ((LightTelemetry)Workbench.Instance.Telemetry).TrackException(moreInfo);
+                (Workbench.Instance.Telemetry as LightTelemetry)?.TrackException(moreInfo);
             }
         }
 
@@ -132,14 +235,18 @@ private void ProcessSubscription()
                     string topicName = topic.Value.TopicName;
                     string subscriptName = topic.Value.Subscription;
                     ReceiveMode receiveMode = ReceiveMode.PeekLock;
+
                     if (topic.Value.DeleteAfterRead)
                     {
                         receiveMode = ReceiveMode.ReceiveAndDelete;
                     }
+
                     int takeQuantity = topic.Value.TakeQuantity;
 
                     //Register Trace on the telemetry 
-                    SubscriptionClient subscriptionClient = new SubscriptionClient(GetConnection(topic), topicName, subscriptName, receiveMode, null);
+                    var subscriptionClient = _subscriptionClientFactory.CreateClient(
+                        GetConnection(topic), topicName, subscriptName, receiveMode
+                    );
 
                     //Register the method to process receive message
                     //The RegisterMessageHandler is validate for all register exist on the queue, without need loop for items
@@ -200,13 +307,57 @@ await subscriptionClient.DeadLetterAsync(message.SystemProperties.LockToken,
             {
                 Exception moreInfo = new Exception($"Error setting up subscription consumption from service bus. See inner exception for details. Message={exception.Message}", exception);
                 //Use the class instead of interface because tracking exceptions directly is not supposed to be done outside AMAW (i.e. by the business code)
-                ((LightTelemetry)Workbench.Instance.Telemetry).TrackException(moreInfo);
+                (Workbench.Instance.Telemetry as LightTelemetry)?.TrackException(moreInfo);
             }
         }
 
         protected override Task ProcessAsync()
         {
             throw new NotImplementedException();
         }
+
+        /// <summary>
+        /// Default implementation for <see cref="IQueueClientFactory"/>,
+        /// creates instances of <see cref="IQueueClient"/>.
+        /// </summary>
+        private class DefaultQueueClientFactory : IQueueClientFactory
+        {
+            /// <inheritdoc/>
+            public IQueueClient CreateClient(string connectionString, string queueName, ReceiveMode receiveMode)
+            {
+                return new QueueClient(connectionString, queueName, receiveMode);
+            }
+        }
+
+        /// <summary>
+        /// Default implementation for <see cref="ISubscriptionClientFactory"/>,
+        /// creates instances of <see cref="SubscriptionClient"/>.
+        /// </summary>
+        private class DefaultSubscriptionClientFactory : ISubscriptionClientFactory
+        {
+            /// <inheritdoc/>
+            public ISubscriptionClient CreateClient(string connectionString, string topicName, string subscriptionName, ReceiveMode mode)
+            {
+                return new SubscriptionClient(connectionString, topicName, subscriptionName, mode, null);
+            }
+        }
+
+        /// <summary>
+        /// Retrieves configuration using <see cref="LightConfigurator"/>.
+        /// </summary>
+        private class DefaultServiceBusConfigurationProvider : IServiceBusConfigurationProvider
+        {
+            /// <inheritdoc/>
+            public ServiceBusConfiguration GetConfiguration()
+            {
+                return LightConfigurator.Config<ServiceBusConfiguration>($"{nameof(ServiceBus)}");
+            }
+
+            /// <inheritdoc/>
+            public ServiceBusConfiguration GetConfiguration(string connectionKey)
+            {
+                return LightConfigurator.Config<ServiceBusConfiguration>($"{nameof(ServiceBus)}_{connectionKey}");
+            }
+        }
     }
 }

test/Liquid.OnAzure.Tests/ServiceBusTests.cs

@@ -0,0 +1,113 @@
+// Copyright (c) Avanade Inc. All rights reserved.
+// Licensed under the MIT License. See LICENSE in the project root for license information.
+
+using System;
+using System.Diagnostics.CodeAnalysis;
+using System.Threading;
+using System.Threading.Tasks;
+using AutoFixture;
+using Liquid.Activation;
+using Liquid.Interfaces;
+using Liquid.Tests;
+using Microsoft.Azure.ServiceBus;
+using NSubstitute;
+using Xunit;
+
+namespace Liquid.OnAzure.Tests
+{
+    public class ServiceBusTests
+    {
+        private const string QueueConnectionString = "QueueConnection";
+        private const string TopicConnectionString = "TopicConnection";
+
+        private readonly Fixture _fixture = new Fixture();
+
+        private readonly IQueueClientFactory _queueClientFactory;
+        private readonly IQueueClient _queueClient;
+        private readonly ISubscriptionClientFactory _subscriptionClientFactory;
+        private readonly ISubscriptionClient _subscriptionClient;
+        private readonly IServiceBusConfigurationProvider _configurationProvider;
+        private readonly ServiceBus _sut;
+
+        public ServiceBusTests()
+        {
+            Workbench.Instance.Reset();
+            Workbench.Instance.AddToCache(WorkbenchServiceType.Telemetry, Substitute.For<ILightTelemetry>());
+
+            // ARRANGE IQueueClientFactory
+            _queueClientFactory = Substitute.For<IQueueClientFactory>();
+            _queueClient = Substitute.For<IQueueClient>();
+
+            _queueClientFactory
+                .CreateClient(default, default, default)
+                .ReturnsForAnyArgs(_queueClient);
+
+            // ARRANGE ISubscriptionClientFactory
+            _subscriptionClientFactory = Substitute.For<ISubscriptionClientFactory>();
+            _subscriptionClient = Substitute.For<ISubscriptionClient>();
+
+            _subscriptionClientFactory
+                .CreateClient(default, default, default, default)
+                .ReturnsForAnyArgs(_subscriptionClient);
+
+            // ARRANGE IServiceBusConfigurationProvider
+            _configurationProvider = Substitute.For<IServiceBusConfigurationProvider>();
+
+            // ARRANGE ServiceBus
+            _sut = new ServiceBus(_queueClientFactory, _subscriptionClientFactory, _configurationProvider);
+        }
+
+        [Fact]
+        public void InitializeWhenExistsLightWorkerForQueueThenListenerIsCreated()
+        {
+            // ARRANGE
+            _configurationProvider
+                .GetConfiguration(QueueConnectionString)
+                .Returns(_fixture.Create<ServiceBusConfiguration>());
+
+            _configurationProvider
+                .GetConfiguration(TopicConnectionString)
+                .Returns(_fixture.Create<ServiceBusConfiguration>());
+
+            // ACT
+            _sut.Initialize();
+
+            // ASSERT
+            _queueClient
+                .Received(1)
+                .RegisterMessageHandler(Arg.Any<Func<Message, CancellationToken, Task>>(), Arg.Any<MessageHandlerOptions>());
+
+            _subscriptionClient
+                .Received(1)
+                .RegisterMessageHandler(Arg.Any<Func<Message, CancellationToken, Task>>(), Arg.Any<MessageHandlerOptions>());
+        }
+
+        [MessageBus(QueueConnectionString)]
+        [SuppressMessage(
+            "Design",
+            "CA1034:Nested types should not be visible",
+            Justification = "Used by tests and must be public")]
+        public class QueueWorker : LightWorker
+        {
+            [Queue("QueueName")]
+            public static string MyQueueMethod(string message)
+            {
+                return message + "asd";
+            }
+        }
+
+        [MessageBus(TopicConnectionString)]
+        [SuppressMessage(
+            "Design",
+            "CA1034:Nested types should not be visible",
+            Justification = "Used by tests and must be public")]
+        public class TopicWorker : LightWorker
+        {
+            [Topic("TopicName", "TopicSubscriber")]
+            public static string MyTopicMethod(string message)
+            {
+                return message + "asd";
+            }
+        }
+    }
+}