### FILE: C:\Users\SaherElm\Documents\Projects\xSaherelmWorkspace\Modules\xEventService\Base\XBaseEvent.cs 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; } } } ### FILE: C:\Users\SaherElm\Documents\Projects\xSaherelmWorkspace\Modules\xEventService\Configuration\XEventServiceConfiguration.cs 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; } } ### 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\DI\XDIHelperExtension.cs 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 partial 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} ..."); } } } ### FILE: C:\Users\SaherElm\Documents\Projects\xSaherelmWorkspace\Modules\xEventService\Extensions\ConfigurationExtensions.cs 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; } } } ### FILE: C:\Users\SaherElm\Documents\Projects\xSaherelmWorkspace\Modules\xEventService\Extensions\XEventModelExtensions.cs 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; } } } ### FILE: C:\Users\SaherElm\Documents\Projects\xSaherelmWorkspace\Modules\xEventService\Interfaces\IXEventBus.cs using xEventService.Models; namespace xEventService.Interfaces { public interface IXEventBus { bool Publish (T @event) where T : XEvent; bool Subscribe () where TEvent : XEvent where THandler : IXEventHandler; } } ### FILE: C:\Users\SaherElm\Documents\Projects\xSaherelmWorkspace\Modules\xEventService\Interfaces\IXEventHandler.cs using System.Threading.Tasks; using xEventService.Models; namespace xEventService.Interfaces { public interface IXEventHandler : IXEventHandler where TEvent : XEvent { Task HandleAsync (TEvent @event); } public interface IXEventHandler { } } ### FILE: C:\Users\SaherElm\Documents\Projects\xSaherelmWorkspace\Modules\xEventService\Models\XChannel.cs 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 } } ### FILE: C:\Users\SaherElm\Documents\Projects\xSaherelmWorkspace\Modules\xEventService\Models\XEvent.cs using xEventService.Base; namespace xEventService.Models { public class XEvent : XBaseEvent { } } ### FILE: C:\Users\SaherElm\Documents\Projects\xSaherelmWorkspace\Modules\xEventService\Providers\XEventBus.cs 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 } }