commit 9eabf45f0636d24b058a2be4b5a3aa75450159ae Author: Hadi Khazaee Asl Date: Thu Jan 25 04:48:33 2024 +0330 Initial Commit ... diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..1bd9d0b --- /dev/null +++ b/.gitignore @@ -0,0 +1,8 @@ +# +# DotNet ... +bin +obj + +# +# Natural Docs ... +Documentation/* diff --git a/Base/XBaseEvent.cs b/Base/XBaseEvent.cs new file mode 100644 index 0000000..2d6c416 --- /dev/null +++ b/Base/XBaseEvent.cs @@ -0,0 +1,14 @@ +using System; + +namespace xEventService.Base { + public abstract class XBaseEvent { + public string Type { get; protected set; } + public DateTime Timestamp { get; protected set; } + + protected XBaseEvent () { + // + Type = GetType ().Name; + Timestamp = DateTime.UtcNow; + } + } +} \ No newline at end of file diff --git a/Configuration/XEventServiceConfiguration.cs b/Configuration/XEventServiceConfiguration.cs new file mode 100644 index 0000000..a1e822e --- /dev/null +++ b/Configuration/XEventServiceConfiguration.cs @@ -0,0 +1,51 @@ +using System.Collections.Generic; +using RabbitMQ.Client; + +namespace xEventService.Configuration { + /// + /// a configuration class for describing how to connect rabbit mq server + /// and how channels must be declared ... + /// + public class XEventServiceConfiguration { + public string Server { get; set; } + public int Port { get; set; } + public string Username { get; set; } + public string Password { get; set; } + + // + public bool AllowDynamicQueues { get; set; } = true; + public XQueueArgs DefaultQueueArgs { get; set; } = new XQueueArgs (); + public IDictionary Queues { get; set; } + + // + public bool AllowDynamicExchangess { get; set; } = true; + public XExchangeArgs DefaultExchangeArgs { get; set; } = new XExchangeArgs (); + public IDictionary Exchanges { get; set; } + + // + public static XEventServiceConfiguration GetDefaults () { + return new XEventServiceConfiguration { + Port = 5672, + Username = "guest", + Password = "guest", + Server = "localhost", + Queues = new Dictionary { { "XMainQueue", new XQueueArgs { Name = "XMainQueue" } } + } + }; + } + } + + public class XQueueArgs { + public string Name { get; set; } + public bool Durable { get; set; } = false; + public bool Exclusive { get; set; } = false; + public bool AutoDelete { get; set; } = true; + } + + public class XExchangeArgs { + public string Name { get; set; } + public string Type { get; set; } = ExchangeType.Fanout; + public bool Durable { get; set; } = false; + public bool AutoDelete { get; set; } = true; + } +} \ No newline at end of file diff --git a/Constants/ConfigurationNodeNames.cs b/Constants/ConfigurationNodeNames.cs new file mode 100644 index 0000000..9c4fbd0 --- /dev/null +++ b/Constants/ConfigurationNodeNames.cs @@ -0,0 +1,6 @@ +namespace xEventService.Constants +{ + public partial struct ConfigurationNodeNames { + public const string EVENT_SERVICE_NODE_NAME = "EventServiceConfiguration"; + } +} \ No newline at end of file diff --git a/DI/XDIHelperExtension.cs b/DI/XDIHelperExtension.cs new file mode 100644 index 0000000..14b4f84 --- /dev/null +++ b/DI/XDIHelperExtension.cs @@ -0,0 +1,140 @@ +using System; +using Microsoft.AspNetCore.Builder; +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.DependencyInjection; +using xCommons.Extensions; +using xEventService.Configuration; +using xEventService.Constants; +using xEventService.Interfaces; +using xEventService.Models; +using xEventService.Providers; + +namespace xEventService.DI { + public static class XDIHelperExtension { + /// + /// Extract EventService Configurations ... + /// + /// + /// + public static XEventServiceConfiguration GetXEventServiceConfigurations (this IConfiguration source) { + // + var configSection = source + .GetSection (ConfigurationNodeNames.EVENT_SERVICE_NODE_NAME); + var result = configSection.Get (); + + // + return result; + } + + /// + /// Register Event Service Configurations ... + /// + /// + /// + public static void AddXEventServiceConfigurations ( + this IServiceCollection services, + IConfiguration configuration + ) { + // + var config = configuration.GetXEventServiceConfigurations (); + services.AddXEventServiceConfigurations (config); + } + + /// + /// Register Event Service Configurations ... + /// + /// + /// + public static void AddXEventServiceConfigurations ( + this IServiceCollection services, + XEventServiceConfiguration configuration + ) { + // + services.AddSingleton (configuration); + } + + /// + /// Register Event Service ... + /// + /// + /// + public static void AddXEventService ( + this IServiceCollection services, + IConfiguration configuration + ) { + // + // Check Configuration Registered ... + var config = services.GetRegisteredService (); + if (config.IsNull ()) { + // + Console.WriteLine ($"XEventService: there is no provided XEventServiceConfiguration, using default ..."); + + // + // Register Default Configuration ... + services.AddXEventServiceConfigurations (configuration); + } + + // + // Register IXEventBus ... + var serviceBas = services.GetRegisteredService (); + if (serviceBas.IsNull ()) { + services.AddSingleton (); + } + } + + /// + /// Register an Event Handler ... + /// + /// + /// + /// + /// + /// + public static void AddXEventHandler ( + this IServiceCollection services, + ServiceLifetime lifetime = ServiceLifetime.Transient + ) + where TEvent : XEvent + where THandler : IXEventHandler { + // + services.Add (new ServiceDescriptor ( + serviceType: typeof (THandler), + implementationType: typeof (THandler), + lifetime: lifetime + )); + + // + services.Add (new ServiceDescriptor ( + serviceType: typeof (IXEventHandler), + implementationType: typeof (THandler), + lifetime: lifetime + )); + } + + /// + /// Subscribe to specific Event ... + /// + /// + /// + /// + /// + public static void XEventSubscribe ( + this IApplicationBuilder app, + bool toExchange = true + ) + where TEvent : XEvent + where THandler : IXEventHandler { + // + var eventName = typeof (TEvent).Name; + + // + // Retrieve EventBus from IOC ... + var eventBus = app.ApplicationServices.GetRequiredService (); + + // + // Subscribe to Event Handler ... + var isSubscribed = eventBus.Subscribe (); + Console.WriteLine ($"XEventService: Event Handler subscription result: {eventName} => {isSubscribed} ..."); + } + } +} \ No newline at end of file diff --git a/Extensions/ConfigurationExtensions.cs b/Extensions/ConfigurationExtensions.cs new file mode 100644 index 0000000..f557422 --- /dev/null +++ b/Extensions/ConfigurationExtensions.cs @@ -0,0 +1,288 @@ +using System; +using RabbitMQ.Client; +using xCommons.Extensions; +using xEventService.Configuration; +using xEventService.Models; +using xExceptions.Constants; + +namespace xEventService.Extensions { + public static class ConfigurationExtensions { + /// + /// Create Connection Factory Based on + /// provided Configuration ... + /// + /// + /// + public static ConnectionFactory GetConnectionFactory (this XEventServiceConfiguration source) { + // + // Validate Args ... + if (source.IsNull ()) { + XException.InvalidConfiguration.Throw (); + } + + // + // Create Connection Factory ... + var result = new ConnectionFactory { + Port = source.Port, + HostName = source.Server, + UserName = source.Username, + Password = source.Password + }; + + // + // Return Result ... + return result; + } + + /// + /// Create a Connection to Server ... + /// + /// + /// + public static IConnection Connect (this XEventServiceConfiguration source) { + // + // Open Connection ... + IConnection result = null; + try { + // + // Retrieve Connection Factory ... + var connectionFactory = source.GetConnectionFactory (); + + // + // Create Connection ... + result = connectionFactory.CreateConnection (); + } catch (Exception ex) { + // + Console.WriteLine ($"XEventService Exception: Connection Failed, {ex.Message} ..."); + + // + XException.ActionFailed.Throw (); + } + + // + return result; + } + + /// + /// Validate a Queue Name based on Configurations ... + /// + /// + /// + public static void ValidateQueue ( + this XEventServiceConfiguration source, + string queueName + ) { + // + // Validate Args ... + if (source.IsNull ()) { + XException.InvalidArgs.Throw (); + } + + // + // Validate Queues ... + if ((source.Queues.IsNull () && + !source.AllowDynamicQueues) || + (!source.Queues.IsNull () && + !source.Queues.Keys.HasChild () && + !source.AllowDynamicQueues)) { + // + Console.WriteLine ($"XEventService Exception: there isn't any configured Queue ..."); + + // + XException.InvalidArgs.Throw (); + } + + // + // Check queueName not empty ... + if (queueName.IsNullOrEmpty ()) { + // + Console.WriteLine ($"XEventService Exception: Invalid QueueName ..."); + + // + XException.InvalidArgs.Throw (); + } + + // + // Check queueName Exists ... + if (!source.Queues.IsNull () && + !source.Queues.ContainsKey (queueName) && + !source.AllowDynamicQueues) { + // + Console.WriteLine ($"XEventService Exception: QueueName {queueName} not found ..."); + + // + XException.InvalidArgs.Throw (); + } + } + + /// + /// Validate a Exchange Name based on Configurations ... + /// + /// + /// + public static void ValidateExchange ( + this XEventServiceConfiguration source, + string exchangeName + ) { + // + // Validate Args ... + if (source.IsNull ()) { + XException.InvalidArgs.Throw (); + } + + // + // Validate Exchanges ... + if ((source.Exchanges.IsNull () && + !source.AllowDynamicExchangess) || + (!source.Exchanges.IsNull () && + !source.Exchanges.Keys.HasChild () && + !source.AllowDynamicExchangess)) { + // + Console.WriteLine ($"XEventService Exception: there isn't any configured Exchanges ..."); + + // + XException.InvalidArgs.Throw (); + } + + // + // Check exchangeName not empty ... + if (exchangeName.IsNullOrEmpty ()) { + // + Console.WriteLine ($"XEventService Exception: Invalid ExchangeName ..."); + + // + XException.InvalidArgs.Throw (); + } + + // + // Check exchangeName Exists ... + if (!source.Exchanges.IsNull () && + !source.Exchanges.ContainsKey (exchangeName) && + !source.AllowDynamicExchangess) { + // + Console.WriteLine ($"XEventService Exception: ExchangeName {exchangeName} not found ..."); + + // + XException.InvalidArgs.Throw (); + } + } + + /// + /// Retrieve specific QueueArgs ... + /// + /// + /// + /// + public static XQueueArgs GetQueueArgs ( + this XEventServiceConfiguration source, + string queueName + ) { + // + // Validate Queue Name ... + source.ValidateQueue (queueName); + + // + var result = source.Queues + .IsNull () ? null : source + .Queues[queueName]; + if (result.IsNull ()) { + // + // Prepare Default Queue ... + result = source.DefaultQueueArgs; + if (result.IsNull ()) { + result = new XQueueArgs () { + Name = queueName + }; + } else { + result.Name = queueName; + } + } + + // + return result; + } + + /// + /// Retrieve specific ExchangeArgs ... + /// + /// + /// + /// + public static XExchangeArgs GetExchangeArgs ( + this XEventServiceConfiguration source, + string exchangeName + ) { + // + // Validate Exchange Name ... + source.ValidateExchange (exchangeName); + + // + var result = source.Exchanges + .IsNull () ? null : source + .Exchanges[exchangeName]; + if (result.IsNull ()) { + // + // Prepare Default Queue ... + result = source.DefaultExchangeArgs; + if (result.IsNull ()) { + result = new XExchangeArgs () { + Name = exchangeName, + Type = ExchangeType.Fanout + }; + } else { + result.Name = exchangeName; + } + } + + // + return result; + } + + /// + /// Create a Channel ... + /// + /// + /// + /// + public static XChannel CreateChannel ( + this XEventServiceConfiguration source, + string exchangeName = null + ) { + // + // Validate ExchangeName and Retrieve ExchangeArgs if provided ... + var exchangeArgs = exchangeName + .IsNullOrEmpty () ? + null : + source.GetExchangeArgs (exchangeName); + + // + // Create Connection ... + var connection = source.Connect (); + var channel = connection.CreateModel (); + + // + // Create XChannel Model ... + var result = new XChannel ( + exchangeName: exchangeName, + connection: connection + ); + + // + // Declaring Queue if Provided ... + if (!exchangeArgs.IsNull () && + !exchangeName.IsNullOrEmpty ()) { + // + // Declaring Queue ... + result.Channel.ExchangeDeclare ( + exchange: exchangeName, + type: exchangeArgs.Type, + durable: exchangeArgs.Durable, + autoDelete: exchangeArgs.AutoDelete + ); + } + + // + return result; + } + } +} \ No newline at end of file diff --git a/Extensions/XEventModelExtensions.cs b/Extensions/XEventModelExtensions.cs new file mode 100644 index 0000000..1bf903f --- /dev/null +++ b/Extensions/XEventModelExtensions.cs @@ -0,0 +1,44 @@ +using System; +using System.Text; +using xCommons.Extensions; + +namespace xEventService.Extensions { + public static class XEventModelExtensions { + /// + /// Convert an Object to bytes array for publishing + /// + /// + /// + /// + public static byte[] ToBody (this T source) { + // + var json = source + .ToJSON (); + + // + var result = json + .ToBytes (); + + // + return result; + } + + /// + /// Convert Recieved Bytes to Specific Type ... + /// + /// + /// + /// + public static T FromBody (this ReadOnlyMemory source) { + // + var jsonString = Encoding.UTF8 + .GetString ( + source.ToArray () + ); + var result = jsonString.FromJSON (); + + // + return result; + } + } +} \ No newline at end of file diff --git a/Interfaces/IXEventBus.cs b/Interfaces/IXEventBus.cs new file mode 100644 index 0000000..547b946 --- /dev/null +++ b/Interfaces/IXEventBus.cs @@ -0,0 +1,11 @@ +using xEventService.Models; + +namespace xEventService.Interfaces { + public interface IXEventBus { + bool Publish (T @event) where T : XEvent; + + bool Subscribe () + where TEvent : XEvent + where THandler : IXEventHandler; + } +} \ No newline at end of file diff --git a/Interfaces/IXEventHandler.cs b/Interfaces/IXEventHandler.cs new file mode 100644 index 0000000..63ec78c --- /dev/null +++ b/Interfaces/IXEventHandler.cs @@ -0,0 +1,11 @@ +using System.Threading.Tasks; +using xEventService.Models; + +namespace xEventService.Interfaces { + public interface IXEventHandler : IXEventHandler + where TEvent : XEvent { + Task HandleAsync (TEvent @event); + } + + public interface IXEventHandler { } +} \ No newline at end of file diff --git a/Models/XChannel.cs b/Models/XChannel.cs new file mode 100644 index 0000000..610f0f8 --- /dev/null +++ b/Models/XChannel.cs @@ -0,0 +1,224 @@ +using System; +using RabbitMQ.Client; +using RabbitMQ.Client.Events; +using xCommons.Extensions; +using xEventService.Configuration; +using xEventService.Extensions; +using xExceptions.Constants; + +namespace xEventService.Models { + public class XChannel { + // + public IModel Channel { get; protected set; } + public string QueueName { get; protected set; } + public string ExchangeName { get; protected set; } + public IConnection Connection { get; protected set; } + + // + #region Constructors ... + public XChannel ( + string exchangeName = null, + IConnection connection = null, + IModel channel = null, + EventingBasicConsumer consumer = null + ) { + // + // Validate Connection ... + if (connection.IsNull ()) { + XException.InvalidArgs.Throw (); + } + Connection = connection; + + // + // Validate Channel ... + if (channel.IsNull ()) { + channel = connection.CreateModel (); + } + Channel = channel; + + // + // Check Queue Name and it's Declaration ... + if (!exchangeName.IsNullOrEmpty ()) { + // + ExchangeName = exchangeName; + } + } + + public XChannel ( + XEventServiceConfiguration configuration, + bool dispatchConsumersAsync = true, + string exchangeName = null + ) { + // + // Generate Factory ... + var factory = configuration.GetConnectionFactory (); + if (dispatchConsumersAsync) { + factory.DispatchConsumersAsync = true; + } + + // + // Create Cahnnel ... + Connection = factory.CreateConnection (); + Channel = Connection.CreateModel (); + + // + // Check Exchange Declaration if provided ... + if (!exchangeName.IsNullOrEmpty ()) { + // + ExchangeName = exchangeName; + var exchangeArgs = configuration.GetExchangeArgs (ExchangeName); + + // + // Declaring Exchange ... + Channel.ExchangeDeclare ( + exchange: exchangeName, + type: exchangeArgs.Type, + durable: exchangeArgs.Durable, + autoDelete: exchangeArgs.AutoDelete + ); + + // + // Create a Queue ... + var queueName = $"{exchangeName}[{Guid.NewGuid().ToString().GetDigits()}]"; + var queueArgs = configuration.GetQueueArgs (queueName); + QueueName = queueName; + Channel.QueueDeclare ( + queue: queueName, + durable: queueArgs.Durable, + exclusive: queueArgs.Exclusive, + autoDelete: queueArgs.AutoDelete + ); + + // + // Bind Queue to Exchange ... + Channel.QueueBind ( + queue: queueName, + exchange: exchangeName, + routingKey: "" + ); + } + } + #endregion + + // + #region Actions ... + /// + /// Close Channel ... + /// + public void CloseChannel () { + // + // Check Channel Exists ... + if (!this.Channel.IsNull ()) { + // + // Close Channel if it's Open ... + if (this.Channel.IsOpen) { + this.Channel.Close (); + } + + // + this.Channel.Dispose (); + this.Channel = null; + } + } + + /// + /// Close Connection ... + /// + public void CloseConnection () { + // + // Check Connection Exists ... + if (!this.Connection.IsNull ()) { + // + // Close Connection if it's Open ... + if (this.Connection.IsOpen) { + this.Connection.Close (); + } + + // + this.Connection.Dispose (); + this.Connection = null; + } + } + + /// + /// Publish a Message through Channel on Declare Queue ... + /// + /// + /// + /// + public bool Publish (T message) { + // + // Validate Args ... + if (this.Connection.IsNull () || + !this.Connection.IsOpen || + this.Channel.IsNull () || + !this.Channel.IsOpen) { + // + Console.WriteLine ($"XEventService Exception: Publish Failed ..."); + + // + XException.InvalidArgs.Throw (); + } + + // + // Check Publish Method ... + // + // Check Exchange Name ... + if (this.ExchangeName.IsNullOrEmpty ()) { + // + Console.WriteLine ($"XEventService Exception: Publish Failed, Invalid Exchange Name {ExchangeName} ..."); + + // + XException.InvalidArgs.Throw (); + } + + // + try { + // + // Publish Message through Channel to Declared Queue ... + var body = message.ToBody (); + this.Channel.BasicPublish ( + exchange: ExchangeName, + routingKey: "", + basicProperties : null, + body : body + ); + + // + return true; + } catch (Exception ex) { + // + Console.WriteLine ($"XEventService Exception: Publish Failed, {ex.Message} ..."); + + // + return false; + } + } + + public void Consume (EventingBasicConsumer consumer) { + Channel.BasicConsume ( + queue: QueueName, + autoAck: true, + consumer: consumer + ); + } + + public void ConsumeAsync (AsyncEventingBasicConsumer consumer) { + Channel.BasicConsume ( + queue: QueueName, + autoAck: true, + consumer: consumer + ); + } + + /// + /// Dispose XChannel ... + /// + public void Dispose () { + // + this.CloseChannel (); + this.CloseConnection (); + } + #endregion + } +} \ No newline at end of file diff --git a/Models/XEvent.cs b/Models/XEvent.cs new file mode 100644 index 0000000..2fdde49 --- /dev/null +++ b/Models/XEvent.cs @@ -0,0 +1,5 @@ +using xEventService.Base; + +namespace xEventService.Models { + public class XEvent : XBaseEvent { } +} \ No newline at end of file diff --git a/Providers/XEventBus.cs b/Providers/XEventBus.cs new file mode 100644 index 0000000..c0a9334 --- /dev/null +++ b/Providers/XEventBus.cs @@ -0,0 +1,260 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Text; +using System.Threading.Tasks; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; +using RabbitMQ.Client.Events; +using xCommons.Extensions; +using xEventService.Configuration; +using xEventService.Extensions; +using xEventService.Interfaces; +using xEventService.Models; +using xExceptions.Constants; + +namespace xEventService.Providers { + public sealed class XEventBus : IXEventBus { + // + #region Properties ... + private readonly List events; + private readonly ILogger logger; + private readonly IServiceScopeFactory serviceScopeFactory; + private readonly XEventServiceConfiguration configuration; + private readonly IDictionary> handlers; + #endregion + + // + #region Constructors ... + public XEventBus ( + ILoggerFactory loggerFactory, + IServiceScopeFactory serviceScopeFactory, + XEventServiceConfiguration configuration + ) { + // + this.configuration = configuration; + this.serviceScopeFactory = serviceScopeFactory; + + // + events = new List (); + handlers = new Dictionary> (); + logger = loggerFactory.CreateLogger (); + } + #endregion + + // + #region Abstracts ... + public bool Publish (T @event) where T : XEvent { + // + // Validate Args ... + if (@event.IsNull ()) { + return false; + } + + // + // Prepare Channel ... + XChannel channel = null; + + // + // Extract event name ... + var exchangeName = @event.GetType ().Name; + + // + // Do Action ... + try { + // + // Get XChannel Model ... + channel = configuration.CreateChannel (exchangeName); + + // + // Publish Command ... + channel.Publish (message: @event); + + // + // Return Result ... + return true; + } catch (Exception ex) { + // + logger.LogError ($"XEventService Exception: Publishing Failed {exchangeName}/{ex.Message} ..."); + + // + return false; + } finally { + // + if (!channel.IsNull ()) { + channel.Dispose (); + } + } + } + + public bool Subscribe () + where TEvent : XEvent + where THandler : IXEventHandler { + // + // Extract required data ... + var eventType = typeof (TEvent); + var eventName = eventType.Name; + var handlerType = typeof (THandler); + + // + // Check event exists in list or not ... + if (!events.Contains (eventType)) { + // + // Add it if not exists ... + events.Add (eventType); + } + + // + // Check handlers Dictionary contains list or not ... + if (!handlers.ContainsKey (eventName)) { + handlers.Add (eventName, new List ()); + } + + // + // Check handlers Subscribed before or not ... + var isHandlerSubscribed = handlers[eventName] + .Any (h => h.GetType () == handlerType); + if (isHandlerSubscribed) { + // + logger.LogError ($"XEventService Exception: Duplicate Event Handler Subscription {eventName}/{handlerType.Name} ..."); + + // + return false; + } + + // + // Register Handler ... + handlers[eventName].Add (handlerType); + + // + // Do Consuming Handler ... + ConsumeEvent (); + + // + return true; + } + #endregion + + // + #region Private ... + /// + /// Consume Specific Event ... + /// + /// + private void ConsumeEvent () + where TEvent : XEvent { + // + // Extract required data ... + var eventType = typeof (TEvent); + var eventName = eventType.Name; + + // + // Create ChannelObject ... + var channel = new XChannel ( + exchangeName: eventName, + configuration: configuration, + dispatchConsumersAsync: true + ); + + // + // Create Consumer ... + var consumer = new AsyncEventingBasicConsumer (channel.Channel); + + // + // Set Consumer Delegate ... + consumer.Received += XConsumerReceivedDelegate; + + // + // Consume Async Consumer ... + channel.ConsumeAsync (consumer); + } + + /// + /// this Delegate method calls when a message recieved ... + /// + private async Task XConsumerReceivedDelegate ( + object sender, + BasicDeliverEventArgs ea + ) { + // + // Generate Required Data ... + var eventName = ea.Exchange; + var json = Encoding.UTF8 + .GetString ( + ea.Body + .ToArray () + ); + + // + // Do Processing Event ... + try { + // + // Process Event in non Blocking Tasks ... + await ProcessEvent (eventName, json) + .ConfigureAwait (false); + } catch (Exception ex) { + // + logger.LogError ($"XEventService Exception: Processing Event failed {eventName}/{ex.Message} ..."); + + // + XException.ActionFailed.Throw (); + } finally { + // + var channel = ((EventingBasicConsumer) sender).Model; + channel.BasicAck ( + deliveryTag: ea.DeliveryTag, + multiple: false + ); + } + } + + /// + /// Here we must call registered Event Handlers to Handle the Event ... + /// + private async Task ProcessEvent ( + string eventName, + string json + ) { + // + // Check Handler Exists for current Event ... + var isExistsHandler = handlers.ContainsKey (eventName); + if (!isExistsHandler) { + return; + } + + // + // Using Service Scope Factory ... + using (var scope = serviceScopeFactory.CreateScope ()) { + // + // Get Handler Subscriptions ... + var subscriptions = handlers[eventName]; + foreach (var subscription in subscriptions) { + // + // Inject Handler from Dependency Injections ... + var handler = scope.ServiceProvider.GetService (subscription); + if (handler.IsNull ()) { + // + logger.LogInformation ($"XEventService: there is no handler Registered in DependencyInjection for {eventName}/{subscription.Name}"); + + // + continue; + } + + // + // Retrieve Event Type ... + var eventType = events.SingleOrDefault (t => t.Name == eventName); + var @event = json.FromJSON (eventType); + var conreteType = typeof (IXEventHandler<>).MakeGenericType (eventType); + + // + // Call EventHandler 'HandleAsync' method ... + await ((Task) conreteType + .GetMethod (nameof (IXEventHandler.HandleAsync)) + .Invoke (handler, new object[] { @event })) + .ConfigureAwait (true); + } + } + } + #endregion + } +} \ No newline at end of file diff --git a/README.md b/README.md new file mode 100644 index 0000000..6227cfe --- /dev/null +++ b/README.md @@ -0,0 +1,195 @@ +# xEventService + +it is a Part of xDashboard on SaherElm IT Center which provides: EventBus core and RabbitMQ implementation of if + +## Implementing + +for using the event bus, you had to create a classlib module and add a reference to this module inside it; then start to create your events. +and for handling these events you had to add reference to this classlib in where you want to react to events, then create a class for handling this event which extends **IXEventHandler<>** interface. +and finally subscribing to event handlers you want. + +## RabbitMQ + +for using RabbitMQ Eventbus you had to: + +- create it's configuration section in your **appSettings.json** file, like this: + + ```javascript + { + ... + "EventServiceConfiguration": { + "Server": "YourRabbitMQHost", + "Port": rabbitMqPort, + "Username": "USERNAME", + "Password": "PASSWORD", + "AllowDynamicQueues": true + }, + ... + } + ``` + +**Note**: in configuration if you set **AllowDynamicQueues=false**, you had to provide all queue named and their configs manually. + +- register RabbitMQ implementation of IXEventBus in your **Startup**, like this: + +```c# + +... +public void ConfigureServices (IServiceCollection services) { + ... + // + // Register RabbitMQ Event Service ... + services.AddXEventService (Configuration); + ... +} +... +``` + +## Events + +for each Event you need to implement 3 classes: + +- a DTO class for provide required properties for creating Event Object instance; all properties in this class is public. + + > this used when you want to get som information from another sources, such as [FromBody] in Controller Actions. + +- one class which carry required data for Event object; this class must extends from **XEvent** class and must contains all Dto class properties in { get; protected set; } style; + +- one or more class which provide Event arrounds the Event object whith propper constructor to Create it's instance; + +for better undrestanding please follow this structure: + +```c# + +public class XPersonEventDto { + public string Firstname { get; set; } + public string Lastname { get; set; } +} + +public class XPersonEvent : XEvent { + public string Firstname { get; protected set; } + public string Lastname { get; protected set; } +} + +public class XCreatePersonEvent : XPersonEvent { + public XCreatePersonEvent ( + string firstname, + string lastname + ) { + Firstname = firstname; + Lastname = lastname; + } +} + +public class XPersonCreatedEvent : XPersonEvent { + public Guid Id { get; protected set; } + + public XPersonCreatedEvent ( + Guid id, + string firstname, + string lastname + ) { + Id = id; + Firstname = firstname; + Lastname = lastname; + } +} +``` + +## EventHandlers + +in where you want to handle this events and react to them, for each event you had to create an event handler class. + +```c# + +public class XCreatePersonEventHandler : IXEventHandler { + private readonly IXEventBus eventBus; + private ILogger logger; + + public XCreatePersonEventHandler ( + IXEventBus eventBus, + ILoggerFactory loggerFactory + ) { + this.logger = loggerFactory.CreateLogger (); + this.eventBus = eventBus; + } + + public async Task HandleAsync (XCreatePersonEvent @event) { + // + logger.LogInformation ($"Received Event: {@event.ToJSON()}"); + + // + // Publish XPersonCreated Event ... + var personCreatedEvent = new XPersonCreatedEvent ( + id: Guid.NewGuid (), + firstname: @event.Firstname, + lastname: @event.Lastname + ); + + // + eventBus.Publish (personCreatedEvent); + + // + await Task.CompletedTask; + } +} + +public class XPersonCreatedEventHandler : IXEventHandler { + private ILogger logger; + + public XPersonCreatedEventHandler (ILoggerFactory loggerFactory) { + this.logger = loggerFactory.CreateLogger (); + } + + public async Task HandleAsync (XPersonCreatedEvent @event) { + // + logger.LogInformation ($"Received Event: {@event.ToJSON()}"); + await Task.CompletedTask; + } +} +``` + +and register your event handlers in **Startup**, like this: + +```c# + +... +public void ConfigureServices (IServiceCollection services) { + ... + // + // Register Event Handlers ... + services.AddXEventHandler (); + services.AddXEventHandler (); + + // + // Register Event Service ... + services.AddXEventService (Configuration); + ... +} +... +``` + +## Subscribe + +for each event you had to respond in yor project, after create related EventHandler, last step is subscribing to the event in IXEventBus implementation in **Startup**, like this: + +```c# + +... +public void Configure (IApplicationBuilder app, IWebHostEnvironment env) { + ... + // + // Subscribe to Events ... + app.XEventSubscribe (); + app.XEventSubscribe (); + ... +} +``` + +## Maintainer + +Hadi Khazaee asl + +[https://www.saherelm.ir](https://www.saherelm.ir) + +[hadi_khazaee_asl@yahoo.com](mailto:hadi_khazaee_asl@yahoo.com) diff --git a/nuget.config b/nuget.config new file mode 100644 index 0000000..87b6eb0 --- /dev/null +++ b/nuget.config @@ -0,0 +1,7 @@ + + + + + + + \ No newline at end of file diff --git a/xEventService.csproj b/xEventService.csproj new file mode 100644 index 0000000..a0d434f --- /dev/null +++ b/xEventService.csproj @@ -0,0 +1,35 @@ + + + + + netstandard2.0 + xDashboard.xEventService + 1.0.0 + Hadi Khazaee Asl + SaherElm IT Center + + it is a Part of xDashboard on SaherElm IT Center which provides: EventBus core and RabbitMQ + implementation of if + + + + icon.png + + + + + + + + + + + + + + + + + + + \ No newline at end of file