### FILE: C:\Users\SaherElm\Documents\Projects\xSaherelmWorkspace\Modules\xEventService\Configuration\XEventServiceConfiguration.cs

using xEventService.Constants;
using xEventService.Models;

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
    {
        /// <summary>
        /// Message Broker Address ...
        /// </summary>
        public string Url { get; set; }

        /// <summary>
        /// Max Retries to Sending Message ...
        /// </summary>
        public int MaxRetries { get; set; } = 3;

        /// <summary>
        /// Allowed Message Broker ...
        /// </summary>
        public XEventBroker Broker { get; set; } = XEventBroker.None;
        
        /// <summary>
        /// ZeroMQ Specific Options ...
        /// Only used when Broker is XZeroMQ ...
        /// </summary>
        public XZeroMQOptions ZeroMQ { get; set; } = new XZeroMQOptions();        
        
        /// <summary>
        /// RabbitMQ Specific Options ...
        /// Only used when Broker is XRabbitMQ ...
        /// </summary>
        public XRabbitMQOptions RabbitMQ { get; set; } = new XRabbitMQOptions();        
    }
}

### FILE: C:\Users\SaherElm\Documents\Projects\xSaherelmWorkspace\Modules\xEventService\Constants\ConfigurationNodeNames.cs

namespace xEventService.Constants
{
    public partial struct ConfigurationNodeNames {
        public const string EVENT_SERVICE_NODE_NAME = "EventServiceConfiguration";
    }
}

### FILE: C:\Users\SaherElm\Documents\Projects\xSaherelmWorkspace\Modules\xEventService\Constants\XEventBroker.cs

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,
    }
}

### FILE: C:\Users\SaherElm\Documents\Projects\xSaherelmWorkspace\Modules\xEventService\Constants\XEventServiceConstants.cs

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 = "*";
    }
}

### FILE: C:\Users\SaherElm\Documents\Projects\xSaherelmWorkspace\Modules\xEventService\Constants\XRabbitMQExchangeType.cs

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,
    }
}

### FILE: C:\Users\SaherElm\Documents\Projects\xSaherelmWorkspace\Modules\xEventService\Constants\XZeroMQPattern.cs

using xExceptions.Attributes;

namespace xEventService.Constants
{
    public enum XZeroMQPattern
    {
        [StringValue(XEventServiceConstants.None)]
        None,

        [StringValue(XEventServiceConstants.PubSub)]
        PubSub,

        [StringValue(XEventServiceConstants.PushPull)]
        PushPull
    }
}

### FILE: C:\Users\SaherElm\Documents\Projects\xSaherelmWorkspace\Modules\xEventService\DI\XDIHelperExtension.cs

using System;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
using xCommons.Extensions;
using xEventService.Configuration;
using xEventService.Constants;
using xEventService.Extensions;
using xEventService.Interfaces;
using xEventService.Providers;
using xExceptions.Constants;

namespace xEventService.DI
{
    public static partial class XDIHelperExtension
    {
        /// <summary>
        /// Extract EventService Configurations ...
        /// </summary>
        /// <param name="source"></param>
        /// <returns></returns>
        public static XEventServiceConfiguration GetXEventServiceConfiguration(
            this IConfiguration source
        )
        {
            //
            var configSection = source
                .GetSection(ConfigurationNodeNames.EVENT_SERVICE_NODE_NAME);
            var result = configSection.Get<XEventServiceConfiguration>();

            //
            return result;
        }

        /// <summary>
        /// Register Event Service Configurations ...
        /// </summary>
        /// <param name="services"></param>
        /// <param name="configuration"></param>
        public static void AddXEventServiceConfiguration(
            this IServiceCollection services,
            IConfiguration configuration
        )
        {
            //
            var config = configuration.GetXEventServiceConfiguration();
            services.AddXEventServiceConfiguration(config);
        }

        /// <summary>
        /// Register Event Service Configurations ...
        /// </summary>
        /// <param name="services"></param>
        /// <param name="configuration"></param>
        public static void AddXEventServiceConfiguration(
            this IServiceCollection services,
            XEventServiceConfiguration configuration
        )
        {
            //
            // Validate ...
            var isValid = configuration.IsValid() &&
                (configuration.Broker == XEventBroker.XKafka ||
                (configuration.Broker == XEventBroker.XZeroMQ &&
                configuration.ZeroMQ.IsValid()) ||
                (configuration.Broker == XEventBroker.XRabbitMQ &&
                configuration.RabbitMQ.IsValid()));
            if (!isValid)
            {
                //
                Log("Registration Failed, due Invalid Configuration ...");
                XException.InvalidConfiguration.Throw();
            }

            //
            // Register Configuration as Singleton ...
            services.AddSingleton(configuration);
        }

