add supports for ZeroMQ, XRabbitMQ, XKafka ...

Configuration/XEventServiceConfiguration.cs

@@ -23,7 +23,13 @@ namespace xEventService.Configuration
         /// 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 ...

Interfaces/IXRabbitMQProducerService.cs

@@ -1,4 +1,3 @@
-using System;
 using System.Threading;
 using System.Threading.Tasks;
 using xEventService.Models;
@@ -29,6 +28,6 @@ namespace xEventService.Interfaces
             string exchange = null,
             string routingKey = null,
             CancellationToken cancellationToken = default
-        );        
+        );
     }
 }

Interfaces/IXZeroMQProducerService.cs

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

Providers/XBaseEventConsumerBackgroundService.cs

@@ -22,12 +22,12 @@ namespace xEventService.Providers
         /// <summary>
         /// Allowed Senders ...
         /// </summary>
-        public readonly string[] allowedSenders;
+        protected readonly string[] allowedSenders;
 
         /// <summary>
         /// Allowed Actions ...
         /// </summary>
-        public readonly string[] allowedActions;
+        protected readonly string[] allowedActions;
 
         /// <summary>
         /// Configuration ...
@@ -41,10 +41,7 @@ namespace xEventService.Providers
             XEventServiceConfiguration configuration,
             ILogger<XResumableBackgroundService> logger,
             string topic = XEventServiceConstants.XDefaultTopic
-        ) : base(
+        ) : base(serviceProvider, logger)
-            logger: logger,
-            serviceProvider: serviceProvider
-        )
         {
             //
             // Validate Actions ...

Providers/XBaseEventProducerService.cs

@@ -0,0 +1,31 @@
+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();
+            }
+        }        
+    }
+}

Providers/XKafkaConsumerServiceBase.cs

@@ -6,6 +6,8 @@ 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
@@ -87,10 +89,21 @@ namespace xEventService.Providers
                     )
                     {
                         //
-                        await DoWorkAsync(
+                        var message = consumeResult
-                            payload: consumeResult.Message,
+                            .Message
-                            cancellationToken: cancellationToken
+                            .Value
-                        );
+                            .FromJSON<XEventMessage>();
+                        if (message.IsValid(
+                            allowedSenders: allowedSenders,
+                            allowedActions: allowedActions)
+                        )
+                        {
+                            //
+                            await DoWorkAsync(
+                                payload: consumeResult.Message,
+                                cancellationToken: cancellationToken
+                            );
+                        }
                     }
                 }
                 catch (ConsumeException)
@@ -105,13 +118,18 @@ namespace xEventService.Providers
         }
 
         /// <summary>
-        /// Dispose Implementation ...
+        /// Dispose Object ...
         /// </summary>
         public override void Dispose()
         {
             //
-            consumer.Close();
+            if (consumer != null)
-            consumer.Dispose();
+            {
+                consumer.Close();
+                consumer.Dispose();
+            }
+
+            //
             GC.SuppressFinalize(this);
             base.Dispose();
         }

Providers/XKafkaProducerService.cs

