apply sorting ...

Base/XBaseEvent.cs

@@ -1,14 +0,0 @@
-using System;
-
-namespace xEventService.Base {
-    public abstract class XBaseEvent {
-        public string Type { get; protected set; }
-        public DateTime Timestamp { get; protected set; }
-
-        protected XBaseEvent () {
-            //
-            Type = GetType ().Name;
-            Timestamp = DateTime.UtcNow;
-        }
-    }
-}

Configuration/XEventServiceConfiguration.cs

@@ -1,51 +1,39 @@
-using System.Collections.Generic;
+using xEventService.Models;
-using RabbitMQ.Client;
+using xEventService.Constants;
 
-namespace xEventService.Configuration {
+namespace xEventService.Configuration
+{
     /// <summary>
     /// a configuration class for describing how to connect rabbit mq server
     /// and how channels must be declared ...
     /// </summary>
-    public class XEventServiceConfiguration {
+    public class XEventServiceConfiguration
-        public string Server { get; set; }
+    {
-        public int Port { get; set; }
+        /// <summary>
-        public string Username { get; set; }
+        /// Message Broker Address ...
-        public string Password { get; set; }
+        /// </summary>
+        public string Url { get; set; }
 
-        //
+        /// <summary>
-        public bool AllowDynamicQueues { get; set; } = true;
+        /// Max Retries to Sending Message ...
-        public XQueueArgs DefaultQueueArgs { get; set; } = new XQueueArgs ();
+        /// </summary>
-        public IDictionary<string, XQueueArgs> Queues { get; set; }
+        public int MaxRetries { get; set; } = 3;
 
-        //
+        /// <summary>
-        public bool AllowDynamicExchangess { get; set; } = true;
+        /// Allowed Message Broker ...
-        public XExchangeArgs DefaultExchangeArgs { get; set; } = new XExchangeArgs ();
+        /// </summary>
-        public IDictionary<string, XExchangeArgs> Exchanges { get; set; }
+        public XEventBroker Broker { get; set; } = XEventBroker.None;
         
-        //
+        /// <summary>
-        public static XEventServiceConfiguration GetDefaults () {
+        /// ZeroMQ Specific Options ...
-            return new XEventServiceConfiguration {
+        /// Only used when Broker is XZeroMQ ...
-                Port = 5672,
+        /// </summary>
-                    Username = "guest",
+        public XZeroMQOptions ZeroMQ { get; set; } = new XZeroMQOptions();        
-                    Password = "guest",
-                    Server = "localhost",
-                    Queues = new Dictionary<string, XQueueArgs> { { "XMainQueue", new XQueueArgs { Name = "XMainQueue" } }
-                    }
-            };
-        }
-    }
         
-    public class XQueueArgs {
+        /// <summary>
-        public string Name { get; set; }
+        /// RabbitMQ Specific Options ...
-        public bool Durable { get; set; } = false;
+        /// Only used when Broker is XRabbitMQ ...
-        public bool Exclusive { get; set; } = false;
+        /// </summary>
-        public bool AutoDelete { get; set; } = true;
+        public XRabbitMQOptions RabbitMQ { get; set; } = new XRabbitMQOptions();        
-    }
-
-    public class XExchangeArgs {
-        public string Name { get; set; }
-        public string Type { get; set; } = ExchangeType.Fanout;
-        public bool Durable { get; set; } = false;
-        public bool AutoDelete { get; set; } = true;
     }
 }

Constants/XEventBroker.cs

@@ -0,0 +1,22 @@
+using xExceptions.Attributes;
+
+namespace xEventService.Constants
+{
+    /// <summary>
+    /// Available Brokers ...
+    /// </summary>
+    public enum XEventBroker
+    {
+        [StringValue(XEventServiceConstants.None)]
+        None,
+
+        [StringValue(XEventServiceConstants.XKafka)]
+        XKafka,
+
+        [StringValue(XEventServiceConstants.XZeroMQ)]
+        XZeroMQ,
+
+        [StringValue(XEventServiceConstants.XRabbitMQ)]
+        XRabbitMQ,
+    }
+}

Constants/XEventServiceConstants.cs

@@ -0,0 +1,25 @@
+namespace xEventService.Constants
+{
+    public struct XEventServiceConstants
+    {
+        //
+        // Service Extensions Log Tag ...
+        public const string XEventServiceDILogTag = "XEventService";
+
+        //
+        // Defaults ...
+        public const string XDefaultTopic = "XSaherelm";
+        public const string XDefaultConsumerGroup = "XSaherElmGroup";
+
+        //
+        public const string None = "None";
+        public const string XKafka = "XKafka";
+        public const string PubSub = "PubSub";
+        public const string XZeroMQ = "XZeroMQ";
+        public const string PushPull = "PushPull";
+        public const string XRabbitMQ = "XRabbitMQ";
+
+        //
+        public const string XWildcard = "*";
+    }
+}

Constants/XRabbitMQExchangeType.cs

@@ -0,0 +1,37 @@
+using xExceptions.Attributes;
+
+namespace xEventService.Constants
+{
+    /// <summary>
+    /// Available RabbitMQ Exchange Types ...
+    /// </summary>
+    public struct XRabbitMQExchangeTypes
+    {
+        public const string None = "none";
+        public const string Direct = "direct";
+        public const string Fanout = "fanout";
+        public const string Topic = "topic";
+        public const string Headers = "headers";
+    }
+
+    /// <summary>
+    /// RabbitMQ Exchange Type Enum ...
+    /// </summary>
+    public enum XRabbitMQExchangeType
+    {
+        [StringValue(XRabbitMQExchangeTypes.None)]
+        None,
+
+        [StringValue(XRabbitMQExchangeTypes.Direct)]
+        Direct,
+
+        [StringValue(XRabbitMQExchangeTypes.Fanout)]
+        Fanout,
+
+        [StringValue(XRabbitMQExchangeTypes.Topic)]
+        Topic,
+
+        [StringValue(XRabbitMQExchangeTypes.Headers)]
+        Headers,
+    }
+}

Constants/XZeroMQPattern.cs

@@ -0,0 +1,16 @@
+using xExceptions.Attributes;
+
+namespace xEventService.Constants
+{
+    public enum XZeroMQPattern
+    {
+        [StringValue(XEventServiceConstants.None)]
+        None,
+
+        [StringValue(XEventServiceConstants.PubSub)]
+        PubSub,
+
+        [StringValue(XEventServiceConstants.PushPull)]
+        PushPull
+    }
+}

DI/XDIHelperExtension.cs

@@ -1,26 +1,29 @@
 using System;
-using Microsoft.AspNetCore.Builder;
+using xCommons.Extensions;
+using xExceptions.Constants;
+using xEventService.Constants;
+using xEventService.Extensions;
+using xEventService.Configuration;
 using Microsoft.Extensions.Configuration;
 using Microsoft.Extensions.DependencyInjection;
-using xCommons.Extensions;
-using xEventService.Configuration;
-using xEventService.Constants;
-using xEventService.Interfaces;
-using xEventService.Models;
-using xEventService.Providers;
 
-namespace xEventService.DI {
+namespace xEventService.DI
-    public static partial class XDIHelperExtension {
+{
+    public static partial class XDIHelperExtension
+    {
         /// <summary>
         /// Extract EventService Configurations ...
         /// </summary>
         /// <param name="source"></param>
         /// <returns></returns>
-        public static XEventServiceConfiguration GetXEventServiceConfigurations (this IConfiguration source) {
+        public static XEventServiceConfiguration GetXEventServiceConfiguration(
+            this IConfiguration source
+        )
+        {
             //
             var configSection = source
-                .GetSection (ConfigurationNodeNames.EVENT_SERVICE_NODE_NAME);
+                .GetSection(ConfigurationNodeNames.EVENT_SERVICE_NODE_NAME);
-            var result = configSection.Get<XEventServiceConfiguration> ();
+            var result = configSection.Get<XEventServiceConfiguration>();
 
             //
             return result;
@@ -31,13 +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 ();
+            var config = configuration.GetXEventServiceConfiguration();
-            services.AddXEventServiceConfigurations (config);
+            services.AddXEventServiceConfiguration(config);
         }
 
         /// <summary>
@@ -45,96 +49,46 @@ 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
-        ) {
+        )
+        {
             //
-            services.AddSingleton<XEventServiceConfiguration> (configuration);
+            // Validate ...
-        }
+            var isValid = configuration.IsValid() &&
-
+                (configuration.Broker == XEventBroker.XKafka ||
-        /// <summary>
+                (configuration.Broker == XEventBroker.XZeroMQ &&
-        /// Register Event Service ...
+                configuration.ZeroMQ.IsValid()) ||
-        /// </summary>
+                (configuration.Broker == XEventBroker.XRabbitMQ &&
-        /// <param name="services"></param>
+                configuration.RabbitMQ.IsValid()));
-        /// <param name="configuration"></param>
+            if (!isValid)
-        public static void AddXEventService (
+            {
-            this IServiceCollection services,
-            IConfiguration configuration
-        ) {
-            //
-            // Check Configuration Registered ...
-            var config = services.GetRegisteredService<XEventServiceConfiguration> ();
-            if (config.IsNull ()) {
                 //
-                Console.WriteLine ($"XEventService: there is no provided XEventServiceConfiguration, using default ...");
+                Log("Registration Failed, due Invalid Configuration ...");
-
+                XException.InvalidConfiguration.Throw();
-                //
-                // Register Default Configuration ...
-                services.AddXEventServiceConfigurations (configuration);
             }
 
             //
-            // Register IXEventBus ...
+            // Register Configuration as Singleton ...
-            var serviceBas = services.GetRegisteredService<IXEventBus> ();
+            services.AddSingleton(configuration);
-            if (serviceBas.IsNull ()) {
-                services.AddSingleton<IXEventBus, XEventBus> ();
-            }
         }
 
+        //
+        #region Private ...
+        /// <summary>
+        /// a LogTag for Service ...
+        /// </summary>
+        private static string XLogTag = XEventServiceConstants.XEventServiceDILogTag;
+
         /// <summary>
-        /// Register an Event Handler ...
+        /// print a log in Console ...
         /// </summary>
-        /// <param name="services"></param>
+        /// <param name="message"></param>
-        /// <param name="lifetime"></param>
+        private static void Log(string message)
-        /// <typeparam name="TEvent"></typeparam>
+        {
-        /// <typeparam name="THandler"></typeparam>
+            Console.WriteLine($"{XLogTag} => {message}");
-        /// <returns></returns>
-        public static void AddXEventHandler<TEvent, THandler> (
-            this IServiceCollection services,
-            ServiceLifetime lifetime = ServiceLifetime.Transient
-        )
-        where TEvent : XEvent
-        where THandler : IXEventHandler<TEvent> {
-            //
-            services.Add (new ServiceDescriptor (
-                serviceType: typeof (THandler),
-                implementationType: typeof (THandler),
-                lifetime: lifetime
-            ));
-
-            //
-            services.Add (new ServiceDescriptor (
-                serviceType: typeof (IXEventHandler<TEvent>),
-                implementationType: typeof (THandler),
-                lifetime: lifetime
-            ));
-        }
-
-        /// <summary>
-        /// Subscribe to specific Event ...
-        /// </summary>
-        /// <param name="app"></param>
-        /// <typeparam name="TEvent"></typeparam>
-        /// <typeparam name="THandler"></typeparam>
-        /// <returns></returns>
-        public static void XEventSubscribe<TEvent, THandler> (
-            this IApplicationBuilder app,
-            bool toExchange = true
-        )
-        where TEvent : XEvent
-        where THandler : IXEventHandler<TEvent> {
-            //
-            var eventName = typeof (TEvent).Name;
-
-            //
-            // Retrieve EventBus from IOC ...
-            var eventBus = app.ApplicationServices.GetRequiredService<IXEventBus> ();
-
-            //
-            // Subscribe to Event Handler ...
-            var isSubscribed = eventBus.Subscribe<TEvent, THandler> ();
-            Console.WriteLine ($"XEventService: Event Handler subscription result: {eventName} => {isSubscribed} ...");
         }
+        #endregion
     }
 }

Extensions/ConfigurationExtensions.cs

@@ -1,288 +0,0 @@
-using System;
-using RabbitMQ.Client;
-using xCommons.Extensions;
-using xEventService.Configuration;
-using xEventService.Models;
-using xExceptions.Constants;
-
-namespace xEventService.Extensions {
-    public static class ConfigurationExtensions {
-        /// <summary>
-        /// Create Connection Factory Based on
-        /// provided Configuration ...
-        /// </summary>
-        /// <param name="source"></param>
-        /// <returns></returns>
-        public static ConnectionFactory GetConnectionFactory (this XEventServiceConfiguration source) {
-            //
-            // Validate Args ...
-            if (source.IsNull ()) {
-                XException.InvalidConfiguration.Throw ();
-            }
-
-            //
-            // Create Connection Factory ...
-            var result = new ConnectionFactory {
-                Port = source.Port,
-                HostName = source.Server,
-                UserName = source.Username,
-                Password = source.Password
-            };
-
-            //
-            // Return Result ...
-            return result;
-        }
-
-        /// <summary>
-        /// Create a Connection to Server ...
-        /// </summary>
-        /// <param name="source"></param>
-        /// <returns></returns>
-        public static IConnection Connect (this XEventServiceConfiguration source) {
-            //
-            // Open Connection ...
-            IConnection result = null;
-            try {
-                //
-                // Retrieve Connection Factory ...
-                var connectionFactory = source.GetConnectionFactory ();
-
-                //
-                // Create Connection ...
-                result = connectionFactory.CreateConnection ();
-            } catch (Exception ex) {
-                //
-                Console.WriteLine ($"XEventService Exception: Connection Failed, {ex.Message} ...");
-
-                //
-                XException.ActionFailed.Throw ();
-            }
-
-            //
-            return result;
-        }
-
-        /// <summary>
-        /// Validate a Queue Name based on Configurations ...
-        /// </summary>
-        /// <param name="source"></param>
-        /// <param name="queueName"></param>
-        public static void ValidateQueue (
-            this XEventServiceConfiguration source,
-            string queueName
-        ) {
-            //
-            // Validate Args ...
-            if (source.IsNull ()) {
-                XException.InvalidArgs.Throw ();
-            }
-
-            //
-            // Validate Queues ...
-            if ((source.Queues.IsNull () &&
-                    !source.AllowDynamicQueues) ||
-                (!source.Queues.IsNull () &&
-                    !source.Queues.Keys.HasChild () &&
-                    !source.AllowDynamicQueues)) {
-                //
-                Console.WriteLine ($"XEventService Exception: there isn't any configured Queue ...");
-
-                //
-                XException.InvalidArgs.Throw ();
-            }
-
-            //
-            // Check queueName not empty ...
-            if (queueName.IsNullOrEmpty ()) {
-                //
-                Console.WriteLine ($"XEventService Exception: Invalid QueueName ...");
-
-                //
-                XException.InvalidArgs.Throw ();
-            }
-
-            //
-            // Check queueName Exists ...
-            if (!source.Queues.IsNull () &&
-                !source.Queues.ContainsKey (queueName) &&
-                !source.AllowDynamicQueues) {
-                //
-                Console.WriteLine ($"XEventService Exception: QueueName {queueName} not found ...");
-
-                //
-                XException.InvalidArgs.Throw ();
-            }
-        }
-
-        /// <summary>
-        /// Validate a Exchange Name based on Configurations ...
-        /// </summary>
-        /// <param name="source"></param>
-        /// <param name="exchangeName"></param>
-        public static void ValidateExchange (
-            this XEventServiceConfiguration source,
-            string exchangeName
-        ) {
-            //
-            // Validate Args ...
-            if (source.IsNull ()) {
-                XException.InvalidArgs.Throw ();
-            }
-
-            //
-            // Validate Exchanges ...
-            if ((source.Exchanges.IsNull () &&
-                    !source.AllowDynamicExchangess) ||
-                (!source.Exchanges.IsNull () &&
-                    !source.Exchanges.Keys.HasChild () &&
-                    !source.AllowDynamicExchangess)) {
-                //
-                Console.WriteLine ($"XEventService Exception: there isn't any configured Exchanges ...");
-
-                //
-                XException.InvalidArgs.Throw ();
-            }
-
-            //
-            // Check exchangeName not empty ...
-            if (exchangeName.IsNullOrEmpty ()) {
-                //
-                Console.WriteLine ($"XEventService Exception: Invalid ExchangeName ...");
-
-                //
-                XException.InvalidArgs.Throw ();
-            }
-
-            //
-            // Check exchangeName Exists ...
-            if (!source.Exchanges.IsNull () &&
-                !source.Exchanges.ContainsKey (exchangeName) &&
-                !source.AllowDynamicExchangess) {
-                //
-                Console.WriteLine ($"XEventService Exception: ExchangeName {exchangeName} not found ...");
-
-                //
-                XException.InvalidArgs.Throw ();
-            }
-        }
-
-        /// <summary>
-        /// Retrieve specific QueueArgs ...
-        /// </summary>
-        /// <param name="source"></param>
-        /// <param name="queueName"></param>
-        /// <returns></returns>
-        public static XQueueArgs GetQueueArgs (
-            this XEventServiceConfiguration source,
-            string queueName
-        ) {
-            //
-            // Validate Queue Name ...
-            source.ValidateQueue (queueName);
-
-            //
-            var result = source.Queues
-                .IsNull () ? null : source
-                .Queues[queueName];
-            if (result.IsNull ()) {
-                //
-                // Prepare Default Queue ...
-                result = source.DefaultQueueArgs;
-                if (result.IsNull ()) {
-                    result = new XQueueArgs () {
-                        Name = queueName
-                    };
-                } else {
-                    result.Name = queueName;
-                }
-            }
-
-            //
-            return result;
-        }
-
-        /// <summary>
-        /// Retrieve specific ExchangeArgs ...
-        /// </summary>
-        /// <param name="source"></param>
-        /// <param name="exchangeName"></param>
-        /// <returns></returns>
-        public static XExchangeArgs GetExchangeArgs (
-            this XEventServiceConfiguration source,
-            string exchangeName
-        ) {
-            //
-            // Validate Exchange Name ...
-            source.ValidateExchange (exchangeName);
-
-            //
-            var result = source.Exchanges
-                .IsNull () ? null : source
-                .Exchanges[exchangeName];
-            if (result.IsNull ()) {
-                //
-                // Prepare Default Queue ...
-                result = source.DefaultExchangeArgs;
-                if (result.IsNull ()) {
-                    result = new XExchangeArgs () {
-                        Name = exchangeName,
-                        Type = ExchangeType.Fanout
-                    };
-                } else {
-                    result.Name = exchangeName;
-                }
-            }
-
-            //
-            return result;
-        }
-
-        /// <summary>
-        /// Create a Channel ...
-        /// </summary>
-        /// <param name="source"></param>
-        /// <param name="exchangeName"></param>
-        /// <returns></returns>
-        public static XChannel CreateChannel (
-            this XEventServiceConfiguration source,
-            string exchangeName = null
-        ) {
-            //
-            // Validate ExchangeName and Retrieve ExchangeArgs if provided ...
-            var exchangeArgs = exchangeName
-                .IsNullOrEmpty () ?
-                null :
-                source.GetExchangeArgs (exchangeName);
-
-            //
-            // Create Connection ...
-            var connection = source.Connect ();
-            var channel = connection.CreateModel ();
-
-            //
-            // Create XChannel Model ...
-            var result = new XChannel (
-                exchangeName: exchangeName,
-                connection: connection
-            );
-
-            //
-            // Declaring Queue if Provided ...
-            if (!exchangeArgs.IsNull () &&
-                !exchangeName.IsNullOrEmpty ()) {
-                //
-                // Declaring Queue ...
-                result.Channel.ExchangeDeclare (
-                    exchange: exchangeName,
-                    type: exchangeArgs.Type,
-                    durable: exchangeArgs.Durable,
-                    autoDelete: exchangeArgs.AutoDelete
-                );
-            }
-
-            //
-            return result;
-        }
-    }
-}

Extensions/XEventModelExtensions.cs

@@ -1,44 +0,0 @@
-using System;
-using System.Text;
-using xCommons.Extensions;
-
-namespace xEventService.Extensions {
-    public static class XEventModelExtensions {
-        /// <summary>
-        /// Convert an Object to bytes array for publishing
-        /// </summary>
-        /// <param name="source"></param>
-        /// <typeparam name="T"></typeparam>
-        /// <returns></returns>
-        public static byte[] ToBody<T> (this T source) {
-            //
-            var json = source
-                .ToJSON ();
-
-            //
-            var result = json
-                .ToBytes ();
-
-            //
-            return result;
-        }
-
-        /// <summary>
-        /// Convert Recieved Bytes to Specific Type ...
-        /// </summary>
-        /// <param name="source"></param>
-        /// <typeparam name="T"></typeparam>
-        /// <returns></returns>
-        public static T FromBody<T> (this ReadOnlyMemory<byte> source) {
-            //
-            var jsonString = Encoding.UTF8
-                .GetString (
-                    source.ToArray ()
-                );
-            var result = jsonString.FromJSON<T> ();
-
-            //
-            return result;
-        }
-    }
-}

Extensions/XEventServiceExtensions.cs

@@ -0,0 +1,344 @@
+using System.Linq;
+using xCommons.Extensions;
+using xEventService.Models;
+using xExceptions.Constants;
+using xEventService.Constants;
+using xEventService.Configuration;
+
+namespace xEventService.Extensions
+{
+    /// <summary>
+    /// Extensions for Event Service ...
+    /// </summary>
+    public static class XEventServiceExtensions
+    {
+        /// <summary>
+        /// Validate Event Broker ...
+        /// </summary>
+        /// <param name="source"></param>
+        /// <returns></returns>
+        public static bool IsValid(this XEventBroker source)
+        {
+            //
+            var result =
+                !source.IsNull() &&
+                source != XEventBroker.None;
+
+            //
+            return result;
+        }
+
+        /// <summary>
+        /// Validate Event Service Configuration ...
+        /// </summary>
+        /// <param name="source"></param>
+        /// <returns></returns>
+        public static bool IsValid(this XEventServiceConfiguration source)
+        {
+            //
+            var result =
+                !source.IsNullOrDefault() &&
+                source.Broker.IsValid() &&
+                !source.Url.IsNullOrEmpty() &&
+                source.Url.IsValidUrl();
+
+            //
+            return result;
+        }
+
+        /// <summary>
+        /// Validate RabbitMQ Exchange Type ...
+        /// </summary>
+        /// <param name="source"></param>
+        /// <returns></returns>
+        public static bool IsValid(this XRabbitMQExchangeType source)
+        {
+            //
+            var result = source != XRabbitMQExchangeType.None;
+
+            //
+            return result;
+        }
+
+        /// <summary>
+        /// Validate RabbitMQ Configuration ...
+        /// </summary>
+        /// <param name="source"></param>
+        /// <returns></returns>
+        public static bool IsValid(this XRabbitMQOptions source)
+        {
+            //
+            var result =
+                !source.IsNullOrDefault() &&
+                !source.Exchange.IsNullOrEmpty() &&
+                source.ExchangeType.IsValid();
+
+            //
+            return result;
+        }
+
+        /// <summary>
+        /// Validate an Event Message ...
+        /// </summary>
+        /// <param name="source"></param>
+        /// <returns></returns>
+        public static bool IsValid(this XEventMessage source)
+        {
+            //
+            var result =
+                !source.IsNullOrDefault() &&
+                !source.Action.IsNullOrEmpty() &&
+                !source.Sender.IsNullOrEmpty();
+
+            //
+            return result;
+        }
+
+        /// <summary>
+        /// Validate an Event Request ...
+        /// </summary>
+        /// <param name="source"></param>
+        /// <returns></returns>
+        public static bool IsValid(this XEventRequest source)
+        {
+            //
+            var result =
+                !source.IsNullOrDefault() &&
+                !source.Topic.IsNullOrEmpty() &&
+                source.Message.IsValid();
+
+            //
+            return result;
+        }
+
+        /// <summary>
+        /// Validate ZeroMQ Pattern ...
+        /// </summary>
+        /// <param name="source"></param>
+        /// <returns></returns>
+        public static bool IsValid(this XZeroMQPattern source)
+        {
+            //
+            var result = source != XZeroMQPattern.None;
+
+            //
+            return result;
+        }
+
+        /// <summary>
+        /// Validate ZeroMQ Options ...
+        /// </summary>
+        /// <param name="source"></param>
+        /// <returns></returns>
+        public static bool IsValid(this XZeroMQOptions source)
+        {
+            //
+            var result =
+                !source.IsNullOrDefault() &&
+                source.Pattern.IsValid();
+
+            //
+            return result;
+        }
+
+        /// <summary>
+        /// Validate an Array is Wildcard passed or not ...
+        /// </summary>
+        /// <param name="source"></param>
+        /// <returns></returns>
+        public static bool IsAllowedWildcard(this string[] source)
+        {
+            //
+            var result =
+                source is not null &&
+                source.Length == 1 &&
+                source.Any(s =>
+                    s == XEventServiceConstants.XWildcard);
+
+            //
+            return result;
+        }
+
+        /// <summary>
+        /// Validate a Collection is Valid for Kafka Consumer ...
+        /// </summary>
+        /// <param name="source"></param>
+        /// <returns></returns>
+        public static bool IsAllowedCollection(
+            this string[] source
+        )
+        {
+            //
+            var result = source.HasChild();
+
+            //
+            return result;
+        }
+
+        /// <summary>
+        /// Validate Sender ...
+        /// </summary>
+        /// <param name="source"></param>
+        /// <param name="allowedSenders"></param>
+        /// <returns></returns>
+        public static bool IsValidSender(
+            this XEventMessage source,
+            string[] allowedSenders
+        )
+        {
+            //
+            // Validate Source and Collection ...
+            var result =
+                source.IsValid() &&
+                allowedSenders
+                    .IsAllowedCollection();
+            if (!result)
+            {
+                return result;
+            }
+
+            //
+            // Validate Sender ...
+            result =
+                allowedSenders.IsAllowedWildcard() ||
+                allowedSenders.Any(x =>
+                    x.Equals(
+                        source.Sender,
+                        System.StringComparison.InvariantCultureIgnoreCase));
+
+            //
+            return result;
+        }
+
+        /// <summary>
+        /// Validate Action ...
+        /// </summary>
+        /// <param name="source"></param>
+        /// <param name="allowedActions"></param>
+        /// <returns></returns>
+        public static bool IsValidAction(
+            this XEventMessage source,
+            string[] allowedActions
+        )
+        {
+            //
+            // Validate Source and Collection ...
+            var result =
+                source.IsValid() &&
+                allowedActions
+                    .IsAllowedCollection();
+            if (!result)
+            {
+                return result;
+            }
+
+            //
+            // Validate Action ...
+            result =
+                allowedActions.IsAllowedWildcard() ||
+                allowedActions.Any(x =>
+                    x.Equals(
+                        source.Action,
+                        System.StringComparison.InvariantCultureIgnoreCase));
+
+            //
+            return result;
+        }
+
+        /// <summary>
+        /// Completely Validate a Message, based on providing:
+        /// - Allowed Actions
+        /// - Allowed Senders
+        /// </summary>
+        /// <param name="source"></param>
+        /// <param name="allowedSenders"></param>
+        /// <param name="allowedActions"></param>
+        /// <returns></returns>
+        public static bool IsValid(
+            this XEventMessage source,
+            string[] allowedSenders,
+            string[] allowedActions
+        )
+        {
+            //
+            var result =
+                source.IsValid() &&
+                source.IsValidAction(allowedActions) &&
+                source.IsValidSender(allowedSenders);
+
+            //
+            return result;
+        }
+
+        /// <summary>
+        /// Parse Url and Extract Host address ...
+        /// </summary>
+        /// <param name="source"></param>
+        /// <returns></returns>
+        public static string GetRabbitMQHost(this XEventServiceConfiguration source)
+        {
+            //
+            var result = string.Empty;
+
+            //
+            if (!source.IsValid())
+            {
+                XException.InvalidConfiguration.Throw();
+            }
+
+            //
+            var parts = source.Url.Split(':');
+            if (parts.Length != 2)
+            {
+                XException.InvalidConfiguration.Throw();
+            }
+
+            //
+            result = parts[0];
+            if (result.IsNullOrEmpty() ||
+                !result.IsValidUrl()
+            )
+            {
+                XException.InvalidConfiguration.Throw();
+            }
+
+            //
+            return result;
+        }
+
+        /// <summary>
+        /// Parse Url and Extract Port Number ...
+        /// </summary>
+        /// <param name="source"></param>
+        /// <returns></returns>
+        public static int GetRabbitMQPort(this XEventServiceConfiguration source)
+        {
+            //
+            int result = -1;
+
+            //
+            if (!source.IsValid())
+            {
+                XException.InvalidConfiguration.Throw();
+            }
+
+            //
+            var parts = source.Url.Split(':');
+            if (parts.Length != 2)
+            {
+                XException.InvalidConfiguration.Throw();
+            }
+
+            //
+            // Try Parse Port as Integer ...
+            int.TryParse(parts[1], out result);
+            if (result <= 0)
+            {
+                XException.InvalidConfiguration.Throw();
+            }
+
+            //
+            return result;
+        }
+    }
+}

Interfaces/IXEventBus.cs

@@ -1,11 +0,0 @@
-using xEventService.Models;
-
-namespace xEventService.Interfaces {
-    public interface IXEventBus {
-        bool Publish<T> (T @event) where T : XEvent;
-
-        bool Subscribe<TEvent, THandler> ()
-        where TEvent : XEvent
-        where THandler : IXEventHandler<TEvent>;
-    }
-}

Interfaces/IXEventHandler.cs

@@ -1,11 +0,0 @@
-using System.Threading.Tasks;
-using xEventService.Models;
-
-namespace xEventService.Interfaces {
-    public interface IXEventHandler<in TEvent> : IXEventHandler
-    where TEvent : XEvent {
-        Task HandleAsync (TEvent @event);
-    }
-
-    public interface IXEventHandler { }
-}

Interfaces/IXEventServiceProvider.cs

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

Interfaces/IXKafkaProducerService.cs

@@ -0,0 +1,8 @@
+namespace xEventService.Interfaces
+{
+    /// <summary>
+    /// a Service for Produce Kafka Messages ...
+    /// </summary>
+    public interface IXKafkaProducerService : IXProducerServiceBase
+    { }
+}

Interfaces/IXProducerServiceBase.cs

@@ -0,0 +1,35 @@
+using System;
+using System.Threading;
+using xEventService.Models;
+using System.Threading.Tasks;
+using xEventService.Constants;
+
+namespace xEventService.Interfaces
+{
+    public interface IXProducerServiceBase : IDisposable
+    {
+        /// <summary>
+        /// Sending Message ...
+        /// </summary>
+        /// <param name="request">an Instance of <see cref="XEventRequest"/> to Provides Kafka Messaging requirement ...</param>
+        /// <param name="cancellationToken"></param>
+        /// <returns></returns>
+        Task<bool> SendMessageAsync(
+            XEventRequest request,
+            CancellationToken cancellationToken = default
+        );
+
+        /// <summary>
+        /// Sending a Message ...
+        /// </summary>
+        /// <param name="message">an Instance of <see cref="XEventMessage"/> to Provides Kafka Messaging requirement ...</param>
+        /// <param name="topic">Topic for Sending Message ...</param>
+        /// <param name="cancellationToken"></param>
+        /// <returns></returns>
+        Task<bool> SendMessageAsync(
+            XEventMessage message,
+            string topic = XEventServiceConstants.XDefaultTopic,
+            CancellationToken cancellationToken = default
+        );
+    }
+}

Interfaces/IXRabbitMQProducerService.cs

@@ -0,0 +1,33 @@
+using System.Threading;
+using xEventService.Models;
+using System.Threading.Tasks;
+
+namespace xEventService.Interfaces
+{
+    /// <summary>
+    /// RabbitMQ Producer Service Implementation ...
+    /// </summary>    
+    public interface IXRabbitMQProducerService : IXProducerServiceBase
+    {
+        /// <summary>
+        /// Sending a Message to Default Exchange ...
+        /// </summary>
+        /// <param name="message">
+        /// an Instance of <see cref="XEventMessage"/> ...
+        /// </param>
+        /// <param name="exchange">
+        /// Target Exchange Name. if empty, uses default from configuration ...
+        /// </param>
+        /// <param name="routingKey">
+        /// Routing Key for message routing ...
+        /// </param>
+        /// <param name="cancellationToken"></param>
+        /// <returns></returns>
+        Task<bool> SendMessageAsync(
+            XEventMessage message,
+            string exchange = null,
+            string routingKey = null,
+            CancellationToken cancellationToken = default
+        );
+    }
+}

Interfaces/IXZeroMQProducerService.cs

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

Models/XChannel.cs

@@ -1,224 +0,0 @@
-using System;
-using RabbitMQ.Client;
-using RabbitMQ.Client.Events;
-using xCommons.Extensions;
-using xEventService.Configuration;
-using xEventService.Extensions;
-using xExceptions.Constants;
-
-namespace xEventService.Models {
-    public class XChannel {
-        //
-        public IModel Channel { get; protected set; }
-        public string QueueName { get; protected set; }
-        public string ExchangeName { get; protected set; }
-        public IConnection Connection { get; protected set; }
-
-        //
-        #region Constructors ...
-        public XChannel (
-            string exchangeName = null,
-            IConnection connection = null,
-            IModel channel = null,
-            EventingBasicConsumer consumer = null
-        ) {
-            //
-            // Validate Connection ...
-            if (connection.IsNull ()) {
-                XException.InvalidArgs.Throw ();
-            }
-            Connection = connection;
-
-            //
-            // Validate Channel ...
-            if (channel.IsNull ()) {
-                channel = connection.CreateModel ();
-            }
-            Channel = channel;
-
-            //
-            // Check Queue Name and it's Declaration ...
-            if (!exchangeName.IsNullOrEmpty ()) {
-                //
-                ExchangeName = exchangeName;
-            }
-        }
-
-        public XChannel (
-            XEventServiceConfiguration configuration,
-            bool dispatchConsumersAsync = true,
-            string exchangeName = null
-        ) {
-            //
-            // Generate Factory ...
-            var factory = configuration.GetConnectionFactory ();
-            if (dispatchConsumersAsync) {
-                factory.DispatchConsumersAsync = true;
-            }
-
-            //
-            // Create Cahnnel ...
-            Connection = factory.CreateConnection ();
-            Channel = Connection.CreateModel ();
-
-            //
-            // Check Exchange Declaration if provided ...
-            if (!exchangeName.IsNullOrEmpty ()) {
-                //
-                ExchangeName = exchangeName;
-                var exchangeArgs = configuration.GetExchangeArgs (ExchangeName);
-
-                //
-                // Declaring Exchange ...
-                Channel.ExchangeDeclare (
-                    exchange: exchangeName,
-                    type: exchangeArgs.Type,
-                    durable: exchangeArgs.Durable,
-                    autoDelete: exchangeArgs.AutoDelete
-                );
-
-                //
-                // Create a Queue ...
-                var queueName = $"{exchangeName}[{Guid.NewGuid().ToString().GetDigits()}]";
-                var queueArgs = configuration.GetQueueArgs (queueName);
-                QueueName = queueName;
-                Channel.QueueDeclare (
-                    queue: queueName,
-                    durable: queueArgs.Durable,
-                    exclusive: queueArgs.Exclusive,
-                    autoDelete: queueArgs.AutoDelete
-                );
-
-                //
-                // Bind Queue to Exchange ...
-                Channel.QueueBind (
-                    queue: queueName,
-                    exchange: exchangeName,
-                    routingKey: ""
-                );
-            }
-        }
-        #endregion
-
-        //
-        #region Actions ...
-        /// <summary>
-        /// Close Channel ...
-        /// </summary>
-        public void CloseChannel () {
-            //
-            // Check Channel Exists ...
-            if (!this.Channel.IsNull ()) {
-                //
-                // Close Channel if it's Open ...
-                if (this.Channel.IsOpen) {
-                    this.Channel.Close ();
-                }
-
-                //
-                this.Channel.Dispose ();
-                this.Channel = null;
-            }
-        }
-
-        /// <summary>
-        /// Close Connection ...
-        /// </summary>
-        public void CloseConnection () {
-            //
-            // Check Connection Exists ...
-            if (!this.Connection.IsNull ()) {
-                //
-                // Close Connection if it's Open ...
-                if (this.Connection.IsOpen) {
-                    this.Connection.Close ();
-                }
-
-                //
-                this.Connection.Dispose ();
-                this.Connection = null;
-            }
-        }
-
-        /// <summary>
-        /// Publish a Message through Channel on Declare Queue ...
-        /// </summary>
-        /// <param name="message"></param>
-        /// <typeparam name="T"></typeparam>
-        /// <returns></returns>
-        public bool Publish<T> (T message) {
-            //
-            // Validate Args ...
-            if (this.Connection.IsNull () ||
-                !this.Connection.IsOpen ||
-                this.Channel.IsNull () ||
-                !this.Channel.IsOpen) {
-                //
-                Console.WriteLine ($"XEventService Exception: Publish Failed ...");
-
-                //
-                XException.InvalidArgs.Throw ();
-            }
-
-            //
-            // Check Publish Method ...
-            //
-            // Check Exchange Name ...
-            if (this.ExchangeName.IsNullOrEmpty ()) {
-                //
-                Console.WriteLine ($"XEventService Exception: Publish Failed, Invalid Exchange Name {ExchangeName} ...");
-
-                //
-                XException.InvalidArgs.Throw ();
-            }
-
-            //
-            try {
-                //
-                // Publish Message through Channel to Declared Queue ...
-                var body = message.ToBody ();
-                this.Channel.BasicPublish (
-                    exchange: ExchangeName,
-                    routingKey: "",
-                    basicProperties : null,
-                    body : body
-                );
-
-                //
-                return true;
-            } catch (Exception ex) {
-                //
-                Console.WriteLine ($"XEventService Exception: Publish Failed, {ex.Message} ...");
-
-                //
-                return false;
-            }
-        }
-
-        public void Consume (EventingBasicConsumer consumer) {
-            Channel.BasicConsume (
-                queue: QueueName,
-                autoAck: true,
-                consumer: consumer
-            );
-        }
-
-        public void ConsumeAsync (AsyncEventingBasicConsumer consumer) {
-            Channel.BasicConsume (
-                queue: QueueName,
-                autoAck: true,
-                consumer: consumer
-            );
-        }
-
-        /// <summary>
-        /// Dispose XChannel ...
-        /// </summary>
-        public void Dispose () {
-            //
-            this.CloseChannel ();
-            this.CloseConnection ();
-        }
-        #endregion        
-    }
-}

Models/XEvent.cs

@@ -1,5 +0,0 @@
-using xEventService.Base;
-
-namespace xEventService.Models {
-    public class XEvent : XBaseEvent { }
-}

Models/XEventMessage.cs

@@ -0,0 +1,40 @@
+using System;
+using Newtonsoft.Json;
+using System.Collections.Generic;
+
+namespace xEventService.Models
+{
+    /// <summary>
+    /// a Message Descriptor for Events ...
+    /// </summary>
+    public class XEventMessage
+    {
+        /// <summary>
+        /// Specified Action for Fire ...
+        /// </summary>
+        [JsonRequired]
+        public string Action { get; set; }
+
+        /// <summary>
+        /// Specified Action Sender ...
+        /// </summary>
+        [JsonRequired]
+        public string Sender { get; set; }
+
+        /// <summary>
+        /// Sending Time ...
+        /// </summary>
+        [JsonRequired]
+        public DateTime Offset { get; set; }
+
+        /// <summary>
+        /// Specified Message ...
+        /// </summary>
+        public string Message { get; set; }
+
+        /// <summary>
+        /// Metadata for Message ...
+        /// </summary>
+        public IDictionary<string, string> Payload { get; set; } = new Dictionary<string, string>();
+    }
+}

Models/XEventRequest.cs

@@ -0,0 +1,20 @@
+using xEventService.Constants;
+
+namespace xEventService.Models
+{
+    /// <summary>
+    /// a Request for Sending an Event ...
+    /// </summary>
+    public class XEventRequest
+    {
+        /// <summary>
+        /// Specified Which Kafka Topic ...
+        /// </summary>
+        public string Topic { get; set; } = XEventServiceConstants.XDefaultTopic;
+
+        /// <summary>
+        /// Specified Kafka Meesage to Send ...
+        /// </summary>
+        public XEventMessage Message { get; set; }
+    }
+}

Models/XRabbitMQOptions.cs

@@ -0,0 +1,92 @@
+using xEventService.Constants;
+
+namespace xEventService.Models
+{
+    /// <summary>
+    /// RabbitMQ-specific connection and messaging options ...
+    /// </summary>    
+    public class XRabbitMQOptions
+    {
+
+        /// <summary>
+        /// Connection Username ...
+        /// </summary>
+        public string Username { get; set; } = "guest";
+
+        /// <summary>
+        /// Connection Password ...
+        /// </summary>
+        public string Password { get; set; } = "guest";
+
+        /// <summary>
+        /// Virtual Host ...
+        /// </summary>
+        public string VirtualHost { get; set; } = "/";
+
+        /// <summary>
+        /// Default Exchange Name ...
+        /// </summary>
+        public string Exchange { get; set; } = "x.saherelm.events";
+
+        /// <summary>
+        /// Default Exchange Type (fanout, direct, topic, headers) ...
+        /// </summary>
+        public XRabbitMQExchangeType ExchangeType { get; set; } = XRabbitMQExchangeType.Fanout;
+
+        /// <summary>
+        /// Default Queue Name ...
+        /// if empty, a dynamic queue will be generated ...
+        /// </summary>
+        public string Queue { get; set; } = string.Empty;
+
+        /// <summary>
+        /// Default Routing Key ...
+        /// </summary>
+        public string RoutingKey { get; set; } = string.Empty;
+
+        /// <summary>
+        /// Exchange Durable Flag ...
+        /// </summary>
+        public bool ExchangeDurable { get; set; } = true;
+
+        /// <summary>
+        /// Exchange AutoDelete Flag ...
+        /// </summary>
+        public bool ExchangeAutoDelete { get; set; } = false;
+
+        /// <summary>
+        /// Queue Durable Flag ...
+        /// </summary>
+        public bool QueueDurable { get; set; } = true;
+
+        /// <summary>
+        /// Queue Exclusive Flag ...
+        /// </summary>
+        public bool QueueExclusive { get; set; } = false;
+
+        /// <summary>
+        /// Queue AutoDelete Flag ...
+        /// </summary>
+        public bool QueueAutoDelete { get; set; } = false;
+
+        /// <summary>
+        /// Enable Automatic Recovery on Connection Loss ...
+        /// </summary>
+        public bool AutomaticRecoveryEnabled { get; set; } = true;
+
+        /// <summary>
+        /// Network Recovery Interval in Seconds ...
+        /// </summary>
+        public int NetworkRecoveryIntervalSeconds { get; set; } = 5;
+
+        /// <summary>
+        /// Prefetch Count for Consumer ...
+        /// </summary>
+        public ushort PrefetchCount { get; set; } = 1;
+
+        /// <summary>
+        /// Enable Manual Acknowledgement ...
+        /// </summary>
+        public bool ManualAck { get; set; } = true;
+    }
+}

Models/XZeroMQOptions.cs

@@ -0,0 +1,28 @@
+using xEventService.Constants;
+
+namespace xEventService.Models
+{
+    public class XZeroMQOptions
+    {
+        /// <summary>
+        /// Socket Pattern (PubSub, PushPull) ...
+        /// </summary>
+        public XZeroMQPattern Pattern { get; set; } = XZeroMQPattern.PubSub;
+
+        /// <summary>
+        /// Bind or Connect ...
+        /// True for Server (Bind), False for Client (Connect) ...
+        /// </summary>
+        public bool IsServer { get; set; } = false;
+
+        /// <summary>
+        /// High Water Mark for Sending ...
+        /// </summary>
+        public int SendHighWaterMark { get; set; } = 1000;
+
+        /// <summary>
+        /// High Water Mark for Receiving ...
+        /// </summary>
+        public int ReceiveHighWaterMark { get; set; } = 1000;
+    }
+}

Providers/XBaseEventConsumerBackgroundService.cs

@@ -0,0 +1,65 @@
+using System;
+using xCommons.Providers;
+using xCommons.Extensions;
+using xExceptions.Constants;
+using xEventService.Constants;
+using xEventService.Extensions;
+using xEventService.Configuration;
+using Microsoft.Extensions.Logging;
+
+namespace xEventService.Providers
+{
+    /// <summary>
+    /// Base Event Consumer Service ...
+    /// </summary>
+    public abstract class XBaseEventConsumerBackgroundService : XResumableBackgroundService
+    {
+        /// <summary>
+        /// Consumed Topic Messages ...
+        /// </summary>
+        protected readonly string topic;
+
+        /// <summary>
+        /// Allowed Senders ...
+        /// </summary>
+        protected readonly string[] allowedSenders;
+
+        /// <summary>
+        /// Allowed Actions ...
+        /// </summary>
+        protected readonly string[] allowedActions;
+
+        /// <summary>
+        /// Configuration ...
+        /// </summary>
+        protected readonly XEventServiceConfiguration configuration;
+
+        protected XBaseEventConsumerBackgroundService(
+            string[] allowedActions,
+            string[] allowedSenders,
+            IServiceProvider serviceProvider,
+            XEventServiceConfiguration configuration,
+            ILogger<XResumableBackgroundService> logger,
+            string topic = XEventServiceConstants.XDefaultTopic
+        ) : base(serviceProvider, logger)
+        {
+            //
+            // Validate Actions ...
+            if (!allowedActions
+                .IsAllowedCollection() ||
+                !allowedSenders
+                .IsAllowedCollection() ||
+                !configuration.IsValid()
+            )
+            {
+                XException.InvalidArgs.Throw();
+            }
+
+            //
+            this.topic = topic;
+            this.configuration = configuration;
+            this.allowedActions = allowedActions;
+            this.allowedSenders = allowedSenders;
+        }
+    }
+}

Providers/XBaseEventProducerService.cs

@@ -0,0 +1,31 @@
+using xCommons.Extensions;
+using xExceptions.Constants;
+using xEventService.Extensions;
+using xEventService.Configuration;
+using Microsoft.Extensions.Logging;
+
+namespace xEventService.Providers
+{
+    public abstract class XBaseEventProducerService
+    {
+        protected readonly XEventServiceConfiguration configuration;
+        protected readonly ILogger<XBaseEventProducerService> logger;
+
+        protected XBaseEventProducerService(
+            XEventServiceConfiguration configuration,
+            ILogger<XBaseEventProducerService> logger
+        )
+        {
+            //
+            this.logger = logger;
+            this.configuration = configuration;
+
+            //
+            // Validate Configurations ...
+            if (!configuration.IsValid())
+            {
+                XException.InvalidConfiguration.Throw();
+            }
+        }        
+    }
+}

Providers/XEventBus.cs

@@ -1,260 +0,0 @@
-using System;
-using System.Collections.Generic;
-using System.Linq;
-using System.Text;
-using System.Threading.Tasks;
-using Microsoft.Extensions.DependencyInjection;
-using Microsoft.Extensions.Logging;
-using RabbitMQ.Client.Events;
-using xCommons.Extensions;
-using xEventService.Configuration;
-using xEventService.Extensions;
-using xEventService.Interfaces;
-using xEventService.Models;
-using xExceptions.Constants;
-
-namespace xEventService.Providers {
-    public sealed class XEventBus : IXEventBus {
-        //
-        #region Properties ...
-        private readonly List<Type> events;
-        private readonly ILogger<XEventBus> logger;
-        private readonly IServiceScopeFactory serviceScopeFactory;
-        private readonly XEventServiceConfiguration configuration;
-        private readonly IDictionary<string, List<Type>> handlers;
-        #endregion
-
-        //
-        #region Constructors ...
-        public XEventBus (
-            ILoggerFactory loggerFactory,
-            IServiceScopeFactory serviceScopeFactory,
-            XEventServiceConfiguration configuration
-        ) {
-            //
-            this.configuration = configuration;
-            this.serviceScopeFactory = serviceScopeFactory;
-
-            //
-            events = new List<Type> ();
-            handlers = new Dictionary<string, List<Type>> ();
-            logger = loggerFactory.CreateLogger<XEventBus> ();
-        }
-        #endregion
-
-        //
-        #region Abstracts ...
-        public bool Publish<T> (T @event) where T : XEvent {
-            //
-            // Validate Args ...
-            if (@event.IsNull ()) {
-                return false;
-            }
-
-            //
-            // Prepare Channel ...
-            XChannel channel = null;
-
-            //
-            // Extract event name ...
-            var exchangeName = @event.GetType ().Name;
-
-            //
-            // Do Action ...
-            try {
-                //
-                // Get XChannel Model ...
-                channel = configuration.CreateChannel (exchangeName);
-
-                //
-                // Publish Command ...
-                channel.Publish (message: @event);
-
-                //
-                // Return Result ...
-                return true;
-            } catch (Exception ex) {
-                //
-                logger.LogError ($"XEventService Exception: Publishing Failed {exchangeName}/{ex.Message} ...");
-
-                //
-                return false;
-            } finally {
-                //
-                if (!channel.IsNull ()) {
-                    channel.Dispose ();
-                }
-            }
-        }
-
-        public bool Subscribe<TEvent, THandler> ()
-        where TEvent : XEvent
-        where THandler : IXEventHandler<TEvent> {
-            //
-            // Extract required data ...
-            var eventType = typeof (TEvent);
-            var eventName = eventType.Name;
-            var handlerType = typeof (THandler);
-
-            //
-            // Check event exists in list or not ...
-            if (!events.Contains (eventType)) {
-                //
-                // Add it if not exists ...
-                events.Add (eventType);
-            }
-
-            //
-            // Check handlers Dictionary contains list or not ...
-            if (!handlers.ContainsKey (eventName)) {
-                handlers.Add (eventName, new List<Type> ());
-            }
-
-            //
-            // Check handlers Subscribed before or not ...
-            var isHandlerSubscribed = handlers[eventName]
-                .Any (h => h.GetType () == handlerType);
-            if (isHandlerSubscribed) {
-                //
-                logger.LogError ($"XEventService Exception: Duplicate Event Handler Subscription {eventName}/{handlerType.Name} ...");
-
-                //
-                return false;
-            }
-
-            //
-            // Register Handler ...
-            handlers[eventName].Add (handlerType);
-
-            //
-            // Do Consuming Handler ...
-            ConsumeEvent<TEvent> ();
-
-            //
-            return true;
-        }
-        #endregion
-
-        //
-        #region Private ...
-        /// <summary>
-        /// Consume Specific Event ...
-        /// </summary>
-        /// <typeparam name="TEvent"></typeparam>
-        private void ConsumeEvent<TEvent> ()
-        where TEvent : XEvent {
-            //
-            // Extract required data ...
-            var eventType = typeof (TEvent);
-            var eventName = eventType.Name;
-
-            //
-            // Create ChannelObject ...
-            var channel = new XChannel (
-                exchangeName: eventName,
-                configuration: configuration,
-                dispatchConsumersAsync: true
-            );
-
-            //
-            // Create Consumer ...
-            var consumer = new AsyncEventingBasicConsumer (channel.Channel);
-
-            //
-            // Set Consumer Delegate ...
-            consumer.Received += XConsumerReceivedDelegate;
-
-            //
-            // Consume Async Consumer ...
-            channel.ConsumeAsync (consumer);
-        }
-
-        /// <summary>
-        /// this Delegate method calls when a message recieved ...
-        /// </summary>
-        private async Task XConsumerReceivedDelegate (
-            object sender,
-            BasicDeliverEventArgs ea
-        ) {
-            //
-            // Generate Required Data ...
-            var eventName = ea.Exchange;
-            var json = Encoding.UTF8
-                .GetString (
-                    ea.Body
-                    .ToArray ()
-                );
-
-            //
-            // Do Processing Event ...
-            try {
-                //
-                // Process Event in non Blocking Tasks ...
-                await ProcessEvent (eventName, json)
-                    .ConfigureAwait (false);
-            } catch (Exception ex) {
-                //
-                logger.LogError ($"XEventService Exception: Processing Event failed {eventName}/{ex.Message} ...");
-
-                //
-                XException.ActionFailed.Throw ();
-            } finally {
-                //
-                var channel = ((EventingBasicConsumer) sender).Model;
-                channel.BasicAck (
-                    deliveryTag: ea.DeliveryTag,
-                    multiple: false
-                );
-            }
-        }
-
-        /// <summary>
-        /// Here we must call registered Event Handlers to Handle the Event ...
-        /// </summary>
-        private async Task ProcessEvent (
-            string eventName,
-            string json
-        ) {
-            //
-            // Check Handler Exists for current Event ...
-            var isExistsHandler = handlers.ContainsKey (eventName);
-            if (!isExistsHandler) {
-                return;
-            }
-
-            //
-            // Using Service Scope Factory ...
-            using (var scope = serviceScopeFactory.CreateScope ()) {
-                //
-                // Get Handler Subscriptions ...
-                var subscriptions = handlers[eventName];
-                foreach (var subscription in subscriptions) {
-                    //
-                    // Inject Handler from Dependency Injections ...
-                    var handler = scope.ServiceProvider.GetService (subscription);
-                    if (handler.IsNull ()) {
-                        //
-                        logger.LogInformation ($"XEventService: there is no handler Registered in DependencyInjection for {eventName}/{subscription.Name}");
-
-                        //
-                        continue;
-                    }
-
-                    //
-                    // Retrieve Event Type ...
-                    var eventType = events.SingleOrDefault (t => t.Name == eventName);
-                    var @event = json.FromJSON (eventType);
-                    var conreteType = typeof (IXEventHandler<>).MakeGenericType (eventType);
-
-                    //
-                    // Call EventHandler 'HandleAsync' method ...
-                    await ((Task) conreteType
-                        .GetMethod (nameof (IXEventHandler<XEvent>.HandleAsync))
-                        .Invoke (handler, new object[] { @event }))
-                    .ConfigureAwait (true);
-                }
-            }
-        }
-        #endregion
-    }
-}

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

@@ -0,0 +1,142 @@
+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.Configuration;
+using Microsoft.Extensions.Logging;
+
+namespace xEventService.Providers
+{
+    /// <summary>
+    /// an Abstract Kafka Consumer Service ...
+    /// </summary>
+    public abstract class XKafkaConsumerServiceBase : XBaseEventConsumerBackgroundService
+    {
+        /// <summary>
+        /// Kafka Consumer Class ...
+        /// </summary>
+        protected readonly IConsumer<Ignore, string> consumer;
+
+        protected XKafkaConsumerServiceBase(
+            string[] allowedActions,
+            string[] allowedSenders,
+            IServiceProvider serviceProvider,
+            XEventServiceConfiguration configuration,
+            ILogger<XKafkaConsumerServiceBase> logger,
+            string topic = XEventServiceConstants.XDefaultTopic
+        ) : base(
+            topic: topic,
+            logger: logger,
+            configuration: configuration,
+            allowedActions: allowedActions,
+            allowedSenders: allowedSenders,
+            serviceProvider: serviceProvider
+        )
+        {
+            //
+            // Validating Configuration ...
+            if (configuration.Broker != XEventBroker.XKafka)
+            {
+                XException.InvalidConfiguration.Throw();
+            }
+
+            //
+            // Prepare Kafka Consumer Configuration ...
+            var config = new ConsumerConfig
+            {
+                EnableAutoCommit = false,
+                BootstrapServers = configuration.Url,
+                AutoOffsetReset = AutoOffsetReset.Earliest,
+                GroupId = XEventServiceConstants.XDefaultConsumerGroup,
+            };
+            consumer = new ConsumerBuilder<Ignore, string>(config).Build();
+        }
+
+        /// <summary>
+        /// Default Action Execution for Background Services ...
+        /// </summary>
+        /// <param name="cancellationToken"></param>
+        /// <returns></returns>
+        protected override async Task ExecuteAsync(
+            CancellationToken cancellationToken = default
+        )
+        {
+            //
+            // For Fix Blocking Synchronous ...
+            await Task.Yield();
+            using var cts = new CancellationTokenSource();
+
+            //
+            // Subscribe to Topic ...
+            consumer.Subscribe(topic);
+            logger.LogInformation($"Kafka Cosume Topic: {topic} ...");
+
+            //
+            // Consuming ...
+            while (!cancellationToken.IsCancellationRequested)
+            {
+                try
+                {
+                    //
+                    // Consuming Topic ...
+                    var consumeResult = consumer.Consume(cancellationToken);
+                    if (!consumeResult.IsNullOrDefault() &&
+                        !consumeResult.Message.IsNullOrDefault()
+                    )
+                    {
+                        //
+                        var message = consumeResult
+                            .Message
+                            .Value
+                            .FromJSON<XEventMessage>();
+                        if (message.IsValid(
+                            allowedSenders: allowedSenders,
+                            allowedActions: allowedActions)
+                        )
+                        {
+                            //
+                            await DoWorkAsync(
+                                payload: message,
+                                cancellationToken: cancellationToken
+                            );
+                        }
+                    }
+                }
+                catch (ConsumeException ex)
+                {
+                    //
+                    logger.LogError($"Kafka Consumer Result Parsing Error: {ex.Message} ...");
+                    await Task.Delay(1000, cancellationToken);
+                }
+                catch (Exception ex)
+                {
+                    //
+                    logger.LogError($"Kafka Consumer Result Parsing Error: {ex.Message} ...");
+                    await Task.Delay(1000, cancellationToken);
+                }
+            }
+        }
+
+        /// <summary>
+        /// Dispose Object ...
+        /// </summary>
+        public override void Dispose()
+        {
+            //
+            if (consumer != null)
+            {
+                consumer.Close();
+                consumer.Dispose();
+            }
+
+            //
+            GC.SuppressFinalize(this);
+            base.Dispose();
+        }
+    }
+}

Providers/XKafkaProducerService.cs

@@ -0,0 +1,140 @@
+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 Microsoft.Extensions.Logging;
+
+namespace xEventService.Providers
+{
+    public class XKafkaProducerService : XBaseEventProducerService, IXKafkaProducerService
+    {
+        private readonly IProducer<Null, string> producer;
+
+        public XKafkaProducerService(
+            ILogger<XKafkaProducerService> logger,
+            XEventServiceConfiguration configuration
+        ) : base(configuration, logger)
+        {
+            //
+            // Validate Configurations ...
+            if (!configuration.IsValid() ||
+                configuration.Broker != XEventBroker.XKafka
+            )
+            {
+                XException.InvalidConfiguration.Throw();
+            }
+
+            //
+            // Prepare Kafka Producer Configuration ...
+            var config = new ProducerConfig
+            {
+                Acks = Acks.All,
+                BootstrapServers = configuration.Url,
+                MessageSendMaxRetries = configuration.MaxRetries,
+            };
+            producer = new ProducerBuilder<Null, string>(config).Build();
+        }
+
+        /// <summary>
+        /// Sending Message ...
+        /// </summary>
+        /// <param name="request">an Instance of <see cref="XEventRequest"/> to Provides Kafka Messaging requirement ...</param>
+        /// <param name="cancellationToken"></param>
+        /// <returns></returns>
+        public async Task<bool> SendMessageAsync(
+            XEventRequest request,
+            CancellationToken cancellationToken = default
+        )
+        {
+            //
+            // Validate ...
+            var result = request.IsValid();
+            if (!result)
+            {
+                XException.InvalidArgs.Throw();
+            }
+
+            //
+            result = await SendMessageAsync(
+                topic: request.Topic,
+                message: request.Message,
+                cancellationToken: cancellationToken
+            );
+
+            //
+            return result;
+        }
+
+        /// <summary>
+        /// Sending a Message ...
+        /// </summary>
+        /// <param name="message">an Instance of <see cref="XEventMessage"/> to Provides Kafka Messaging requirement ...</param>
+        /// <param name="topic">Topic for Sending Message ...</param>
+        /// <param name="cancellationToken"></param>
+        /// <returns></returns>
+        public async Task<bool> SendMessageAsync(
+            XEventMessage message,
+            string topic = XEventServiceConstants.XDefaultTopic,
+            CancellationToken cancellationToken = default
+        )
+        {
+            //
+            // Validate ...
+            var result = message.IsValid() &&
+                !topic.IsNullOrEmpty();
+            if (!result)
+            {
+                XException.InvalidArgs.Throw();
+            }
+
+            //
+            try
+            {
+                //
+                var response = await producer.ProduceAsync(
+                    topic: topic,
+                    message: new Message<Null, string>
+                    {
+                        Value = message
+                            .ToJSON(camelCase: true)
+                    },
+                    cancellationToken: cancellationToken
+                );
+
+                //
+                result = !response.IsNullOrDefault();
+            }
+            catch (Exception ex)
+            {
+                //
+                logger.LogError($"Kafka Produce Error: {ex.Message} ...");
+                result = false;
+            }
+
+            //
+            return result;
+        }
+
+        /// <summary>
+        /// Dispose Required Objects ...
+        /// </summary>
+        public void Dispose()
+        {
+            //
+            if (producer != null)
+            {
+                producer.Dispose();
+            }
+
+            //
+            GC.SuppressFinalize(this);
+        }
+    }
+}

Providers/XRabbitMQConsumerServiceBase.cs

@@ -0,0 +1,277 @@
+using System;
+using System.Text;
+using RabbitMQ.Client;
+using System.Threading;
+using xCommons.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
+{
+    /// <summary>
+    /// an Abstract Kafka Consumer Service ...
+    /// </summary>
+    public abstract class XRabbitMQConsumerServiceBase : XBaseEventConsumerBackgroundService
+    {
+        //
+        protected IModel channel;
+        protected IConnection connection;
+        protected AsyncEventingBasicConsumer consumer;
+
+        protected XRabbitMQConsumerServiceBase(
+            string[] allowedActions,
+            string[] allowedSenders,
+            IServiceProvider serviceProvider,
+            XEventServiceConfiguration configuration,
+            ILogger<XRabbitMQConsumerServiceBase> logger,
+            string topic = XEventServiceConstants.XDefaultTopic
+        ) : base(
+            topic: topic,
+            logger: logger,
+            configuration: configuration,
+            allowedActions: allowedActions,
+            allowedSenders: allowedSenders,
+            serviceProvider: serviceProvider
+        )
+        {
+            //
+            // Validating Configuration ...
+            if (!configuration.RabbitMQ.IsValid() ||
+                configuration.Broker != XEventBroker.XRabbitMQ
+            )
+            {
+                XException.InvalidConfiguration.Throw();
+            }
+
+            //
+            SetupConnection();
+            DeclareTopology(topic);
+            consumer = new AsyncEventingBasicConsumer(channel);
+        }
+
+        /// <summary>
+        /// Default Action Execution for Background Services ...
+        /// </summary>
+        /// <param name="cancellationToken"></param>
+        /// <returns></returns>
+        protected override async Task ExecuteAsync(
+            CancellationToken cancellationToken = default
+        )
+        {
+            //
+            // For Fix Blocking Synchronous ...
+            await Task.Yield();
+            using var cts = new CancellationTokenSource();
+            logger.LogInformation($"RabbitMQ Cosume Topic: {topic} ...");
+
+            //
+            // Subscribe to Topic ...
+            consumer.Received += async (model, ea) =>
+            {
+                //
+                try
+                {
+                    //
+                    var body = ea.Body.ToArray();
+                    var json = Encoding.UTF8.GetString(body);
+                    var message = json.FromJSON<XEventMessage>();
+                    if (!message.IsNullOrDefault() &&
+                        message.IsValid(
+                        allowedSenders: allowedSenders,
+                        allowedActions: allowedActions)
+                    )
+                    {
+                        //
+                        await DoWorkAsync(
+                            payload: message,
+                            cancellationToken: cancellationToken
+                        );
+                    }
+
+                    //
+                    if (configuration.RabbitMQ.ManualAck)
+                    {
+                        //
+                        channel.BasicAck(
+                            multiple: false,
+                            deliveryTag: ea.DeliveryTag
+                        );
+                    }
+                }
+                catch (Exception ex)
+                {
+                    //
+                    logger.LogError($"RabbitMQ Cosume Error: {ex.Message} ...");
+                    if (configuration.RabbitMQ.ManualAck)
+                    {
+                        //
+                        channel.BasicNack(
+                            requeue: true,
+                            multiple: false,
+                            deliveryTag: ea.DeliveryTag
+                        );
+                    }
+                }
+            };
+
+            //
+            consumer.Shutdown += (model, ea) =>
+            {
+                return Task.CompletedTask;
+            };
+
+            //
+            channel.BasicQos(
+                prefetchSize: 0,
+                global: false,
+                prefetchCount: configuration.RabbitMQ.PrefetchCount
+            );
+
+            //
+            channel.BasicConsume(
+                queue: GetQueueName(),
+                consumer: consumer,
+                autoAck: !configuration.RabbitMQ.ManualAck
+            );
+
+            //
+            while (!cancellationToken.IsCancellationRequested)
+            {
+                await Task.Delay(1000, cancellationToken);
+            }
+        }
+
+        /// <summary>
+        /// Overriding Stop Action ...
+        /// </summary>
+        /// <param name="cancellationToken"></param>
+        /// <returns></returns>
+        public override async Task StopAsync(CancellationToken cancellationToken)
+        {
+            //
+            if (channel != null)
+            {
+                //
+                channel.Close();
+                channel.Dispose();
+            }
+
+            //
+            if (connection != null)
+            {
+                //
+                connection.Close();
+                connection.Dispose();
+            }
+
+            //
+            await base.StopAsync(cancellationToken);
+        }
+
+        /// <summary>
+        /// Dispose Implementation ...
+        /// </summary>
+        public override void Dispose()
+        {
+            //
+            if (channel != null)
+            {
+                channel.Dispose();
+            }
+
+            //
+            if (connection != null)
+            {
+                connection.Dispose();
+            }
+
+            //
+            GC.SuppressFinalize(this);
+            base.Dispose();
+        }
+
+        //
+        #region Private ...
+        private string GetQueueName()
+        {
+            //
+            var result = string.Empty;
+            if (!configuration.RabbitMQ.Queue.IsNullOrEmpty())
+            {
+                result = configuration.RabbitMQ.Queue;
+            }
+
+            //
+            if (result.IsNullOrEmpty())
+            {
+                //
+                var exchangeName = topic ?? configuration.RabbitMQ.Exchange;
+                var digits = Guid.NewGuid().ToString().GetDigits();
+                result = $"{exchangeName}[{digits}]";
+            }
+
+            //
+            return result;
+        }
+
+        private void SetupConnection()
+        {
+            var factory = new ConnectionFactory
+            {
+                DispatchConsumersAsync = true,
+                Port = configuration.GetRabbitMQPort(),
+                HostName = configuration.GetRabbitMQHost(),
+                UserName = configuration.RabbitMQ.Username,
+                Password = configuration.RabbitMQ.Password,
+                NetworkRecoveryInterval = TimeSpan.FromSeconds(
+                    configuration.RabbitMQ.NetworkRecoveryIntervalSeconds
+                ),
+                VirtualHost = configuration.RabbitMQ.VirtualHost,
+                AutomaticRecoveryEnabled = configuration.RabbitMQ.AutomaticRecoveryEnabled,
+            };
+
+            //
+            connection = factory.CreateConnection();
+            channel = connection.CreateModel();
+        }
+
+        private void DeclareTopology(string topic)
+        {
+            //
+            channel.ExchangeDeclare(
+                arguments: null,
+                exchange: topic ?? configuration.RabbitMQ.Exchange,
+                durable: configuration.RabbitMQ.ExchangeDurable,
+                autoDelete: configuration.RabbitMQ.ExchangeAutoDelete,
+                type: configuration.RabbitMQ.ExchangeType.GetStringValue()
+            );
+
+            //
+            var queueName = GetQueueName();
+
+            //
+            channel.QueueDeclare(
+                arguments: null,
+                queue: queueName,
+                durable: configuration.RabbitMQ.QueueDurable,
+                exclusive: configuration.RabbitMQ.QueueExclusive,
+                autoDelete: configuration.RabbitMQ.QueueAutoDelete
+            );
+
+            //
+            channel.QueueBind(
+                arguments: null,
+                queue: queueName,
+                exchange: topic ?? configuration.RabbitMQ.Exchange,
+                routingKey: configuration.RabbitMQ.RoutingKey
+            );
+        }
+        #endregion
+    }
+}

Providers/XRabbitMQProducerService.cs

@@ -0,0 +1,315 @@
+using System;
+using System.Text;
+using RabbitMQ.Client;
+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 Microsoft.Extensions.Logging;
+
+namespace xEventService.Providers
+{
+    /// <summary>
+    /// RabbitMQ Producer Service Implementation ...
+    /// </summary>
+    public class XRabbitMQProducerService : XBaseEventProducerService, IXRabbitMQProducerService
+    {
+        private IModel channel;
+        private IConnection connection;
+        private readonly object connectionLock = new();
+
+        public XRabbitMQProducerService(
+            ILogger<XRabbitMQProducerService> logger,
+            XEventServiceConfiguration configuration
+        ) : base(configuration, logger)
+        {
+            //
+            // Validate Configurations ...
+            if (!configuration.IsValid() ||
+                !configuration.RabbitMQ.IsValid() ||
+                configuration.Broker != XEventBroker.XRabbitMQ
+            )
+            {
+                XException.InvalidConfiguration.Throw();
+            }
+
+            //
+            // Initialize Connection ...
+            CreateConnection();
+        }
+
+        /// <summary>
+        /// Sending Message with Full Request Descriptor ...
+        /// </summary>
+        /// <param name="request"></param>
+        /// <param name="cancellationToken"></param>
+        /// <returns></returns>
+        public async Task<bool> SendMessageAsync(
+            XEventRequest request,
+            CancellationToken cancellationToken = default
+        )
+        {
+            //
+            // Validate ...
+            var result = request.IsValid();
+            if (!result)
+            {
+                XException.InvalidArgs.Throw();
+            }
+
+            //
+            result = await SendMessageAsync(
+                message: request.Message,
+                exchange: request.Topic,
+                cancellationToken: cancellationToken,
+                routingKey: configuration.RabbitMQ.RoutingKey
+            );
+
+            //
+            return result;
+        }
+
+        /// <summary>
+        /// Sending a Message to Topic ...
+        /// </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
+        )
+        {
+            //
+            var result = await SendMessageAsync(
+                request: new XEventRequest
+                {
+                    Topic = topic,
+                    Message = message
+                },
+                cancellationToken: cancellationToken
+            );
+
+            //
+            return result;
+        }
+
+        /// <summary>
+        /// Sending a Message to Exchange ...
+        /// </summary>
+        /// <param name="message"></param>
+        /// <param name="exchange"></param>
+        /// <param name="routingKey"></param>
+        /// <param name="cancellationToken"></param>
+        /// <returns></returns>
+        public async Task<bool> SendMessageAsync(
+            XEventMessage message,
+            string exchange = null,
+            string routingKey = null,
+            CancellationToken cancellationToken = default
+        )
+        {
+            //
+            // Validate ...
+            var result = message.IsValid();
+            if (!result)
+            {
+                XException.InvalidArgs.Throw();
+            }
+
+            //
+            // Normalize Exchange ...
+            if (exchange.IsNullOrEmpty())
+            {
+                exchange = configuration.RabbitMQ.Exchange;
+            }
+
+            //
+            // Normalize Routing Key ...
+            if (routingKey.IsNull())
+            {
+                routingKey = configuration.RabbitMQ.RoutingKey;
+            }
+
+            //
+            try
+            {
+                //
+                // Ensure Connection is Alive ...
+                EnsureChannel();
+
+                //
+                // Serialize Message ...
+                var body = Encoding.UTF8.GetBytes(
+                    message.ToJSON(camelCase: true)
+                );
+
+                //
+                // Prepare Message Properties ...
+                var properties = channel.CreateBasicProperties();
+                properties.Persistent = true;
+                properties.ContentType = "application/json";
+                properties.Timestamp = new AmqpTimestamp(
+                    DateTimeOffset.UtcNow.ToUnixTimeSeconds()
+                );
+                properties.MessageId = Guid.NewGuid().ToString();
+                properties.Type = message.Action;
+                properties.AppId = message.Sender;
+
+                //
+                // Publish Message ...
+                await Task.Run(() =>
+                {
+                    //
+                    channel.BasicPublish(
+                        exchange: exchange,
+                        routingKey: routingKey,
+                        mandatory: false,
+                        basicProperties: properties,
+                        body: body
+                    );
+                },
+                cancellationToken: cancellationToken);
+
+                //
+                result = true;
+            }
+            catch (Exception ex)
+            {
+                //
+                logger.LogError($"RabbitMQ Produce Error: {ex.Message} ...");
+                result = false;
+            }
+
+            //
+            return result;
+        }
+
+        /// <summary>
+        /// Dispose Object ...
+        /// </summary>
+        public void Dispose()
+        {
+            //
+            try
+            {
+                //
+                if (channel != null)
+                {
+                    //
+                    channel.Close();
+                    channel.Dispose();
+                    channel = null;
+                }
+
+                //
+                if (connection != null)
+                {
+                    //
+                    connection.Close();
+                    connection.Dispose();
+                    connection = null;
+                }
+            }
+            catch
+            { }
+
+            //
+            GC.SuppressFinalize(this);
+        }
+
+        //
+        #region Private ...
+        /// <summary>
+        /// Create Connection to RabbitMQ Server ...
+        /// </summary>
+        private void CreateConnection()
+        {
+            //
+            var factory = new ConnectionFactory
+            {
+                Port = configuration.GetRabbitMQPort(),
+                HostName = configuration.GetRabbitMQHost(),
+                UserName = configuration.RabbitMQ.Username,
+                Password = configuration.RabbitMQ.Password,
+                VirtualHost = configuration.RabbitMQ.VirtualHost,
+                AutomaticRecoveryEnabled = configuration.RabbitMQ.AutomaticRecoveryEnabled,
+                NetworkRecoveryInterval = TimeSpan.FromSeconds(
+                    configuration.RabbitMQ.NetworkRecoveryIntervalSeconds
+                ),
+                DispatchConsumersAsync = true
+            };
+
+            //
+            connection = factory.CreateConnection();
+        }
+
+        /// <summary>
+        /// Check Connection is Alive ...
+        /// </summary>
+        private bool IsConnected()
+        {
+            //
+            var result =
+                !connection.IsNullOrDefault() &&
+                connection.IsOpen &&
+                !channel.IsNull() &&
+                channel.IsOpen;
+
+            //
+            return result;
+        }
+
+        /// <summary>
+        /// Ensure Channel is Open and Exchange/Queue Declared ...
+        /// </summary>
+        private void EnsureChannel()
+        {
+            //
+            if (IsConnected())
+            {
+                return;
+            }
+
+            //
+            lock (connectionLock)
+            {
+                //
+                // Double Check after Lock ...
+                if (IsConnected())
+                {
+                    return;
+                }
+
+                //
+                // Recreate Connection if Dead ...
+                if (connection is null || !connection.IsOpen)
+                {
+                    CreateConnection();
+                }
+
+                //
+                // Create Channel ...
+                channel = connection.CreateModel();
+
+                //
+                // Declare Exchange ...
+                channel.ExchangeDeclare(
+                    arguments: null,
+                    exchange: configuration.RabbitMQ.Exchange,
+                    durable: configuration.RabbitMQ.ExchangeDurable,
+                    autoDelete: configuration.RabbitMQ.ExchangeAutoDelete,
+                    type: configuration.RabbitMQ.ExchangeType.GetStringValue()
+                );
+            }
+        }
+        #endregion
+    }
+}

Providers/XZeroMQConsumerServiceBase.cs

@@ -0,0 +1,206 @@
+using NetMQ;
+using System;
+using System.Text;
+using NetMQ.Sockets;
+using System.Threading;
+using xCommons.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
+{
+    /// <summary>
+    /// an Abstract ZeroMQ Consumer Service ...
+    /// </summary>
+    public abstract class XZeroMQConsumerServiceBase : XBaseEventConsumerBackgroundService
+    {
+        //
+        private NetMQSocket socket;
+
+        protected XZeroMQConsumerServiceBase(
+            string[] allowedActions,
+            string[] allowedSenders,
+            IServiceProvider serviceProvider,
+            XEventServiceConfiguration configuration,
+            ILogger<XZeroMQConsumerServiceBase> logger,
+            string topic = XEventServiceConstants.XDefaultTopic
+        ) : base(
+            topic: topic,
+            logger: logger,
+            configuration: configuration,
+            allowedActions: allowedActions,
+            allowedSenders: allowedSenders,
+            serviceProvider: serviceProvider
+        )
+        {
+            //
+            // Validating Configuration ...
+            if (!configuration.ZeroMQ.IsValid() ||
+                configuration.Broker != XEventBroker.XZeroMQ)
+            {
+                XException.InvalidConfiguration.Throw();
+            }
+
+            //
+            // Initialize ...
+            InitializeSocket();
+        }
+
+        /// <summary>
+        /// Default Action Execution for Background Services ...
+        /// </summary>
+        /// <param name="cancellationToken"></param>
+        /// <returns></returns>
+        protected override async Task ExecuteAsync(
+            CancellationToken cancellationToken = default
+        )
+        {
+            //
+            await Task.Yield();
+            logger.LogInformation($"ZeroMQ Cosume Topic: {topic} ...");
+            while (!cancellationToken.IsCancellationRequested)
+            {
+                //
+                try
+                {
+                    //
+                    string receivedTopic = string.Empty;
+                    byte[] body = null;
+
+                    //
+                    if (configuration.ZeroMQ.Pattern == XZeroMQPattern.PubSub)
+                    {
+                        //
+                        if (socket.TryReceiveFrameString(TimeSpan.FromMilliseconds(100), out receivedTopic))
+                        {
+                            // 
+                            if (socket.TryReceiveFrameBytes(TimeSpan.FromMilliseconds(100), out body))
+                            {
+                                // Process
+                            }
+                            else
+                            {
+                                logger.LogError($"ZeroMQ Cosume Error ...");
+                            }
+                        }
+                    }
+                    else
+                    {
+                        if (socket.TryReceiveFrameBytes(TimeSpan.FromMilliseconds(100), out body))
+                        {
+                            // Process
+                        }
+                        else
+                        {
+                            logger.LogError($"ZeroMQ Cosume Error ...");
+                        }
+                    }
+
+                    //
+                    if (body != null && body.Length > 0)
+                    {
+                        //
+                        var json = Encoding.UTF8.GetString(body);
+                        var message = json.FromJSON<XEventMessage>();
+                        if (!message.IsNullOrDefault() &&
+                            message.IsValid(
+                            allowedSenders: allowedSenders,
+                            allowedActions: allowedActions)
+                        )
+                        {
+                            //
+                            await DoWorkAsync(
+                                payload: message,
+                                cancellationToken: cancellationToken
+                            );
+                        }
+                    }
+                    else
+                    {
+                        await Task.Delay(10, cancellationToken);
+                    }
+                }
+                catch (Exception ex)
+                {
+                    //
+                    logger.LogError($"ZeroMQ Cosume Error: {ex.Message} ...");
+                    await Task.Delay(1000, cancellationToken);
+                }
+            }
+        }
+
+        public override async Task StopAsync(CancellationToken cancellationToken)
+        {
+            //
+            if (socket != null)
+            {
+                socket.Close();
+            }
+
+            //
+            await base.StopAsync(cancellationToken);
+        }
+
+        /// <summary>
+        /// Dispose Object ...
+        /// </summary>
+        public override void Dispose()
+        {
+            //
+            if (socket != null)
+            {
+                socket.Dispose();
+            }
+
+            //
+            GC.SuppressFinalize(this);
+            base.Dispose();
+        }
+
+        //
+        #region Private ...
+        private void InitializeSocket()
+        {
+            //
+            if (configuration.ZeroMQ.Pattern == XZeroMQPattern.PubSub)
+            {
+                //
+                socket = new SubscriberSocket();
+                socket.Options.ReceiveHighWatermark = configuration.ZeroMQ.ReceiveHighWaterMark;
+
+                //  
+                // Subscribe to specific topic or all
+                if (topic.IsNullOrEmpty() || topic == XEventServiceConstants.XWildcard)
+                {
+                    ((SubscriberSocket)socket).SubscribeToAnyTopic();
+                }
+                else
+                {
+                    ((SubscriberSocket)socket).Subscribe(topic);
+                }
+            }
+            else
+            {
+                //
+                socket = new PullSocket();
+                socket.Options.ReceiveHighWatermark = configuration.ZeroMQ.ReceiveHighWaterMark;
+            }
+
+            //
+            if (configuration.ZeroMQ.IsServer)
+            {
+                socket.Bind(configuration.Url);
+            }
+            else
+            {
+                socket.Connect(configuration.Url);
+            }
+        }
+        #endregion
+    }
+}

Providers/XZeroMQProducerService.cs

@@ -0,0 +1,198 @@
+using NetMQ;
+using System;
+using System.Text;
+using NetMQ.Sockets;
+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 Microsoft.Extensions.Logging;
+
+namespace xEventService.Providers
+{
+    public class XZeroMQProducerService : XBaseEventProducerService, IXZeroMQProducerService
+    {
+        //
+        private NetMQSocket socket;
+        private readonly object socketLock = new();
+
+        public XZeroMQProducerService(
+            XEventServiceConfiguration configuration,
+            ILogger<XZeroMQProducerService> logger
+        ) : base(configuration, logger)
+        {
+            //
+            // Validate Configurations ...
+            if (!configuration.IsValid() ||
+                !configuration.ZeroMQ.IsValid() ||
+                configuration.Broker != XEventBroker.XZeroMQ
+            )
+            {
+                XException.InvalidConfiguration.Throw();
+            }
+
+            //
+            InitializeSocket();
+        }
+
+        /// <summary>
+        /// Sending Message with Full Request Descriptor ...
+        /// </summary>
+        /// <param name="request"></param>
+        /// <param name="cancellationToken"></param>
+        /// <returns></returns>
+        public async Task<bool> SendMessageAsync(
+            XEventRequest request,
+            CancellationToken cancellationToken = default
+        )
+        {
+            //
+            var result = request.IsValid();
+            if (!result)
+            {
+                XException.InvalidArgs.Throw();
+            }
+
+            //
+            result = await SendMessageAsync(
+                message: request.Message,
+                topic: request.Topic,
+                cancellationToken: cancellationToken
+            );
+
+            //
+            return result;
+        }
+
+        /// <summary>
+        /// Sending a Message to Topic ...
+        /// </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
+        )
+        {
+            //
+            var result = message.IsValid();
+            if (!result)
+            {
+                XException.InvalidArgs.Throw();
+            }
+
+            //
+            // Normalize Topic ...
+            if (topic.IsNullOrEmpty())
+            {
+                topic = XEventServiceConstants.XDefaultTopic;
+            }
+
+            //
+            try
+            {
+                //
+                var json = message.ToJSON(camelCase: true);
+                var body = Encoding.UTF8.GetBytes(json);
+                await Task.Run(() =>
+                {
+                    //
+                    lock (socketLock)
+                    {
+                        //
+                        if (configuration.ZeroMQ.Pattern == XZeroMQPattern.PubSub)
+                        {
+                            socket.SendMoreFrame(topic).SendFrame(body);
+                        }
+                        else
+                        {
+                            socket.SendFrame(body);
+                        }
+                    }
+                }, cancellationToken);
+
+                //
+                result = true;
+            }
+            catch (Exception)
+            {
+                result = false;
+            }
+
+            //
+            return result;
+        }
+
+        /// <summary>
+        /// Dispose Object ...
+        /// </summary>
+        public void Dispose()
+        {
+            //
+            try
+            {
+                //
+                lock (socketLock)
+                {
+                    //
+                    if (socket != null)
+                    {
+                        //
+                        socket.Close();
+                        socket.Dispose();
+                        socket = null;
+                    }
+                }
+            }
+            catch { }
+
+            //
+            GC.SuppressFinalize(this);
+        }
+
+        //
+        #region Private ...
+        /// <summary>
+        /// Initialize Socket ...
+        /// </summary>
+        private void InitializeSocket()
+        {
+            //
+            lock (socketLock)
+            {
+                //
+                if (configuration.ZeroMQ.Pattern == XZeroMQPattern.PubSub)
+                {
+                    //
+                    socket = new PublisherSocket();
+                    socket.Options.SendHighWatermark = configuration.ZeroMQ.SendHighWaterMark;
+                }
+                else
+                {
+                    //
+                    socket = new PushSocket();
+                    socket.Options.SendHighWatermark = configuration.ZeroMQ.SendHighWaterMark;
+                }
+
+                //
+                if (configuration.ZeroMQ.IsServer)
+                {
+                    socket.Bind(configuration.Url);
+                }
+                else
+                {
+                    socket.Connect(configuration.Url);
+                }
+            }
+        }
+        #endregion
+    }
+}

nuget.config

@@ -3,6 +3,6 @@
   <packageSources>
     <!-- <add key="megan" value="https://hub.megan.ir/nuget/index.json" /> -->
     <add key="nuget" value="https://api.nuget.org/v3/index.json" protocolVersion="3" />
-    <add key="baget" value="https://nuget.saherelmhub.ir/v3/index.json" protocolVersion="3" disableTLSCertificateValidation="true" />
+    <!-- <add key="baget" value="https://nuget.saherelmhub.ir/v3/index.json" protocolVersion="3" disableTLSCertificateValidation="true" /> -->
   </packageSources>
 </configuration>

xEventService.csproj

@@ -35,14 +35,16 @@
 
   <!-- Dependencies -->
   <ItemGroup>
+    <PackageReference Include="NetMQ" Version="4.0.4.3" />
     <PackageReference Include="RabbitMQ.Client" Version="6.2.2" />
+    <PackageReference Include="Confluent.Kafka" Version="2.14.2" />
   </ItemGroup>
 
   <!-- For XML Documentation Support -->
   <PropertyGroup>
-    <CopyLocalLockFileAssemblies>true</CopyLocalLockFileAssemblies>
-    <GenerateDocumentationFile>true</GenerateDocumentationFile>
     <NoWarn>$(NoWarn);1591</NoWarn>
+    <GenerateDocumentationFile>true</GenerateDocumentationFile>
+    <CopyLocalLockFileAssemblies>true</CopyLocalLockFileAssemblies>
   </PropertyGroup>
 
 </Project>