saherelm/xEventService
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Complete xEventService ...

Browse Source
This commit is contained in:
saherelm
2026-08-19 22:44:39 +03:30
parent e05420b1aa
commit 079048dff5
8 changed files with 389 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Providers/XEventServiceProvider.cs
+159
View File
@@ -0,0 +1,159 @@
using System.Threading;
using System.Threading.Tasks;
using xCommons.Extensions;
using xEventService.Configuration;
using xEventService.Constants;
using xEventService.Extensions;
using xEventService.Interfaces;
using xEventService.Models;
using xExceptions.Constants;
namespace xEventService.Providers
{
public class XEventServiceProvider : IXEventServiceProvider
{
//
private readonly XEventServiceConfiguration configuration;
private readonly IXKafkaProducerService kafkaProducerService;
private readonly IXZeroMQProducerService zeroMQProducerService;
private readonly IXRabbitMQProducerService rabbitMQProducerService;
public XEventServiceProvider(
XEventServiceConfiguration configuration,
IXKafkaProducerService kafkaProducerService = null,
IXZeroMQProducerService zeroMQProducerService = null,
IXRabbitMQProducerService rabbitMQProducerService = null
)
{
//
this.configuration = configuration;
this.kafkaProducerService = kafkaProducerService;
this.zeroMQProducerService = zeroMQProducerService;
this.rabbitMQProducerService = rabbitMQProducerService;
//
// Validate ...
var isValid = configuration.IsValid() &&
configuration.Broker == XEventBroker.XKafka
? kafkaProducerService != null
: configuration.Broker == XEventBroker.XZeroMQ
? zeroMQProducerService != null
: configuration.Broker == XEventBroker.XRabbitMQ
? rabbitMQProducerService != null
: false;
if (!isValid)
{
XException.InvalidConfiguration.Throw();
}
}
/// <summary>
/// Send Message to Broker Using XEventRequest instance ...
/// </summary>
/// <param name="request"></param>
/// <param name="cancellationToken"></param>
/// <returns></returns>
public async Task<bool> SendMessageAsync(
XEventRequest request,
CancellationToken cancellationToken = default
)
{
//
// Validate ...
if (!request.IsValid())
{
XException.InvalidArgs.Throw();
}
//
var result = await SendMessageAsync(
topic: request.Topic,
message: request.Message,
cancellationToken: cancellationToken
);
//
return result;
}
/// <summary>
/// Send Message to Broker Using XMessage instance ...
/// </summary>
/// <param name="message"></param>
/// <param name="topic"></param>
/// <param name="cancellationToken"></param>
/// <returns></returns>
public async Task<bool> SendMessageAsync(
XEventMessage message,
string topic = XEventServiceConstants.XDefaultTopic,
CancellationToken cancellationToken = default
)
{
//
// Validate Args ...
if (!message.IsValid())
{
XException.InvalidArgs.Throw();
}
//
var result = false;
switch (configuration.Broker)
{
//
case XEventBroker.XKafka:
result = await kafkaProducerService.SendMessageAsync(
topic: topic,
message: message,
cancellationToken: cancellationToken
);
break;
//
case XEventBroker.XZeroMQ:
result = await zeroMQProducerService.SendMessageAsync(
topic: topic,
message: message,
cancellationToken: cancellationToken
);
break;
//
case XEventBroker.XRabbitMQ:
result = await rabbitMQProducerService.SendMessageAsync(
topic: topic,
message: message,
cancellationToken: cancellationToken
);
break;
}
//
return result;
}
/// <summary>
/// Dispose ...
/// </summary>
public void Dispose()
{
//
if (kafkaProducerService != null)
{
kafkaProducerService.Dispose();
}
//
if (zeroMQProducerService != null)
{
zeroMQProducerService.Dispose();
}
//
if (rabbitMQProducerService != null)
{
rabbitMQProducerService.Dispose();
}
}
}
}
Providers/XKafkaConsumerServiceBase.cs
+8 -3
View File
@@ -74,6 +74,7 @@ namespace xEventService.Providers
//
// Subscribe to Topic ...
consumer.Subscribe(topic);
logger.LogInformation($"Kafka Cosume Topic: {topic} ...");
//
// Consuming ...
@@ -100,18 +101,22 @@ namespace xEventService.Providers
{
//
await DoWorkAsync(
payload: consumeResult.Message,
payload: message,
cancellationToken: cancellationToken
);
}
}
}
catch (ConsumeException)
catch (ConsumeException ex)
{
//
logger.LogError($"Kafka Consumer Result Parsing Error: {ex.Message} ...");
await Task.Delay(1000, cancellationToken);
}
catch (Exception)
catch (Exception ex)
{
//
logger.LogError($"Kafka Consumer Result Parsing Error: {ex.Message} ...");
await Task.Delay(1000, cancellationToken);
}
}
Providers/XKafkaProducerService.cs
+4 -2
View File
@@ -63,8 +63,8 @@ namespace xEventService.Providers
//
result = await SendMessageAsync(
message: request.Message,
topic: request.Topic,
message: request.Message,
cancellationToken: cancellationToken
);
@@ -111,8 +111,10 @@ namespace xEventService.Providers
//
result = !response.IsNullOrDefault();
}
catch
catch (Exception ex)
{
//
logger.LogError($"Kafka Produce Error: {ex.Message} ...");
result = false;
}
Providers/XRabbitMQConsumerServiceBase.cs
+23 -10
View File
@@ -51,7 +51,7 @@ namespace xEventService.Providers
//
SetupConnection();
DeclareTopology();
DeclareTopology(topic);
consumer = new AsyncEventingBasicConsumer(channel);
}
@@ -68,6 +68,7 @@ namespace xEventService.Providers
// For Fix Blocking Synchronous ...
await Task.Yield();
using var cts = new CancellationTokenSource();
logger.LogInformation($"RabbitMQ Cosume Topic: {topic} ...");
//
// Subscribe to Topic ...
@@ -103,9 +104,10 @@ namespace xEventService.Providers
);
}
}
catch
catch (Exception ex)
{
//
logger.LogError($"RabbitMQ Cosume Error: {ex.Message} ...");
if (configuration.RabbitMQ.ManualAck)
{
//
@@ -153,10 +155,20 @@ namespace xEventService.Providers
public override async Task StopAsync(CancellationToken cancellationToken)
{
//
channel.Close();
channel.Dispose();
connection.Close();
connection.Dispose();
if (channel != null)
{
//
channel.Close();
channel.Dispose();
}
//
if (connection != null)
{
//
connection.Close();
connection.Dispose();
}
//
await base.StopAsync(cancellationToken);
@@ -199,8 +211,9 @@ namespace xEventService.Providers
if (result.IsNullOrEmpty())
{
//
var exchangeName = topic ?? configuration.RabbitMQ.Exchange;
var digits = Guid.NewGuid().ToString().GetDigits();
result = $"{configuration.RabbitMQ.Exchange}[{digits}]";
result = $"{exchangeName}[{digits}]";
}
//
@@ -228,12 +241,12 @@ namespace xEventService.Providers
channel = connection.CreateModel();
}
private void DeclareTopology()
private void DeclareTopology(string topic)
{
//
channel.ExchangeDeclare(
arguments: null,
exchange: configuration.RabbitMQ.Exchange,
exchange: topic ?? configuration.RabbitMQ.Exchange,
durable: configuration.RabbitMQ.ExchangeDurable,
autoDelete: configuration.RabbitMQ.ExchangeAutoDelete,
type: configuration.RabbitMQ.ExchangeType.GetStringValue()
@@ -255,7 +268,7 @@ namespace xEventService.Providers
channel.QueueBind(
arguments: null,
queue: queueName,
exchange: configuration.RabbitMQ.Exchange,
exchange: topic ?? configuration.RabbitMQ.Exchange,
routingKey: configuration.RabbitMQ.RoutingKey
);
}
Providers/XRabbitMQProducerService.cs
+5 -3
View File
@@ -143,7 +143,7 @@ namespace xEventService.Providers
{
//
// Ensure Connection is Alive ...
EnsureChannelAsync();
EnsureChannel();
//
// Serialize Message ...
@@ -181,8 +181,10 @@ namespace xEventService.Providers
//
result = true;
}
catch
catch (Exception ex)
{
//
logger.LogError($"RabbitMQ Produce Error: {ex.Message} ...");
result = false;
}
@@ -268,7 +270,7 @@ namespace xEventService.Providers
/// <summary>
/// Ensure Channel is Open and Exchange/Queue Declared ...
/// </summary>
private void EnsureChannelAsync()
private void EnsureChannel()
{
//
if (IsConnected())
Providers/XZeroMQConsumerServiceBase.cs
+22 -3
View File
@@ -40,7 +40,8 @@ namespace xEventService.Providers
{
//
// Validating Configuration ...
if (configuration.Broker != XEventBroker.XKafka)
if (!configuration.ZeroMQ.IsValid() ||
configuration.Broker != XEventBroker.XZeroMQ)
{
XException.InvalidConfiguration.Throw();
}
@@ -61,6 +62,7 @@ namespace xEventService.Providers
{
//
await Task.Yield();
logger.LogInformation($"ZeroMQ Cosume Topic: {topic} ...");
while (!cancellationToken.IsCancellationRequested)
{
//
@@ -70,6 +72,7 @@ namespace xEventService.Providers
string receivedTopic = string.Empty;
byte[] body = null;
//
if (configuration.ZeroMQ.Pattern == XZeroMQPattern.PubSub)
{
//
@@ -80,6 +83,10 @@ namespace xEventService.Providers
{
// Process
}
else
{
logger.LogError($"ZeroMQ Cosume Error ...");
}
}
}
else
@@ -88,8 +95,13 @@ namespace xEventService.Providers
{
// Process
}
else
{
logger.LogError($"ZeroMQ Cosume Error ...");
}
}
//
if (body != null && body.Length > 0)
{
//
@@ -113,8 +125,10 @@ namespace xEventService.Providers
await Task.Delay(10, cancellationToken);
}
}
catch (Exception)
catch (Exception ex)
{
//
logger.LogError($"ZeroMQ Cosume Error: {ex.Message} ...");
await Task.Delay(1000, cancellationToken);
}
}
@@ -123,7 +137,12 @@ namespace xEventService.Providers
public override async Task StopAsync(CancellationToken cancellationToken)
{
//
socket.Close();
if (socket != null)
{
socket.Close();
}
//
await base.StopAsync(cancellationToken);
}