Remove Old Stuffs for Refactor xEvent Service for Kafka and RabbitMQ ...
This commit is contained in:
@@ -1,14 +0,0 @@
|
||||
using System;
|
||||
|
||||
namespace xEventService.Base {
|
||||
public abstract class XBaseEvent {
|
||||
public string Type { get; protected set; }
|
||||
public DateTime Timestamp { get; protected set; }
|
||||
|
||||
protected XBaseEvent () {
|
||||
//
|
||||
Type = GetType ().Name;
|
||||
Timestamp = DateTime.UtcNow;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,51 +1,28 @@
|
||||
using System.Collections.Generic;
|
||||
using RabbitMQ.Client;
|
||||
using xEventService.Constants;
|
||||
|
||||
namespace xEventService.Configuration {
|
||||
namespace xEventService.Configuration
|
||||
{
|
||||
/// <summary>
|
||||
/// a configuration class for describing how to connect rabbit mq server
|
||||
/// and how channels must be declared ...
|
||||
/// </summary>
|
||||
public class XEventServiceConfiguration {
|
||||
public string Server { get; set; }
|
||||
public int Port { get; set; }
|
||||
public string Username { get; set; }
|
||||
public string Password { get; set; }
|
||||
public class XEventServiceConfiguration
|
||||
{
|
||||
/// <summary>
|
||||
/// Message Broker Address ...
|
||||
/// </summary>
|
||||
public string Url { get; set; }
|
||||
|
||||
//
|
||||
public bool AllowDynamicQueues { get; set; } = true;
|
||||
public XQueueArgs DefaultQueueArgs { get; set; } = new XQueueArgs ();
|
||||
public IDictionary<string, XQueueArgs> Queues { get; set; }
|
||||
/// <summary>
|
||||
/// Max Retries to Sending Message ...
|
||||
/// </summary>
|
||||
public int MaxRetries { get; set; } = 3;
|
||||
|
||||
//
|
||||
public bool AllowDynamicExchangess { get; set; } = true;
|
||||
public XExchangeArgs DefaultExchangeArgs { get; set; } = new XExchangeArgs ();
|
||||
public IDictionary<string, XExchangeArgs> Exchanges { get; set; }
|
||||
|
||||
//
|
||||
public static XEventServiceConfiguration GetDefaults () {
|
||||
return new XEventServiceConfiguration {
|
||||
Port = 5672,
|
||||
Username = "guest",
|
||||
Password = "guest",
|
||||
Server = "localhost",
|
||||
Queues = new Dictionary<string, XQueueArgs> { { "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;
|
||||
/// <summary>
|
||||
/// Allowed Message Broker ...
|
||||
/// </summary>
|
||||
public XEventBroker Broker { get; set; } = XEventBroker.None;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,19 @@
|
||||
using xExceptions.Attributes;
|
||||
|
||||
namespace xEventService.Constants
|
||||
{
|
||||
/// <summary>
|
||||
/// Available Brokers ...
|
||||
/// </summary>
|
||||
public enum XEventBroker
|
||||
{
|
||||
[StringValue("None")]
|
||||
None,
|
||||
|
||||
[StringValue("XKafka")]
|
||||
XKafka,
|
||||
|
||||
[StringValue("XRabbitMQ")]
|
||||
XRabbitMQ
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,13 @@
|
||||
namespace xEventService.Constants
|
||||
{
|
||||
public struct XEventServiceConstants
|
||||
{
|
||||
//
|
||||
// Service Extensions Log Tag ...
|
||||
public const string XEventServiceDILogTag = "XEventService";
|
||||
|
||||
//
|
||||
// Defaults ...
|
||||
public const string XDefaultTopic = "XSaherelm";
|
||||
}
|
||||
}
|
||||
+40
-92
@@ -1,26 +1,29 @@
|
||||
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;
|
||||
using xEventService.Extensions;
|
||||
using xExceptions.Constants;
|
||||
|
||||
namespace xEventService.DI {
|
||||
public static partial class XDIHelperExtension {
|
||||
namespace xEventService.DI
|
||||
{
|
||||
public static partial class XDIHelperExtension
|
||||
{
|
||||
/// <summary>
|
||||
/// Extract EventService Configurations ...
|
||||
/// </summary>
|
||||
/// <param name="source"></param>
|
||||
/// <returns></returns>
|
||||
public static XEventServiceConfiguration GetXEventServiceConfigurations (this IConfiguration source) {
|
||||
public static XEventServiceConfiguration GetXEventServiceConfigurations(
|
||||
this IConfiguration source
|
||||
)
|
||||
{
|
||||
//
|
||||
var configSection = source
|
||||
.GetSection (ConfigurationNodeNames.EVENT_SERVICE_NODE_NAME);
|
||||
var result = configSection.Get<XEventServiceConfiguration> ();
|
||||
.GetSection(ConfigurationNodeNames.EVENT_SERVICE_NODE_NAME);
|
||||
var result = configSection.Get<XEventServiceConfiguration>();
|
||||
|
||||
//
|
||||
return result;
|
||||
@@ -31,13 +34,14 @@ namespace xEventService.DI {
|
||||
/// </summary>
|
||||
/// <param name="services"></param>
|
||||
/// <param name="configuration"></param>
|
||||
public static void AddXEventServiceConfigurations (
|
||||
public static void AddXEventServiceConfigurations(
|
||||
this IServiceCollection services,
|
||||
IConfiguration configuration
|
||||
) {
|
||||
)
|
||||
{
|
||||
//
|
||||
var config = configuration.GetXEventServiceConfigurations ();
|
||||
services.AddXEventServiceConfigurations (config);
|
||||
var config = configuration.GetXEventServiceConfigurations();
|
||||
services.AddXEventServiceConfigurations(config);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
@@ -45,96 +49,40 @@ namespace xEventService.DI {
|
||||
/// </summary>
|
||||
/// <param name="services"></param>
|
||||
/// <param name="configuration"></param>
|
||||
public static void AddXEventServiceConfigurations (
|
||||
public static void AddXEventServiceConfigurations(
|
||||
this IServiceCollection services,
|
||||
XEventServiceConfiguration configuration
|
||||
) {
|
||||
)
|
||||
{
|
||||
//
|
||||
services.AddSingleton<XEventServiceConfiguration> (configuration);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Register Event Service ...
|
||||
/// </summary>
|
||||
/// <param name="services"></param>
|
||||
/// <param name="configuration"></param>
|
||||
public static void AddXEventService (
|
||||
this IServiceCollection services,
|
||||
IConfiguration configuration
|
||||
) {
|
||||
//
|
||||
// Check Configuration Registered ...
|
||||
var config = services.GetRegisteredService<XEventServiceConfiguration> ();
|
||||
if (config.IsNull ()) {
|
||||
// Validate ...
|
||||
if (!configuration.IsValid())
|
||||
{
|
||||
//
|
||||
Console.WriteLine ($"XEventService: there is no provided XEventServiceConfiguration, using default ...");
|
||||
|
||||
//
|
||||
// Register Default Configuration ...
|
||||
services.AddXEventServiceConfigurations (configuration);
|
||||
Log("Registration Failed, due Invalid Configuration ...");
|
||||
XException.InvalidConfiguration.Throw();
|
||||
}
|
||||
|
||||
//
|
||||
// Register IXEventBus ...
|
||||
var serviceBas = services.GetRegisteredService<IXEventBus> ();
|
||||
if (serviceBas.IsNull ()) {
|
||||
services.AddSingleton<IXEventBus, XEventBus> ();
|
||||
}
|
||||
// Register Configuration as Singleton ...
|
||||
services.AddSingleton(configuration);
|
||||
}
|
||||
|
||||
//
|
||||
#region Private ...
|
||||
/// <summary>
|
||||
/// a LogTag for Service ...
|
||||
/// </summary>
|
||||
private static string XLogTag = XEventServiceConstants.XEventServiceDILogTag;
|
||||
|
||||
/// <summary>
|
||||
/// Register an Event Handler ...
|
||||
/// print a log in Console ...
|
||||
/// </summary>
|
||||
/// <param name="services"></param>
|
||||
/// <param name="lifetime"></param>
|
||||
/// <typeparam name="TEvent"></typeparam>
|
||||
/// <typeparam name="THandler"></typeparam>
|
||||
/// <returns></returns>
|
||||
public static void AddXEventHandler<TEvent, THandler> (
|
||||
this IServiceCollection services,
|
||||
ServiceLifetime lifetime = ServiceLifetime.Transient
|
||||
)
|
||||
where TEvent : XEvent
|
||||
where THandler : IXEventHandler<TEvent> {
|
||||
//
|
||||
services.Add (new ServiceDescriptor (
|
||||
serviceType: typeof (THandler),
|
||||
implementationType: typeof (THandler),
|
||||
lifetime: lifetime
|
||||
));
|
||||
|
||||
//
|
||||
services.Add (new ServiceDescriptor (
|
||||
serviceType: typeof (IXEventHandler<TEvent>),
|
||||
implementationType: typeof (THandler),
|
||||
lifetime: lifetime
|
||||
));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Subscribe to specific Event ...
|
||||
/// </summary>
|
||||
/// <param name="app"></param>
|
||||
/// <typeparam name="TEvent"></typeparam>
|
||||
/// <typeparam name="THandler"></typeparam>
|
||||
/// <returns></returns>
|
||||
public static void XEventSubscribe<TEvent, THandler> (
|
||||
this IApplicationBuilder app,
|
||||
bool toExchange = true
|
||||
)
|
||||
where TEvent : XEvent
|
||||
where THandler : IXEventHandler<TEvent> {
|
||||
//
|
||||
var eventName = typeof (TEvent).Name;
|
||||
|
||||
//
|
||||
// Retrieve EventBus from IOC ...
|
||||
var eventBus = app.ApplicationServices.GetRequiredService<IXEventBus> ();
|
||||
|
||||
//
|
||||
// Subscribe to Event Handler ...
|
||||
var isSubscribed = eventBus.Subscribe<TEvent, THandler> ();
|
||||
Console.WriteLine ($"XEventService: Event Handler subscription result: {eventName} => {isSubscribed} ...");
|
||||
/// <param name="message"></param>
|
||||
private static void Log(string message)
|
||||
{
|
||||
Console.WriteLine($"{XLogTag} => {message}");
|
||||
}
|
||||
#endregion
|
||||
}
|
||||
}
|
||||
@@ -1,288 +0,0 @@
|
||||
using System;
|
||||
using RabbitMQ.Client;
|
||||
using xCommons.Extensions;
|
||||
using xEventService.Configuration;
|
||||
using xEventService.Models;
|
||||
using xExceptions.Constants;
|
||||
|
||||
namespace xEventService.Extensions {
|
||||
public static class ConfigurationExtensions {
|
||||
/// <summary>
|
||||
/// Create Connection Factory Based on
|
||||
/// provided Configuration ...
|
||||
/// </summary>
|
||||
/// <param name="source"></param>
|
||||
/// <returns></returns>
|
||||
public static ConnectionFactory GetConnectionFactory (this XEventServiceConfiguration source) {
|
||||
//
|
||||
// Validate Args ...
|
||||
if (source.IsNull ()) {
|
||||
XException.InvalidConfiguration.Throw ();
|
||||
}
|
||||
|
||||
//
|
||||
// Create Connection Factory ...
|
||||
var result = new ConnectionFactory {
|
||||
Port = source.Port,
|
||||
HostName = source.Server,
|
||||
UserName = source.Username,
|
||||
Password = source.Password
|
||||
};
|
||||
|
||||
//
|
||||
// Return Result ...
|
||||
return result;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Create a Connection to Server ...
|
||||
/// </summary>
|
||||
/// <param name="source"></param>
|
||||
/// <returns></returns>
|
||||
public static IConnection Connect (this XEventServiceConfiguration source) {
|
||||
//
|
||||
// Open Connection ...
|
||||
IConnection result = null;
|
||||
try {
|
||||
//
|
||||
// Retrieve Connection Factory ...
|
||||
var connectionFactory = source.GetConnectionFactory ();
|
||||
|
||||
//
|
||||
// Create Connection ...
|
||||
result = connectionFactory.CreateConnection ();
|
||||
} catch (Exception ex) {
|
||||
//
|
||||
Console.WriteLine ($"XEventService Exception: Connection Failed, {ex.Message} ...");
|
||||
|
||||
//
|
||||
XException.ActionFailed.Throw ();
|
||||
}
|
||||
|
||||
//
|
||||
return result;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Validate a Queue Name based on Configurations ...
|
||||
/// </summary>
|
||||
/// <param name="source"></param>
|
||||
/// <param name="queueName"></param>
|
||||
public static void ValidateQueue (
|
||||
this XEventServiceConfiguration source,
|
||||
string queueName
|
||||
) {
|
||||
//
|
||||
// Validate Args ...
|
||||
if (source.IsNull ()) {
|
||||
XException.InvalidArgs.Throw ();
|
||||
}
|
||||
|
||||
//
|
||||
// Validate Queues ...
|
||||
if ((source.Queues.IsNull () &&
|
||||
!source.AllowDynamicQueues) ||
|
||||
(!source.Queues.IsNull () &&
|
||||
!source.Queues.Keys.HasChild () &&
|
||||
!source.AllowDynamicQueues)) {
|
||||
//
|
||||
Console.WriteLine ($"XEventService Exception: there isn't any configured Queue ...");
|
||||
|
||||
//
|
||||
XException.InvalidArgs.Throw ();
|
||||
}
|
||||
|
||||
//
|
||||
// Check queueName not empty ...
|
||||
if (queueName.IsNullOrEmpty ()) {
|
||||
//
|
||||
Console.WriteLine ($"XEventService Exception: Invalid QueueName ...");
|
||||
|
||||
//
|
||||
XException.InvalidArgs.Throw ();
|
||||
}
|
||||
|
||||
//
|
||||
// Check queueName Exists ...
|
||||
if (!source.Queues.IsNull () &&
|
||||
!source.Queues.ContainsKey (queueName) &&
|
||||
!source.AllowDynamicQueues) {
|
||||
//
|
||||
Console.WriteLine ($"XEventService Exception: QueueName {queueName} not found ...");
|
||||
|
||||
//
|
||||
XException.InvalidArgs.Throw ();
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Validate a Exchange Name based on Configurations ...
|
||||
/// </summary>
|
||||
/// <param name="source"></param>
|
||||
/// <param name="exchangeName"></param>
|
||||
public static void ValidateExchange (
|
||||
this XEventServiceConfiguration source,
|
||||
string exchangeName
|
||||
) {
|
||||
//
|
||||
// Validate Args ...
|
||||
if (source.IsNull ()) {
|
||||
XException.InvalidArgs.Throw ();
|
||||
}
|
||||
|
||||
//
|
||||
// Validate Exchanges ...
|
||||
if ((source.Exchanges.IsNull () &&
|
||||
!source.AllowDynamicExchangess) ||
|
||||
(!source.Exchanges.IsNull () &&
|
||||
!source.Exchanges.Keys.HasChild () &&
|
||||
!source.AllowDynamicExchangess)) {
|
||||
//
|
||||
Console.WriteLine ($"XEventService Exception: there isn't any configured Exchanges ...");
|
||||
|
||||
//
|
||||
XException.InvalidArgs.Throw ();
|
||||
}
|
||||
|
||||
//
|
||||
// Check exchangeName not empty ...
|
||||
if (exchangeName.IsNullOrEmpty ()) {
|
||||
//
|
||||
Console.WriteLine ($"XEventService Exception: Invalid ExchangeName ...");
|
||||
|
||||
//
|
||||
XException.InvalidArgs.Throw ();
|
||||
}
|
||||
|
||||
//
|
||||
// Check exchangeName Exists ...
|
||||
if (!source.Exchanges.IsNull () &&
|
||||
!source.Exchanges.ContainsKey (exchangeName) &&
|
||||
!source.AllowDynamicExchangess) {
|
||||
//
|
||||
Console.WriteLine ($"XEventService Exception: ExchangeName {exchangeName} not found ...");
|
||||
|
||||
//
|
||||
XException.InvalidArgs.Throw ();
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Retrieve specific QueueArgs ...
|
||||
/// </summary>
|
||||
/// <param name="source"></param>
|
||||
/// <param name="queueName"></param>
|
||||
/// <returns></returns>
|
||||
public static XQueueArgs GetQueueArgs (
|
||||
this XEventServiceConfiguration source,
|
||||
string queueName
|
||||
) {
|
||||
//
|
||||
// Validate Queue Name ...
|
||||
source.ValidateQueue (queueName);
|
||||
|
||||
//
|
||||
var result = source.Queues
|
||||
.IsNull () ? null : source
|
||||
.Queues[queueName];
|
||||
if (result.IsNull ()) {
|
||||
//
|
||||
// Prepare Default Queue ...
|
||||
result = source.DefaultQueueArgs;
|
||||
if (result.IsNull ()) {
|
||||
result = new XQueueArgs () {
|
||||
Name = queueName
|
||||
};
|
||||
} else {
|
||||
result.Name = queueName;
|
||||
}
|
||||
}
|
||||
|
||||
//
|
||||
return result;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Retrieve specific ExchangeArgs ...
|
||||
/// </summary>
|
||||
/// <param name="source"></param>
|
||||
/// <param name="exchangeName"></param>
|
||||
/// <returns></returns>
|
||||
public static XExchangeArgs GetExchangeArgs (
|
||||
this XEventServiceConfiguration source,
|
||||
string exchangeName
|
||||
) {
|
||||
//
|
||||
// Validate Exchange Name ...
|
||||
source.ValidateExchange (exchangeName);
|
||||
|
||||
//
|
||||
var result = source.Exchanges
|
||||
.IsNull () ? null : source
|
||||
.Exchanges[exchangeName];
|
||||
if (result.IsNull ()) {
|
||||
//
|
||||
// Prepare Default Queue ...
|
||||
result = source.DefaultExchangeArgs;
|
||||
if (result.IsNull ()) {
|
||||
result = new XExchangeArgs () {
|
||||
Name = exchangeName,
|
||||
Type = ExchangeType.Fanout
|
||||
};
|
||||
} else {
|
||||
result.Name = exchangeName;
|
||||
}
|
||||
}
|
||||
|
||||
//
|
||||
return result;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Create a Channel ...
|
||||
/// </summary>
|
||||
/// <param name="source"></param>
|
||||
/// <param name="exchangeName"></param>
|
||||
/// <returns></returns>
|
||||
public static XChannel CreateChannel (
|
||||
this XEventServiceConfiguration source,
|
||||
string exchangeName = null
|
||||
) {
|
||||
//
|
||||
// Validate ExchangeName and Retrieve ExchangeArgs if provided ...
|
||||
var exchangeArgs = exchangeName
|
||||
.IsNullOrEmpty () ?
|
||||
null :
|
||||
source.GetExchangeArgs (exchangeName);
|
||||
|
||||
//
|
||||
// Create Connection ...
|
||||
var connection = source.Connect ();
|
||||
var channel = connection.CreateModel ();
|
||||
|
||||
//
|
||||
// Create XChannel Model ...
|
||||
var result = new XChannel (
|
||||
exchangeName: exchangeName,
|
||||
connection: connection
|
||||
);
|
||||
|
||||
//
|
||||
// Declaring Queue if Provided ...
|
||||
if (!exchangeArgs.IsNull () &&
|
||||
!exchangeName.IsNullOrEmpty ()) {
|
||||
//
|
||||
// Declaring Queue ...
|
||||
result.Channel.ExchangeDeclare (
|
||||
exchange: exchangeName,
|
||||
type: exchangeArgs.Type,
|
||||
durable: exchangeArgs.Durable,
|
||||
autoDelete: exchangeArgs.AutoDelete
|
||||
);
|
||||
}
|
||||
|
||||
//
|
||||
return result;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,44 +0,0 @@
|
||||
using System;
|
||||
using System.Text;
|
||||
using xCommons.Extensions;
|
||||
|
||||
namespace xEventService.Extensions {
|
||||
public static class XEventModelExtensions {
|
||||
/// <summary>
|
||||
/// Convert an Object to bytes array for publishing
|
||||
/// </summary>
|
||||
/// <param name="source"></param>
|
||||
/// <typeparam name="T"></typeparam>
|
||||
/// <returns></returns>
|
||||
public static byte[] ToBody<T> (this T source) {
|
||||
//
|
||||
var json = source
|
||||
.ToJSON ();
|
||||
|
||||
//
|
||||
var result = json
|
||||
.ToBytes ();
|
||||
|
||||
//
|
||||
return result;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Convert Recieved Bytes to Specific Type ...
|
||||
/// </summary>
|
||||
/// <param name="source"></param>
|
||||
/// <typeparam name="T"></typeparam>
|
||||
/// <returns></returns>
|
||||
public static T FromBody<T> (this ReadOnlyMemory<byte> source) {
|
||||
//
|
||||
var jsonString = Encoding.UTF8
|
||||
.GetString (
|
||||
source.ToArray ()
|
||||
);
|
||||
var result = jsonString.FromJSON<T> ();
|
||||
|
||||
//
|
||||
return result;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,81 @@
|
||||
using xCommons.Extensions;
|
||||
using xEventService.Configuration;
|
||||
using xEventService.Constants;
|
||||
using xEventService.Models;
|
||||
|
||||
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 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;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,11 +0,0 @@
|
||||
using xEventService.Models;
|
||||
|
||||
namespace xEventService.Interfaces {
|
||||
public interface IXEventBus {
|
||||
bool Publish<T> (T @event) where T : XEvent;
|
||||
|
||||
bool Subscribe<TEvent, THandler> ()
|
||||
where TEvent : XEvent
|
||||
where THandler : IXEventHandler<TEvent>;
|
||||
}
|
||||
}
|
||||
@@ -1,11 +0,0 @@
|
||||
using System.Threading.Tasks;
|
||||
using xEventService.Models;
|
||||
|
||||
namespace xEventService.Interfaces {
|
||||
public interface IXEventHandler<in TEvent> : IXEventHandler
|
||||
where TEvent : XEvent {
|
||||
Task HandleAsync (TEvent @event);
|
||||
}
|
||||
|
||||
public interface IXEventHandler { }
|
||||
}
|
||||
@@ -0,0 +1,35 @@
|
||||
using xEventService.Models;
|
||||
|
||||
namespace xEventService.Interfaces
|
||||
{
|
||||
/// <summary>
|
||||
/// a Service for Produce Kafka Messages ...
|
||||
/// </summary>
|
||||
public interface IXKafkaProducerService : 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,
|
||||
string topic = XEventServiceConstants.XDefaultTopic,
|
||||
CancellationToken cancellationToken = default
|
||||
);
|
||||
|
||||
/// <summary>
|
||||
/// Sending a Message ...
|
||||
/// </summary>
|
||||
/// <param name="message"></param>
|
||||
/// <param name="topic"></param>
|
||||
/// <param name="cancellationToken"></param>
|
||||
/// <returns></returns>
|
||||
Task<bool> SendMessageAsync(
|
||||
XEventMessage message,
|
||||
string topic = XEventServiceConstants.XDefaultTopic,
|
||||
CancellationToken cancellationToken = default
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -1,224 +0,0 @@
|
||||
using System;
|
||||
using RabbitMQ.Client;
|
||||
using RabbitMQ.Client.Events;
|
||||
using xCommons.Extensions;
|
||||
using xEventService.Configuration;
|
||||
using xEventService.Extensions;
|
||||
using xExceptions.Constants;
|
||||
|
||||
namespace xEventService.Models {
|
||||
public class XChannel {
|
||||
//
|
||||
public IModel Channel { get; protected set; }
|
||||
public string QueueName { get; protected set; }
|
||||
public string ExchangeName { get; protected set; }
|
||||
public IConnection Connection { get; protected set; }
|
||||
|
||||
//
|
||||
#region Constructors ...
|
||||
public XChannel (
|
||||
string exchangeName = null,
|
||||
IConnection connection = null,
|
||||
IModel channel = null,
|
||||
EventingBasicConsumer consumer = null
|
||||
) {
|
||||
//
|
||||
// Validate Connection ...
|
||||
if (connection.IsNull ()) {
|
||||
XException.InvalidArgs.Throw ();
|
||||
}
|
||||
Connection = connection;
|
||||
|
||||
//
|
||||
// Validate Channel ...
|
||||
if (channel.IsNull ()) {
|
||||
channel = connection.CreateModel ();
|
||||
}
|
||||
Channel = channel;
|
||||
|
||||
//
|
||||
// Check Queue Name and it's Declaration ...
|
||||
if (!exchangeName.IsNullOrEmpty ()) {
|
||||
//
|
||||
ExchangeName = exchangeName;
|
||||
}
|
||||
}
|
||||
|
||||
public XChannel (
|
||||
XEventServiceConfiguration configuration,
|
||||
bool dispatchConsumersAsync = true,
|
||||
string exchangeName = null
|
||||
) {
|
||||
//
|
||||
// Generate Factory ...
|
||||
var factory = configuration.GetConnectionFactory ();
|
||||
if (dispatchConsumersAsync) {
|
||||
factory.DispatchConsumersAsync = true;
|
||||
}
|
||||
|
||||
//
|
||||
// Create Cahnnel ...
|
||||
Connection = factory.CreateConnection ();
|
||||
Channel = Connection.CreateModel ();
|
||||
|
||||
//
|
||||
// Check Exchange Declaration if provided ...
|
||||
if (!exchangeName.IsNullOrEmpty ()) {
|
||||
//
|
||||
ExchangeName = exchangeName;
|
||||
var exchangeArgs = configuration.GetExchangeArgs (ExchangeName);
|
||||
|
||||
//
|
||||
// Declaring Exchange ...
|
||||
Channel.ExchangeDeclare (
|
||||
exchange: exchangeName,
|
||||
type: exchangeArgs.Type,
|
||||
durable: exchangeArgs.Durable,
|
||||
autoDelete: exchangeArgs.AutoDelete
|
||||
);
|
||||
|
||||
//
|
||||
// Create a Queue ...
|
||||
var queueName = $"{exchangeName}[{Guid.NewGuid().ToString().GetDigits()}]";
|
||||
var queueArgs = configuration.GetQueueArgs (queueName);
|
||||
QueueName = queueName;
|
||||
Channel.QueueDeclare (
|
||||
queue: queueName,
|
||||
durable: queueArgs.Durable,
|
||||
exclusive: queueArgs.Exclusive,
|
||||
autoDelete: queueArgs.AutoDelete
|
||||
);
|
||||
|
||||
//
|
||||
// Bind Queue to Exchange ...
|
||||
Channel.QueueBind (
|
||||
queue: queueName,
|
||||
exchange: exchangeName,
|
||||
routingKey: ""
|
||||
);
|
||||
}
|
||||
}
|
||||
#endregion
|
||||
|
||||
//
|
||||
#region Actions ...
|
||||
/// <summary>
|
||||
/// Close Channel ...
|
||||
/// </summary>
|
||||
public void CloseChannel () {
|
||||
//
|
||||
// Check Channel Exists ...
|
||||
if (!this.Channel.IsNull ()) {
|
||||
//
|
||||
// Close Channel if it's Open ...
|
||||
if (this.Channel.IsOpen) {
|
||||
this.Channel.Close ();
|
||||
}
|
||||
|
||||
//
|
||||
this.Channel.Dispose ();
|
||||
this.Channel = null;
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Close Connection ...
|
||||
/// </summary>
|
||||
public void CloseConnection () {
|
||||
//
|
||||
// Check Connection Exists ...
|
||||
if (!this.Connection.IsNull ()) {
|
||||
//
|
||||
// Close Connection if it's Open ...
|
||||
if (this.Connection.IsOpen) {
|
||||
this.Connection.Close ();
|
||||
}
|
||||
|
||||
//
|
||||
this.Connection.Dispose ();
|
||||
this.Connection = null;
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Publish a Message through Channel on Declare Queue ...
|
||||
/// </summary>
|
||||
/// <param name="message"></param>
|
||||
/// <typeparam name="T"></typeparam>
|
||||
/// <returns></returns>
|
||||
public bool Publish<T> (T message) {
|
||||
//
|
||||
// Validate Args ...
|
||||
if (this.Connection.IsNull () ||
|
||||
!this.Connection.IsOpen ||
|
||||
this.Channel.IsNull () ||
|
||||
!this.Channel.IsOpen) {
|
||||
//
|
||||
Console.WriteLine ($"XEventService Exception: Publish Failed ...");
|
||||
|
||||
//
|
||||
XException.InvalidArgs.Throw ();
|
||||
}
|
||||
|
||||
//
|
||||
// Check Publish Method ...
|
||||
//
|
||||
// Check Exchange Name ...
|
||||
if (this.ExchangeName.IsNullOrEmpty ()) {
|
||||
//
|
||||
Console.WriteLine ($"XEventService Exception: Publish Failed, Invalid Exchange Name {ExchangeName} ...");
|
||||
|
||||
//
|
||||
XException.InvalidArgs.Throw ();
|
||||
}
|
||||
|
||||
//
|
||||
try {
|
||||
//
|
||||
// Publish Message through Channel to Declared Queue ...
|
||||
var body = message.ToBody ();
|
||||
this.Channel.BasicPublish (
|
||||
exchange: ExchangeName,
|
||||
routingKey: "",
|
||||
basicProperties : null,
|
||||
body : body
|
||||
);
|
||||
|
||||
//
|
||||
return true;
|
||||
} catch (Exception ex) {
|
||||
//
|
||||
Console.WriteLine ($"XEventService Exception: Publish Failed, {ex.Message} ...");
|
||||
|
||||
//
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
public void Consume (EventingBasicConsumer consumer) {
|
||||
Channel.BasicConsume (
|
||||
queue: QueueName,
|
||||
autoAck: true,
|
||||
consumer: consumer
|
||||
);
|
||||
}
|
||||
|
||||
public void ConsumeAsync (AsyncEventingBasicConsumer consumer) {
|
||||
Channel.BasicConsume (
|
||||
queue: QueueName,
|
||||
autoAck: true,
|
||||
consumer: consumer
|
||||
);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Dispose XChannel ...
|
||||
/// </summary>
|
||||
public void Dispose () {
|
||||
//
|
||||
this.CloseChannel ();
|
||||
this.CloseConnection ();
|
||||
}
|
||||
#endregion
|
||||
}
|
||||
}
|
||||
@@ -1,5 +0,0 @@
|
||||
using xEventService.Base;
|
||||
|
||||
namespace xEventService.Models {
|
||||
public class XEvent : XBaseEvent { }
|
||||
}
|
||||
@@ -0,0 +1,40 @@
|
||||
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>();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,18 @@
|
||||
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; }
|
||||
}
|
||||
}
|
||||
@@ -1,260 +0,0 @@
|
||||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.Linq;
|
||||
using System.Text;
|
||||
using System.Threading.Tasks;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using RabbitMQ.Client.Events;
|
||||
using xCommons.Extensions;
|
||||
using xEventService.Configuration;
|
||||
using xEventService.Extensions;
|
||||
using xEventService.Interfaces;
|
||||
using xEventService.Models;
|
||||
using xExceptions.Constants;
|
||||
|
||||
namespace xEventService.Providers {
|
||||
public sealed class XEventBus : IXEventBus {
|
||||
//
|
||||
#region Properties ...
|
||||
private readonly List<Type> events;
|
||||
private readonly ILogger<XEventBus> logger;
|
||||
private readonly IServiceScopeFactory serviceScopeFactory;
|
||||
private readonly XEventServiceConfiguration configuration;
|
||||
private readonly IDictionary<string, List<Type>> handlers;
|
||||
#endregion
|
||||
|
||||
//
|
||||
#region Constructors ...
|
||||
public XEventBus (
|
||||
ILoggerFactory loggerFactory,
|
||||
IServiceScopeFactory serviceScopeFactory,
|
||||
XEventServiceConfiguration configuration
|
||||
) {
|
||||
//
|
||||
this.configuration = configuration;
|
||||
this.serviceScopeFactory = serviceScopeFactory;
|
||||
|
||||
//
|
||||
events = new List<Type> ();
|
||||
handlers = new Dictionary<string, List<Type>> ();
|
||||
logger = loggerFactory.CreateLogger<XEventBus> ();
|
||||
}
|
||||
#endregion
|
||||
|
||||
//
|
||||
#region Abstracts ...
|
||||
public bool Publish<T> (T @event) where T : XEvent {
|
||||
//
|
||||
// Validate Args ...
|
||||
if (@event.IsNull ()) {
|
||||
return false;
|
||||
}
|
||||
|
||||
//
|
||||
// Prepare Channel ...
|
||||
XChannel channel = null;
|
||||
|
||||
//
|
||||
// Extract event name ...
|
||||
var exchangeName = @event.GetType ().Name;
|
||||
|
||||
//
|
||||
// Do Action ...
|
||||
try {
|
||||
//
|
||||
// Get XChannel Model ...
|
||||
channel = configuration.CreateChannel (exchangeName);
|
||||
|
||||
//
|
||||
// Publish Command ...
|
||||
channel.Publish (message: @event);
|
||||
|
||||
//
|
||||
// Return Result ...
|
||||
return true;
|
||||
} catch (Exception ex) {
|
||||
//
|
||||
logger.LogError ($"XEventService Exception: Publishing Failed {exchangeName}/{ex.Message} ...");
|
||||
|
||||
//
|
||||
return false;
|
||||
} finally {
|
||||
//
|
||||
if (!channel.IsNull ()) {
|
||||
channel.Dispose ();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public bool Subscribe<TEvent, THandler> ()
|
||||
where TEvent : XEvent
|
||||
where THandler : IXEventHandler<TEvent> {
|
||||
//
|
||||
// Extract required data ...
|
||||
var eventType = typeof (TEvent);
|
||||
var eventName = eventType.Name;
|
||||
var handlerType = typeof (THandler);
|
||||
|
||||
//
|
||||
// Check event exists in list or not ...
|
||||
if (!events.Contains (eventType)) {
|
||||
//
|
||||
// Add it if not exists ...
|
||||
events.Add (eventType);
|
||||
}
|
||||
|
||||
//
|
||||
// Check handlers Dictionary contains list or not ...
|
||||
if (!handlers.ContainsKey (eventName)) {
|
||||
handlers.Add (eventName, new List<Type> ());
|
||||
}
|
||||
|
||||
//
|
||||
// Check handlers Subscribed before or not ...
|
||||
var isHandlerSubscribed = handlers[eventName]
|
||||
.Any (h => h.GetType () == handlerType);
|
||||
if (isHandlerSubscribed) {
|
||||
//
|
||||
logger.LogError ($"XEventService Exception: Duplicate Event Handler Subscription {eventName}/{handlerType.Name} ...");
|
||||
|
||||
//
|
||||
return false;
|
||||
}
|
||||
|
||||
//
|
||||
// Register Handler ...
|
||||
handlers[eventName].Add (handlerType);
|
||||
|
||||
//
|
||||
// Do Consuming Handler ...
|
||||
ConsumeEvent<TEvent> ();
|
||||
|
||||
//
|
||||
return true;
|
||||
}
|
||||
#endregion
|
||||
|
||||
//
|
||||
#region Private ...
|
||||
/// <summary>
|
||||
/// Consume Specific Event ...
|
||||
/// </summary>
|
||||
/// <typeparam name="TEvent"></typeparam>
|
||||
private void ConsumeEvent<TEvent> ()
|
||||
where TEvent : XEvent {
|
||||
//
|
||||
// Extract required data ...
|
||||
var eventType = typeof (TEvent);
|
||||
var eventName = eventType.Name;
|
||||
|
||||
//
|
||||
// Create ChannelObject ...
|
||||
var channel = new XChannel (
|
||||
exchangeName: eventName,
|
||||
configuration: configuration,
|
||||
dispatchConsumersAsync: true
|
||||
);
|
||||
|
||||
//
|
||||
// Create Consumer ...
|
||||
var consumer = new AsyncEventingBasicConsumer (channel.Channel);
|
||||
|
||||
//
|
||||
// Set Consumer Delegate ...
|
||||
consumer.Received += XConsumerReceivedDelegate;
|
||||
|
||||
//
|
||||
// Consume Async Consumer ...
|
||||
channel.ConsumeAsync (consumer);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// this Delegate method calls when a message recieved ...
|
||||
/// </summary>
|
||||
private async Task XConsumerReceivedDelegate (
|
||||
object sender,
|
||||
BasicDeliverEventArgs ea
|
||||
) {
|
||||
//
|
||||
// Generate Required Data ...
|
||||
var eventName = ea.Exchange;
|
||||
var json = Encoding.UTF8
|
||||
.GetString (
|
||||
ea.Body
|
||||
.ToArray ()
|
||||
);
|
||||
|
||||
//
|
||||
// Do Processing Event ...
|
||||
try {
|
||||
//
|
||||
// Process Event in non Blocking Tasks ...
|
||||
await ProcessEvent (eventName, json)
|
||||
.ConfigureAwait (false);
|
||||
} catch (Exception ex) {
|
||||
//
|
||||
logger.LogError ($"XEventService Exception: Processing Event failed {eventName}/{ex.Message} ...");
|
||||
|
||||
//
|
||||
XException.ActionFailed.Throw ();
|
||||
} finally {
|
||||
//
|
||||
var channel = ((EventingBasicConsumer) sender).Model;
|
||||
channel.BasicAck (
|
||||
deliveryTag: ea.DeliveryTag,
|
||||
multiple: false
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Here we must call registered Event Handlers to Handle the Event ...
|
||||
/// </summary>
|
||||
private async Task ProcessEvent (
|
||||
string eventName,
|
||||
string json
|
||||
) {
|
||||
//
|
||||
// Check Handler Exists for current Event ...
|
||||
var isExistsHandler = handlers.ContainsKey (eventName);
|
||||
if (!isExistsHandler) {
|
||||
return;
|
||||
}
|
||||
|
||||
//
|
||||
// Using Service Scope Factory ...
|
||||
using (var scope = serviceScopeFactory.CreateScope ()) {
|
||||
//
|
||||
// Get Handler Subscriptions ...
|
||||
var subscriptions = handlers[eventName];
|
||||
foreach (var subscription in subscriptions) {
|
||||
//
|
||||
// Inject Handler from Dependency Injections ...
|
||||
var handler = scope.ServiceProvider.GetService (subscription);
|
||||
if (handler.IsNull ()) {
|
||||
//
|
||||
logger.LogInformation ($"XEventService: there is no handler Registered in DependencyInjection for {eventName}/{subscription.Name}");
|
||||
|
||||
//
|
||||
continue;
|
||||
}
|
||||
|
||||
//
|
||||
// Retrieve Event Type ...
|
||||
var eventType = events.SingleOrDefault (t => t.Name == eventName);
|
||||
var @event = json.FromJSON (eventType);
|
||||
var conreteType = typeof (IXEventHandler<>).MakeGenericType (eventType);
|
||||
|
||||
//
|
||||
// Call EventHandler 'HandleAsync' method ...
|
||||
await ((Task) conreteType
|
||||
.GetMethod (nameof (IXEventHandler<XEvent>.HandleAsync))
|
||||
.Invoke (handler, new object[] { @event }))
|
||||
.ConfigureAwait (true);
|
||||
}
|
||||
}
|
||||
}
|
||||
#endregion
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,32 @@
|
||||
using xCommons.Extensions;
|
||||
using xExceptions.Constants;
|
||||
using xEventService.Extensions;
|
||||
using xEventService.Interfaces;
|
||||
using xEventService.Configuration;
|
||||
|
||||
namespace xEventService.Providers
|
||||
{
|
||||
public class XKafkaProducerService : IXKafkaProducerService
|
||||
{
|
||||
private readonly IProducer<Null, string> producer;
|
||||
private readonly ILogger<RpkKafkaProducerService> logger;
|
||||
|
||||
public XKafkaProducerService(
|
||||
ILoggeer<XKafkaProducerService> logger,
|
||||
XEventServiceConfiguration configuration
|
||||
)
|
||||
{
|
||||
//
|
||||
this.logger = logger;
|
||||
|
||||
//
|
||||
// Validate Configurations ...
|
||||
if (!configuration.IsValid() ||
|
||||
configuration.Broker != Constants.XEventBroker.XKafka
|
||||
)
|
||||
{
|
||||
XException.InvalidConfiguration.Throw();
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -36,6 +36,7 @@
|
||||
<!-- Dependencies -->
|
||||
<ItemGroup>
|
||||
<PackageReference Include="RabbitMQ.Client" Version="6.2.2" />
|
||||
<PackageReference Include="Confluent.Kafka" Version="2.14.2" />
|
||||
</ItemGroup>
|
||||
|
||||
<!-- For XML Documentation Support -->
|
||||
|
||||
Reference in New Issue
Block a user