@@ -13,21 +13,15 @@ using System;
 
 namespace xEventService.Providers
 {
-    public class XKafkaProducerService : IXKafkaProducerService
+    public class XKafkaProducerService : XBaseEventProducerService, IXKafkaProducerService
     {
         private readonly IProducer<Null, string> producer;
-        private readonly ILogger<XKafkaProducerService> logger;
-        private readonly XEventServiceConfiguration configuration;
 
         public XKafkaProducerService(
             ILogger<XKafkaProducerService> logger,
             XEventServiceConfiguration configuration
-        )
+        ) : base(configuration, logger)
         {
-            //
-            this.logger = logger;
-            this.configuration = configuration;
-
             //
             // Validate Configurations ...
             if (!configuration.IsValid() ||
@@ -132,7 +126,12 @@ namespace xEventService.Providers
         public void Dispose()
         {
             //
-            producer.Dispose();
+            if (producer != null)
+            {
+                producer.Dispose();
+            }
+
+            //
             GC.SuppressFinalize(this);
         }
     }

Providers/XRabbitMQConsumerServiceBase.cs

@@ -19,6 +19,7 @@ namespace xEventService.Providers
     /// </summary>
     public abstract class XRabbitMQConsumerServiceBase : XBaseEventConsumerBackgroundService
     {
+        //
         protected IModel channel;
         protected IConnection connection;
         protected AsyncEventingBasicConsumer consumer;
@@ -79,7 +80,11 @@ namespace xEventService.Providers
                     var body = ea.Body.ToArray();
                     var json = Encoding.UTF8.GetString(body);
                     var message = json.FromJSON<XEventMessage>();
-                    if (!message.IsNullOrDefault())
+                    if (!message.IsNullOrDefault() &&
+                        message.IsValid(
+                        allowedSenders: allowedSenders,
+                        allowedActions: allowedActions)
+                    )
                     {
                         //
                         await DoWorkAsync(
@@ -163,8 +168,18 @@ namespace xEventService.Providers
         public override void Dispose()
         {
             //
-            channel.Dispose();
+            if (channel != null)
-            connection.Dispose();
+            {
+                channel.Dispose();
+            }
+
+            //
+            if (connection != null)
+            {
+                connection.Dispose();
+            }
+
+            //
             GC.SuppressFinalize(this);
             base.Dispose();
         }

Providers/XRabbitMQProducerService.cs

@@ -17,23 +17,17 @@ namespace xEventService.Providers
     /// <summary>
     /// RabbitMQ Producer Service Implementation ...
     /// </summary>
-    public class XRabbitMQProducerService : IXRabbitMQProducerService
+    public class XRabbitMQProducerService : XBaseEventProducerService, IXRabbitMQProducerService
     {
         private IModel channel;
         private IConnection connection;
         private readonly object connectionLock = new();
-        private readonly ILogger<XRabbitMQProducerService> logger;
-        private readonly XEventServiceConfiguration configuration;
 
         public XRabbitMQProducerService(
             ILogger<XRabbitMQProducerService> logger,
             XEventServiceConfiguration configuration
-        )
+        ) : base(configuration, logger)
         {
-            //
-            this.logger = logger;
-            this.configuration = configuration;
-
             //
             // Validate Configurations ...
             if (!configuration.IsValid() ||
@@ -52,6 +46,9 @@ namespace xEventService.Providers
         /// <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
@@ -77,9 +74,41 @@ namespace xEventService.Providers
             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,
@@ -114,7 +143,7 @@ namespace xEventService.Providers
             {
                 //
                 // Ensure Connection is Alive ...
-                await EnsureChannelAsync(cancellationToken);
+                EnsureChannelAsync();
 
                 //
                 // Serialize Message ...
@@ -170,14 +199,22 @@ namespace xEventService.Providers
             try
             {
                 //
-                channel.Close();
+                if (channel != null)
-                channel.Dispose();
+                {
-                channel = null;
+                    //
+                    channel.Close();
+                    channel.Dispose();
+                    channel = null;
+                }
 
                 //
-                connection.Close();
+                if (connection != null)
-                connection.Dispose();
+                {
-                connection = null;
+                    //
+                    connection.Close();
+                    connection.Dispose();
+                    connection = null;
+                }
             }
             catch
             { }
@@ -228,29 +265,10 @@ namespace xEventService.Providers
             return result;
         }
 
-        /// <summary>
-        /// Ensure Connection Established ...
-        /// </summary>
-        private async Task<bool> EnsureConnectionAsync()
-        {
-            //
-            try
-            {
-                await EnsureChannelAsync();
-                return true;
-            }
-            catch
-            {
-                return false;
-            }
-        }
-
         /// <summary>
         /// Ensure Channel is Open and Exchange/Queue Declared ...
         /// </summary>
-        private async Task EnsureChannelAsync(
+        private void EnsureChannelAsync()
-            CancellationToken cancellationToken = default
-        )
         {
             //
             if (IsConnected())

Providers/XZeroMQConsumerServiceBase.cs

@@ -0,0 +1,187 @@
+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.Broker != XEventBroker.XKafka)
+            {
+                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)
+        {
+            //
+            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 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
+    }
+}