        /// <summary>
        /// Register Service ...
        /// </summary>
        /// <param name="source"></param>
        /// <param name="configuration"></param>
        /// <param name="lifeTime"></param>
        public static void AddXEventService(
            this IServiceCollection source,
            IConfiguration configuration,
            ServiceLifetime lifeTime = ServiceLifetime.Scoped
        )
        {
            //
            var config = configuration.GetXEventServiceConfiguration();
            source.AddEventService(
                lifeTime: lifeTime,
                configuration: config
            );
        }

        /// <summary>
        /// Register Service ...
        /// </summary>
        /// <param name="source"></param>
        /// <param name="configuration"></param>
        /// <param name="lifeTime"></param>
        public static void AddXEventService(
            this IServiceCollection source,
            XEventServiceConfiguration configuration,
            ServiceLifetime lifeTime = ServiceLifetime.Scoped
        )
        {
            //
            source.AddEventService(
                lifeTime: lifeTime,
                configuration: configuration
            );
        }

        /// <summary>
        /// Event Consumer Service Registration ... 
        /// </summary>
        /// <typeparam name="TConsumer"></typeparam>
        /// <param name="source"></param>
        public static void AddXEventServiceConsumer<TConsumer>(
           this IServiceCollection source
        ) where TConsumer : XBaseEventConsumerBackgroundService
        {
            //
            var config = source.GetRegisteredService<XEventServiceConfiguration>();
            if (config.IsNullOrDefault())
            {
                //
                Log("Consumer Registration Faild, due Invalid Configuration issue ...");
                XException.InvalidConfiguration.Throw();
            }

            //
            var consumerType = typeof(TConsumer);
            switch (config.Broker)
            {
                //
                case XEventBroker.XKafka:
                    //
                    if (!typeof(XKafkaConsumerServiceBase).IsAssignableFrom(consumerType))
                    {
                        //
                        Log($"Consumer Registration Failed. TConsumer must inherit from XKafkaConsumerServiceBase when Broker is {config.Broker.GetStringValue()} ...");
                        XException.InvalidConfiguration.Throw();
                    }
                    break;

                //
                case XEventBroker.XZeroMQ:
                    if (!typeof(XZeroMQConsumerServiceBase).IsAssignableFrom(consumerType))
                    {
                        Log($"Consumer Registration Failed. TConsumer must inherit from XZeroMQConsumerServiceBase when Broker is {config.Broker.GetStringValue()} ...");
                        XException.InvalidConfiguration.Throw();
                    }
                    break;

                //
                case XEventBroker.XRabbitMQ:
                    if (!typeof(XRabbitMQConsumerServiceBase).IsAssignableFrom(consumerType))
                    {
                        Log($"Consumer Registration Failed. TConsumer must inherit from XRabbitMQConsumerServiceBase when Broker is {config.Broker.GetStringValue()} ...");
                        XException.InvalidConfiguration.Throw();
                    }
                    break;
            }

            //
            source.AddXBackgroundService<TConsumer>();

            //
            Log($"Consumer Registered Successfully: {consumerType.Name} for Broker: {config.Broker.GetStringValue()} ...");
        }

        //
        #region Private ...
        /// <summary>
        /// a LogTag for Service ...
        /// </summary>
        private static string XLogTag = XEventServiceConstants.XEventServiceDILogTag;

        /// <summary>
        /// print a log in Console ...
        /// </summary>
        /// <param name="message"></param>
        private static void Log(string message)
        {
            Console.WriteLine($"{XLogTag} => {message}");
        }

