apply sorting ...

Configuration/XEventServiceConfiguration.cs

@@ -1,5 +1,5 @@
-using xEventService.Constants;
 using xEventService.Models;
+using xEventService.Constants;
 
 namespace xEventService.Configuration
 {

DI/XDIHelperExtension.cs

@@ -1,11 +1,11 @@
 using System;
-using Microsoft.Extensions.Configuration;
-using Microsoft.Extensions.DependencyInjection;
 using xCommons.Extensions;
-using xEventService.Configuration;
+using xExceptions.Constants;
 using xEventService.Constants;
 using xEventService.Extensions;
-using xExceptions.Constants;
+using xEventService.Configuration;
+using Microsoft.Extensions.Configuration;
+using Microsoft.Extensions.DependencyInjection;
 
 namespace xEventService.DI
 {
@@ -16,7 +16,7 @@ namespace xEventService.DI
         /// </summary>
         /// <param name="source"></param>
         /// <returns></returns>
-        public static XEventServiceConfiguration GetXEventServiceConfigurations(
+        public static XEventServiceConfiguration GetXEventServiceConfiguration(
             this IConfiguration source
         )
         {
@@ -34,14 +34,14 @@ namespace xEventService.DI
         /// </summary>
         /// <param name="services"></param>
         /// <param name="configuration"></param>
-        public static void AddXEventServiceConfigurations(
+        public static void AddXEventServiceConfiguration(
             this IServiceCollection services,
             IConfiguration configuration
         )
         {
             //
-            var config = configuration.GetXEventServiceConfigurations();
-            services.AddXEventServiceConfigurations(config);
+            var config = configuration.GetXEventServiceConfiguration();
+            services.AddXEventServiceConfiguration(config);
         }
 
         /// <summary>
@@ -49,14 +49,20 @@ namespace xEventService.DI
         /// </summary>
         /// <param name="services"></param>
         /// <param name="configuration"></param>
-        public static void AddXEventServiceConfigurations(
+        public static void AddXEventServiceConfiguration(
             this IServiceCollection services,
             XEventServiceConfiguration configuration
         )
         {
             //
             // Validate ...
-            if (!configuration.IsValid())
+            var isValid = configuration.IsValid() &&
+                (configuration.Broker == XEventBroker.XKafka ||
+                (configuration.Broker == XEventBroker.XZeroMQ &&
+                configuration.ZeroMQ.IsValid()) ||
+                (configuration.Broker == XEventBroker.XRabbitMQ &&
+                configuration.RabbitMQ.IsValid()));
+            if (!isValid)
             {
                 //
                 Log("Registration Failed, due Invalid Configuration ...");

Extensions/XEventServiceExtensions.cs

@@ -1,9 +1,9 @@
 using System.Linq;
 using xCommons.Extensions;
-using xEventService.Configuration;
-using xEventService.Constants;
 using xEventService.Models;
 using xExceptions.Constants;
+using xEventService.Constants;
+using xEventService.Configuration;
 
 namespace xEventService.Extensions
 {

Interfaces/IXEventServiceProvider.cs

@@ -0,0 +1,5 @@
+namespace xEventService.Interfaces
+{
+    public interface IXEventServiceProvider : IXProducerServiceBase
+    { }
+}

Interfaces/IXProducerServiceBase.cs

@@ -1,8 +1,8 @@
 using System;
 using System.Threading;
+using xEventService.Models;
 using System.Threading.Tasks;
 using xEventService.Constants;
-using xEventService.Models;
 
 namespace xEventService.Interfaces
 {

Interfaces/IXRabbitMQProducerService.cs

@@ -1,6 +1,6 @@
 using System.Threading;
-using System.Threading.Tasks;
 using xEventService.Models;
+using System.Threading.Tasks;
 
 namespace xEventService.Interfaces
 {

Models/XEventMessage.cs

@@ -1,6 +1,6 @@
 using System;
-using System.Collections.Generic;
 using Newtonsoft.Json;
+using System.Collections.Generic;
 
 namespace xEventService.Models
 {

Providers/XBaseEventConsumerBackgroundService.cs

@@ -1,11 +1,11 @@
 using System;
-using Microsoft.Extensions.Logging;
-using xCommons.Extensions;
 using xCommons.Providers;
-using xEventService.Configuration;
+using xCommons.Extensions;
+using xExceptions.Constants;
 using xEventService.Constants;
 using xEventService.Extensions;
-using xExceptions.Constants;
+using xEventService.Configuration;
+using Microsoft.Extensions.Logging;
 
 namespace xEventService.Providers
 {

Providers/XBaseEventProducerService.cs

@@ -1,8 +1,8 @@
-using Microsoft.Extensions.Logging;
 using xCommons.Extensions;
-using xEventService.Configuration;
-using xEventService.Extensions;
 using xExceptions.Constants;
+using xEventService.Extensions;
+using xEventService.Configuration;
+using Microsoft.Extensions.Logging;
 
 namespace xEventService.Providers
 {

Providers/XEventServiceProvider.cs

@@ -0,0 +1,159 @@
+using System.Threading;
+using xCommons.Extensions;
+using xEventService.Models;
+using xExceptions.Constants;
+using System.Threading.Tasks;
+using xEventService.Constants;
+using xEventService.Extensions;
+using xEventService.Interfaces;
+using xEventService.Configuration;
+
+namespace xEventService.Providers
+{
+    public class XEventServiceProvider : IXEventServiceProvider
+    {
+        //
+        private readonly XEventServiceConfiguration configuration;
+        private readonly IXKafkaProducerService kafkaProducerService;
+        private readonly IXZeroMQProducerService zeroMQProducerService;
+        private readonly IXRabbitMQProducerService rabbitMQProducerService;
+
+        public XEventServiceProvider(
+            XEventServiceConfiguration configuration,
+            IXKafkaProducerService kafkaProducerService = null,
+            IXZeroMQProducerService zeroMQProducerService = null,
+            IXRabbitMQProducerService rabbitMQProducerService = null
+        )
+        {
+            //
+            this.configuration = configuration;
+            this.kafkaProducerService = kafkaProducerService;
+            this.zeroMQProducerService = zeroMQProducerService;
+            this.rabbitMQProducerService = rabbitMQProducerService;
+
+            //
+            // Validate ...
+            var isValid = configuration.IsValid() &&
+                configuration.Broker == XEventBroker.XKafka
+                ? kafkaProducerService != null
+                : configuration.Broker == XEventBroker.XZeroMQ
+                ? zeroMQProducerService != null
+                : configuration.Broker == XEventBroker.XRabbitMQ
+                ? rabbitMQProducerService != null
+                : false;
+            if (!isValid)
+            {
+                XException.InvalidConfiguration.Throw();
+            }
+        }
+
+        /// <summary>
+        /// Send Message to Broker Using XEventRequest instance ...
+        /// </summary>
+        /// <param name="request"></param>
+        /// <param name="cancellationToken"></param>
+        /// <returns></returns>
+        public async Task<bool> SendMessageAsync(
+            XEventRequest request,
+            CancellationToken cancellationToken = default
+        )
+        {
+            //
+            // Validate ...
+            if (!request.IsValid())
+            {
+                XException.InvalidArgs.Throw();
+            }
+
+            //
+            var result = await SendMessageAsync(
+                topic: request.Topic,
+                message: request.Message,
+                cancellationToken: cancellationToken
+            );
+
+            //
+            return result;
+        }
+
+        /// <summary>
+        /// Send Message to Broker Using XMessage instance ...
+        /// </summary>
+        /// <param name="message"></param>
+        /// <param name="topic"></param>
+        /// <param name="cancellationToken"></param>
+        /// <returns></returns>
+        public async Task<bool> SendMessageAsync(
+            XEventMessage message,
+            string topic = XEventServiceConstants.XDefaultTopic,
+            CancellationToken cancellationToken = default
+        )
+        {
+            //
+            // Validate Args ...
+            if (!message.IsValid())
+            {
+                XException.InvalidArgs.Throw();
+            }
+
+            //
+            var result = false;
+            switch (configuration.Broker)
+            {
+                //
+                case XEventBroker.XKafka:
+                    result = await kafkaProducerService.SendMessageAsync(
+                        topic: topic,
+                        message: message,
+                        cancellationToken: cancellationToken
+                    );
+                    break;
+
+                //
+                case XEventBroker.XZeroMQ:
+                    result = await zeroMQProducerService.SendMessageAsync(
+                        topic: topic,
+                        message: message,
+                        cancellationToken: cancellationToken
+                    );
+                    break;
+
+                //
+                case XEventBroker.XRabbitMQ:
+                    result = await rabbitMQProducerService.SendMessageAsync(
+                        topic: topic,
+                        message: message,
+                        cancellationToken: cancellationToken
+                    );
+                    break;
+            }
+
+            //
+            return result;
+        }
+
+        /// <summary>
+        /// Dispose ...
+        /// </summary>
+        public void Dispose()
+        {
+            //
+            if (kafkaProducerService != null)
+            {
+                kafkaProducerService.Dispose();
+            }
+
+            //
+            if (zeroMQProducerService != null)
+            {
+                zeroMQProducerService.Dispose();
+            }
+
+            //
+            if (rabbitMQProducerService != null)
+            {
+                rabbitMQProducerService.Dispose();
+            }
+        }
+    }
+}

Providers/XKafkaConsumerServiceBase.cs

@@ -1,14 +1,14 @@
 using System;
-using System.Threading;
-using System.Threading.Tasks;
 using Confluent.Kafka;
-using Microsoft.Extensions.Logging;
+using System.Threading;
 using xCommons.Extensions;
-using xEventService.Configuration;
-using xEventService.Constants;
-using xEventService.Extensions;
 using xEventService.Models;
 using xExceptions.Constants;
+using System.Threading.Tasks;
+using xEventService.Constants;
+using xEventService.Extensions;
+using xEventService.Configuration;
+using Microsoft.Extensions.Logging;
 
 namespace xEventService.Providers
 {
@@ -74,6 +74,7 @@ namespace xEventService.Providers
             //
             // Subscribe to Topic ...
             consumer.Subscribe(topic);
+            logger.LogInformation($"Kafka Cosume Topic: {topic} ...");
 
             //
             // Consuming ...
@@ -100,18 +101,22 @@ namespace xEventService.Providers
                         {
                             //
                             await DoWorkAsync(
-                                payload: consumeResult.Message,
+                                payload: message,
                                 cancellationToken: cancellationToken
                             );
                         }
                     }
                 }
-                catch (ConsumeException)
+                catch (ConsumeException ex)
                 {
+                    //
+                    logger.LogError($"Kafka Consumer Result Parsing Error: {ex.Message} ...");
                     await Task.Delay(1000, cancellationToken);
                 }
-                catch (Exception)
+                catch (Exception ex)
                 {
+                    //
+                    logger.LogError($"Kafka Consumer Result Parsing Error: {ex.Message} ...");
                     await Task.Delay(1000, cancellationToken);
                 }
             }

Providers/XKafkaProducerService.cs

@@ -1,15 +1,15 @@
+using System;
+using Confluent.Kafka;
+using System.Threading;
 using xCommons.Extensions;
+using xEventService.Models;
 using xExceptions.Constants;
+using System.Threading.Tasks;
+using xEventService.Constants;
 using xEventService.Extensions;
 using xEventService.Interfaces;
 using xEventService.Configuration;
-using xEventService.Models;
-using Confluent.Kafka;
 using Microsoft.Extensions.Logging;
-using System.Threading.Tasks;
-using System.Threading;
-using xEventService.Constants;
-using System;
 
 namespace xEventService.Providers
 {
@@ -63,8 +63,8 @@ namespace xEventService.Providers
 
             //
             result = await SendMessageAsync(
-                message: request.Message,
                 topic: request.Topic,
+                message: request.Message,
                 cancellationToken: cancellationToken
             );
 
@@ -111,8 +111,10 @@ namespace xEventService.Providers
                 //
                 result = !response.IsNullOrDefault();
             }
-            catch
+            catch (Exception ex)
             {
+                //
+                logger.LogError($"Kafka Produce Error: {ex.Message} ...");
                 result = false;
             }
 

Providers/XRabbitMQConsumerServiceBase.cs

@@ -1,16 +1,16 @@
 using System;
 using System.Text;
-using System.Threading;
-using System.Threading.Tasks;
-using Microsoft.Extensions.Logging;
 using RabbitMQ.Client;
-using RabbitMQ.Client.Events;
+using System.Threading;
 using xCommons.Extensions;
-using xEventService.Configuration;
-using xEventService.Constants;
-using xEventService.Extensions;
 using xEventService.Models;
 using xExceptions.Constants;
+using System.Threading.Tasks;
+using RabbitMQ.Client.Events;
+using xEventService.Constants;
+using xEventService.Extensions;
+using xEventService.Configuration;
+using Microsoft.Extensions.Logging;
 
 namespace xEventService.Providers
 {
@@ -51,7 +51,7 @@ namespace xEventService.Providers
 
             //
             SetupConnection();
-            DeclareTopology();
+            DeclareTopology(topic);
             consumer = new AsyncEventingBasicConsumer(channel);
         }
 
@@ -68,6 +68,7 @@ namespace xEventService.Providers
             // For Fix Blocking Synchronous ...
             await Task.Yield();
             using var cts = new CancellationTokenSource();
+            logger.LogInformation($"RabbitMQ Cosume Topic: {topic} ...");
 
             //
             // Subscribe to Topic ...
@@ -103,9 +104,10 @@ namespace xEventService.Providers
                         );
                     }
                 }
-                catch
+                catch (Exception ex)
                 {
                     //
+                    logger.LogError($"RabbitMQ Cosume Error: {ex.Message} ...");
                     if (configuration.RabbitMQ.ManualAck)
                     {
                         //
@@ -153,10 +155,20 @@ namespace xEventService.Providers
         public override async Task StopAsync(CancellationToken cancellationToken)
         {
             //
-            channel.Close();
-            channel.Dispose();
-            connection.Close();
-            connection.Dispose();
+            if (channel != null)
+            {
+                //
+                channel.Close();
+                channel.Dispose();
+            }
+
+            //
+            if (connection != null)
+            {
+                //
+                connection.Close();
+                connection.Dispose();
+            }
 
             //
             await base.StopAsync(cancellationToken);
@@ -199,8 +211,9 @@ namespace xEventService.Providers
             if (result.IsNullOrEmpty())
             {
                 //
+                var exchangeName = topic ?? configuration.RabbitMQ.Exchange;
                 var digits = Guid.NewGuid().ToString().GetDigits();
-                result = $"{configuration.RabbitMQ.Exchange}[{digits}]";
+                result = $"{exchangeName}[{digits}]";
             }
 
             //
@@ -228,12 +241,12 @@ namespace xEventService.Providers
             channel = connection.CreateModel();
         }
 
-        private void DeclareTopology()
+        private void DeclareTopology(string topic)
         {
             //
             channel.ExchangeDeclare(
                 arguments: null,
-                exchange: configuration.RabbitMQ.Exchange,
+                exchange: topic ?? configuration.RabbitMQ.Exchange,
                 durable: configuration.RabbitMQ.ExchangeDurable,
                 autoDelete: configuration.RabbitMQ.ExchangeAutoDelete,
                 type: configuration.RabbitMQ.ExchangeType.GetStringValue()
@@ -255,7 +268,7 @@ namespace xEventService.Providers
             channel.QueueBind(
                 arguments: null,
                 queue: queueName,
-                exchange: configuration.RabbitMQ.Exchange,
+                exchange: topic ?? configuration.RabbitMQ.Exchange,
                 routingKey: configuration.RabbitMQ.RoutingKey
             );
         }

Providers/XRabbitMQProducerService.cs

@@ -1,16 +1,16 @@
 using System;
 using System.Text;
-using System.Threading;
-using System.Threading.Tasks;
-using Microsoft.Extensions.Logging;
 using RabbitMQ.Client;
+using System.Threading;
 using xCommons.Extensions;
-using xEventService.Configuration;
+using xEventService.Models;
+using xExceptions.Constants;
+using System.Threading.Tasks;
 using xEventService.Constants;
 using xEventService.Extensions;
 using xEventService.Interfaces;
-using xEventService.Models;
-using xExceptions.Constants;
+using xEventService.Configuration;
+using Microsoft.Extensions.Logging;
 
 namespace xEventService.Providers
 {
@@ -143,7 +143,7 @@ namespace xEventService.Providers
             {
                 //
                 // Ensure Connection is Alive ...
-                EnsureChannelAsync();
+                EnsureChannel();
 
                 //
                 // Serialize Message ...
@@ -181,8 +181,10 @@ namespace xEventService.Providers
                 //
                 result = true;
             }
-            catch
+            catch (Exception ex)
             {
+                //
+                logger.LogError($"RabbitMQ Produce Error: {ex.Message} ...");
                 result = false;
             }
 
@@ -268,7 +270,7 @@ namespace xEventService.Providers
         /// <summary>
         /// Ensure Channel is Open and Exchange/Queue Declared ...
         /// </summary>
-        private void EnsureChannelAsync()
+        private void EnsureChannel()
         {
             //
             if (IsConnected())

Providers/XZeroMQConsumerServiceBase.cs

@@ -1,16 +1,16 @@
+using NetMQ;
 using System;
 using System.Text;
-using System.Threading;
-using System.Threading.Tasks;
-using Microsoft.Extensions.Logging;
-using NetMQ;
 using NetMQ.Sockets;
+using System.Threading;
 using xCommons.Extensions;
-using xEventService.Configuration;
-using xEventService.Constants;
-using xEventService.Extensions;
 using xEventService.Models;
 using xExceptions.Constants;
+using System.Threading.Tasks;
+using xEventService.Constants;
+using xEventService.Extensions;
+using xEventService.Configuration;
+using Microsoft.Extensions.Logging;
 
 namespace xEventService.Providers
 {
@@ -40,7 +40,8 @@ namespace xEventService.Providers
         {
             //
             // Validating Configuration ...
-            if (configuration.Broker != XEventBroker.XKafka)
+            if (!configuration.ZeroMQ.IsValid() ||
+                configuration.Broker != XEventBroker.XZeroMQ)
             {
                 XException.InvalidConfiguration.Throw();
             }
@@ -61,6 +62,7 @@ namespace xEventService.Providers
         {
             //
             await Task.Yield();
+            logger.LogInformation($"ZeroMQ Cosume Topic: {topic} ...");
             while (!cancellationToken.IsCancellationRequested)
             {
                 //
@@ -70,6 +72,7 @@ namespace xEventService.Providers
                     string receivedTopic = string.Empty;
                     byte[] body = null;
 
+                    //
                     if (configuration.ZeroMQ.Pattern == XZeroMQPattern.PubSub)
                     {
                         //
@@ -80,6 +83,10 @@ namespace xEventService.Providers
                             {
                                 // Process
                             }
+                            else
+                            {
+                                logger.LogError($"ZeroMQ Cosume Error ...");
+                            }
                         }
                     }
                     else
@@ -88,8 +95,13 @@ namespace xEventService.Providers
                         {
                             // Process
                         }
+                        else
+                        {
+                            logger.LogError($"ZeroMQ Cosume Error ...");
+                        }
                     }
 
+                    //
                     if (body != null && body.Length > 0)
                     {
                         //
@@ -113,8 +125,10 @@ namespace xEventService.Providers
                         await Task.Delay(10, cancellationToken);
                     }
                 }
-                catch (Exception)
+                catch (Exception ex)
                 {
+                    //
+                    logger.LogError($"ZeroMQ Cosume Error: {ex.Message} ...");
                     await Task.Delay(1000, cancellationToken);
                 }
             }
@@ -123,7 +137,12 @@ namespace xEventService.Providers
         public override async Task StopAsync(CancellationToken cancellationToken)
         {
             //
-            socket.Close();
+            if (socket != null)
+            {
+                socket.Close();
+            }
+
+            //
             await base.StopAsync(cancellationToken);
         }
 

Providers/XZeroMQProducerService.cs

@@ -1,17 +1,17 @@
+using NetMQ;
 using System;
 using System.Text;
-using System.Threading;
-using System.Threading.Tasks;
-using Microsoft.Extensions.Logging;
-using NetMQ;
 using NetMQ.Sockets;
+using System.Threading;
 using xCommons.Extensions;
-using xEventService.Configuration;
+using xEventService.Models;
+using xExceptions.Constants;
+using System.Threading.Tasks;
 using xEventService.Constants;
 using xEventService.Extensions;
 using xEventService.Interfaces;
-using xEventService.Models;
-using xExceptions.Constants;
+using xEventService.Configuration;
+using Microsoft.Extensions.Logging;
 
 namespace xEventService.Providers
 {

xEventService.csproj

@@ -20,7 +20,8 @@
 
   <!-- Icon Handling -->
   <ItemGroup>
-    <None Include="../../Documents/Resources/Images/favicon.png" Link="icon.png" Pack="true" PackagePath="\icon.png" />
+    <None Include="../../Documents/Resources/Images/favicon.png" Link="icon.png" Pack="true"
+      PackagePath="\icon.png" />
   </ItemGroup>
 
   <!-- Local Modules -->
@@ -41,9 +42,9 @@
 
   <!-- For XML Documentation Support -->
   <PropertyGroup>
-    <CopyLocalLockFileAssemblies>true</CopyLocalLockFileAssemblies>
-    <GenerateDocumentationFile>true</GenerateDocumentationFile>
     <NoWarn>$(NoWarn);1591</NoWarn>
+    <GenerateDocumentationFile>true</GenerateDocumentationFile>
+    <CopyLocalLockFileAssemblies>true</CopyLocalLockFileAssemblies>
   </PropertyGroup>
 
 </Project>