        /// <summary>
        /// Register Event Service ...
        /// </summary>
        /// <param name="source"></param>
        /// <param name="configuration"></param>
        /// <param name="lifeTime"></param>
        private static void AddEventService(
            this IServiceCollection source,
            XEventServiceConfiguration configuration,
            ServiceLifetime lifeTime = ServiceLifetime.Scoped
        )
        {
            //
            // Register Configuration ...
            source.AddXEventServiceConfiguration(configuration);

            //
            // Register based on Broker ...
            switch (configuration.Broker)
            {
                //
                case XEventBroker.XKafka:
                    source.Add(
                        new ServiceDescriptor(typeof(IXKafkaProducerService), typeof(XKafkaProducerService), lifeTime)
                    );
                    break;

                //
                case XEventBroker.XZeroMQ:
                    source.Add(
                        new ServiceDescriptor(typeof(IXZeroMQProducerService), typeof(XZeroMQProducerService), lifeTime)
                    );
                    break;

                //
                case XEventBroker.XRabbitMQ:
                    source.Add(
                        new ServiceDescriptor(typeof(IXRabbitMQProducerService), typeof(XRabbitMQProducerService), lifeTime)
                    );
                    break;
            }

            //
            source.Add(
                new ServiceDescriptor(typeof(IXEventServiceProvider), typeof(XEventServiceProvider), lifeTime)
            );

            //
            Log($"Service Registered Sucessfull, Provider: {configuration.Broker.GetStringValue()}");
        }
        #endregion
    }
}

### FILE: C:\Users\SaherElm\Documents\Projects\xSaherelmWorkspace\Modules\xEventService\Extensions\XEventServiceExtensions.cs

using System.Linq;
using xCommons.Extensions;
using xEventService.Configuration;
using xEventService.Constants;
using xEventService.Models;
using xExceptions.Constants;

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;
        }
    }
}

### FILE: C:\Users\SaherElm\Documents\Projects\xSaherelmWorkspace\Modules\xEventService\Interfaces\IXEventServiceProvider.cs

namespace xEventService.Interfaces
{
    public interface IXEventServiceProvider : IXProducerServiceBase
    { }
}

### FILE: C:\Users\SaherElm\Documents\Projects\xSaherelmWorkspace\Modules\xEventService\Interfaces\IXKafkaProducerService.cs

namespace xEventService.Interfaces
{
    /// <summary>
    /// a Service for Produce Kafka Messages ...
    /// </summary>
    public interface IXKafkaProducerService : IXProducerServiceBase
    { }
}

### FILE: C:\Users\SaherElm\Documents\Projects\xSaherelmWorkspace\Modules\xEventService\Interfaces\IXProducerServiceBase.cs

using System;
using System.Threading;
using System.Threading.Tasks;
using xEventService.Constants;
using xEventService.Models;

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
        );
    }
}

### FILE: C:\Users\SaherElm\Documents\Projects\xSaherelmWorkspace\Modules\xEventService\Interfaces\IXRabbitMQProducerService.cs

using System.Threading;
using System.Threading.Tasks;
using xEventService.Models;

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
        );
    }
}

### FILE: C:\Users\SaherElm\Documents\Projects\xSaherelmWorkspace\Modules\xEventService\Interfaces\IXZeroMQProducerService.cs

namespace xEventService.Interfaces
{
    public interface IXZeroMQProducerService : IXProducerServiceBase
    { }
}

### FILE: C:\Users\SaherElm\Documents\Projects\xSaherelmWorkspace\Modules\xEventService\Models\XEventMessage.cs

using System;
using System.Collections.Generic;
using Newtonsoft.Json;

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>();
    }
}

### FILE: C:\Users\SaherElm\Documents\Projects\xSaherelmWorkspace\Modules\xEventService\Models\XEventRequest.cs

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; }
    }
}

### FILE: C:\Users\SaherElm\Documents\Projects\xSaherelmWorkspace\Modules\xEventService\Models\XRabbitMQOptions.cs

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;
    }
}

### FILE: C:\Users\SaherElm\Documents\Projects\xSaherelmWorkspace\Modules\xEventService\Models\XZeroMQOptions.cs

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;
    }
}

### FILE: C:\Users\SaherElm\Documents\Projects\xSaherelmWorkspace\Modules\xEventService\Providers\XBaseEventConsumerBackgroundService.cs

using System;
using Microsoft.Extensions.Logging;
using xCommons.Extensions;
using xCommons.Providers;
using xEventService.Configuration;
using xEventService.Constants;
using xEventService.Extensions;
using xExceptions.Constants;

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;
        }
    }
}

### FILE: C:\Users\SaherElm\Documents\Projects\xSaherelmWorkspace\Modules\xEventService\Providers\XBaseEventProducerService.cs

using Microsoft.Extensions.Logging;
using xCommons.Extensions;
using xEventService.Configuration;
using xEventService.Extensions;
using xExceptions.Constants;

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();
            }
        }        
    }
}

### FILE: C:\Users\SaherElm\Documents\Projects\xSaherelmWorkspace\Modules\xEventService\Providers\XEventServiceProvider.cs

using System.Threading;
using System.Threading.Tasks;
using xCommons.Extensions;
using xEventService.Configuration;
using xEventService.Constants;
using xEventService.Extensions;
using xEventService.Interfaces;
using xEventService.Models;
using xExceptions.Constants;

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();
            }
        }
    }
}

### FILE: C:\Users\SaherElm\Documents\Projects\xSaherelmWorkspace\Modules\xEventService\Providers\XKafkaConsumerServiceBase.cs

using System;
using System.Threading;
using System.Threading.Tasks;
using Confluent.Kafka;
using Microsoft.Extensions.Logging;
using xCommons.Extensions;
using xEventService.Configuration;
using xEventService.Constants;
using xEventService.Extensions;
using xEventService.Models;
using xExceptions.Constants;

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);

            //
            // 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)
                {
                    await Task.Delay(1000, cancellationToken);
                }
                catch (Exception)
                {
                    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();
        }
    }
}

### FILE: C:\Users\SaherElm\Documents\Projects\xSaherelmWorkspace\Modules\xEventService\Providers\XKafkaProducerService.cs

using xCommons.Extensions;
using xExceptions.Constants;
using xEventService.Extensions;
using xEventService.Interfaces;
using xEventService.Configuration;
using xEventService.Models;
using Confluent.Kafka;
using Microsoft.Extensions.Logging;
using System.Threading.Tasks;
using System.Threading;
using xEventService.Constants;
using System;

namespace xEventService.Providers
{
    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(
                message: request.Message,
                topic: request.Topic,
                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
            {
                result = false;
            }

            //
            return result;
        }

        /// <summary>
        /// Dispose Required Objects ...
        /// </summary>
        public void Dispose()
        {
            //
            if (producer != null)
            {
                producer.Dispose();
            }

            //
            GC.SuppressFinalize(this);
        }
    }
}

### FILE: C:\Users\SaherElm\Documents\Projects\xSaherelmWorkspace\Modules\xEventService\Providers\XRabbitMQConsumerServiceBase.cs

using System;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.Logging;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using xCommons.Extensions;
using xEventService.Configuration;
using xEventService.Constants;
using xEventService.Extensions;
using xEventService.Models;
using xExceptions.Constants;

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();

            //
            // 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
                {
                    //
                    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 digits = Guid.NewGuid().ToString().GetDigits();
                result = $"{configuration.RabbitMQ.Exchange}[{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
    }
}

### FILE: C:\Users\SaherElm\Documents\Projects\xSaherelmWorkspace\Modules\xEventService\Providers\XRabbitMQProducerService.cs

using System;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.Logging;
using RabbitMQ.Client;
using xCommons.Extensions;
using xEventService.Configuration;
using xEventService.Constants;
using xEventService.Extensions;
using xEventService.Interfaces;
using xEventService.Models;
using xExceptions.Constants;

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
            {
                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
    }
}

### FILE: C:\Users\SaherElm\Documents\Projects\xSaherelmWorkspace\Modules\xEventService\Providers\XZeroMQConsumerServiceBase.cs

using System;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.Logging;
using NetMQ;
using NetMQ.Sockets;
using xCommons.Extensions;
using xEventService.Configuration;
using xEventService.Constants;
using xEventService.Extensions;
using xEventService.Models;
using xExceptions.Constants;

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();
            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
                    {
                        if (socket.TryReceiveFrameBytes(TimeSpan.FromMilliseconds(100), out body))
                        {
                            // Process
                        }
                    }

                    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)
                {
                    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
    }
}

### FILE: C:\Users\SaherElm\Documents\Projects\xSaherelmWorkspace\Modules\xEventService\Providers\XZeroMQProducerService.cs

using System;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.Logging;
using NetMQ;
using NetMQ.Sockets;
using xCommons.Extensions;
using xEventService.Configuration;
using xEventService.Constants;
using xEventService.Extensions;
using xEventService.Interfaces;
using xEventService.Models;
using xExceptions.Constants;

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
    }
}