rama creada apartir de master en j
This commit is contained in:
jrojas
@@ -10,6 +10,9 @@ using Newtonsoft.Json;
|
||||
|
||||
namespace adas_core.Infrastructure.Services;
|
||||
|
||||
/// <summary>
|
||||
/// Provides functionality for receiving and processing incoming data, messages, or requests.
|
||||
/// </summary>
|
||||
public class ReceiverService
|
||||
{
|
||||
private readonly ILogger<ReceiverService> _logger;
|
||||
@@ -63,65 +66,78 @@ public class ReceiverService
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Registers all application queues defined in settings, including queues for observations, treatments, patients, pumps, appointments, recordings, recording alerts, and alarm observations, by adding each one through the <c>AddQueue</c> call.
|
||||
/// </summary>
|
||||
private void SetQueues()
|
||||
{
|
||||
AddQueue(_settings.ObservationsQueue);
|
||||
AddQueue(_settings.TreatmentsQueue);
|
||||
AddQueue(_settings.PatientsQueue);
|
||||
AddQueue(_settings.PumpsQueue);
|
||||
AddQueue(_settings.AppointmentsQueue);
|
||||
AddQueue(_settings.RecordingQueue);
|
||||
AddQueue(_settings.RecordingAlertQueue);
|
||||
AddQueue(_settings.AlarmObservationQueue);
|
||||
}
|
||||
{
|
||||
AddQueue(_settings.ObservationsQueue);
|
||||
AddQueue(_settings.TreatmentsQueue);
|
||||
AddQueue(_settings.PatientsQueue);
|
||||
AddQueue(_settings.PumpsQueue);
|
||||
AddQueue(_settings.AppointmentsQueue);
|
||||
AddQueue(_settings.RecordingQueue);
|
||||
AddQueue(_settings.RecordingAlertQueue);
|
||||
AddQueue(_settings.AlarmObservationQueue);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Adds a queue to the internal collection only when the supplied value is not null, empty, or whitespace; otherwise the call is ignored.
|
||||
/// </summary>
|
||||
/// <param name="queue">The queue identifier to add to the collection. Null, empty, or whitespace values are silently skipped.</param>
|
||||
private void AddQueue(string? queue)
|
||||
{
|
||||
if (!string.IsNullOrWhiteSpace(queue))
|
||||
_queues.Add(queue!);
|
||||
}
|
||||
{
|
||||
if (!string.IsNullOrWhiteSpace(queue))
|
||||
_queues.Add(queue!);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Attempts to establish a connection to RabbitMQ by disposing any existing bus, configuring EasyNetQ with the configured connection string, and registering consumers for all configured queues. If the connection fails, the failure is logged and a timer is scheduled to retry the connection after a five-second delay, recursively invoking the method when the timer elapses.
|
||||
/// </summary>
|
||||
private void TryToConnect()
|
||||
{
|
||||
try
|
||||
{
|
||||
DisposeBus();
|
||||
|
||||
var services = new ServiceCollection();
|
||||
services.AddEasyNetQ(_settings.ConnectionString);
|
||||
var provider = services.BuildServiceProvider();
|
||||
|
||||
_bus = provider.GetRequiredService<IBus>();
|
||||
|
||||
|
||||
foreach (var queue in _queues)
|
||||
try
|
||||
{
|
||||
RegisterConsumer(queue);
|
||||
DisposeBus();
|
||||
|
||||
var services = new ServiceCollection();
|
||||
services.AddEasyNetQ(_settings.ConnectionString);
|
||||
var provider = services.BuildServiceProvider();
|
||||
|
||||
_bus = provider.GetRequiredService<IBus>();
|
||||
|
||||
|
||||
foreach (var queue in _queues)
|
||||
{
|
||||
RegisterConsumer(queue);
|
||||
}
|
||||
|
||||
_logger.LogInformation("RabbitMQ connected. Registered {Count} queues", _queues.Count);
|
||||
}
|
||||
|
||||
_logger.LogInformation("RabbitMQ connected. Registered {Count} queues", _queues.Count);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, "RabbitMQ connection failed. Retrying in 5 seconds...");
|
||||
|
||||
var timer = new System.Timers.Timer(5000);
|
||||
timer.Elapsed += (_, _) =>
|
||||
catch (Exception ex)
|
||||
{
|
||||
timer.Stop();
|
||||
TryToConnect();
|
||||
};
|
||||
timer.Start();
|
||||
_logger.LogError(ex, "RabbitMQ connection failed. Retrying in 5 seconds...");
|
||||
|
||||
var timer = new System.Timers.Timer(5000);
|
||||
timer.Elapsed += (_, _) =>
|
||||
{
|
||||
timer.Stop();
|
||||
TryToConnect();
|
||||
};
|
||||
timer.Start();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Disposes the underlying bus instance if it implements <see cref="IDisposable"/>.
|
||||
/// </summary>
|
||||
private void DisposeBus()
|
||||
{
|
||||
if (_bus is IDisposable disposable)
|
||||
{
|
||||
disposable.Dispose();
|
||||
if (_bus is IDisposable disposable)
|
||||
{
|
||||
disposable.Dispose();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private void RegisterConsumer(string queue)
|
||||
{
|
||||
@@ -154,58 +170,67 @@ public class ReceiverService
|
||||
});
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Resolves and returns the <see cref="IApiRequestService"/> instance associated with the specified queue name by matching it against the configured queue settings for observations, treatments, patients, pumps, recordings, recording alerts, alarm observations, and appointments.
|
||||
/// </summary>
|
||||
/// <param name="queue">The queue name to resolve to its corresponding API request service.</param>
|
||||
/// <returns>The matching <see cref="IApiRequestService"/> instance if the queue is recognized; otherwise, <c>null</c>.</returns>
|
||||
private IApiRequestService? ResolveService(string queue)
|
||||
{
|
||||
return queue switch
|
||||
{
|
||||
var q when q == _settings.ObservationsQueue => _observationService,
|
||||
var q when q == _settings.TreatmentsQueue => _treatmentService,
|
||||
var q when q == _settings.PatientsQueue => _patientsService,
|
||||
var q when q == _settings.PumpsQueue => _pumpService,
|
||||
var q when q == _settings.RecordingQueue => _recordingService,
|
||||
var q when q == _settings.RecordingAlertQueue => _recordingAlertService,
|
||||
var q when q == _settings.AlarmObservationQueue => _alarmService,
|
||||
var q when q == _settings.AppointmentsQueue => _appointmentService,
|
||||
_ => null
|
||||
};
|
||||
}
|
||||
|
||||
public async Task ProcessErrorQueue(string errorQueueName)
|
||||
{
|
||||
if (_bus == null)
|
||||
{
|
||||
_logger.LogError("Cannot process error queue - bus is null");
|
||||
return;
|
||||
return queue switch
|
||||
{
|
||||
var q when q == _settings.ObservationsQueue => _observationService,
|
||||
var q when q == _settings.TreatmentsQueue => _treatmentService,
|
||||
var q when q == _settings.PatientsQueue => _patientsService,
|
||||
var q when q == _settings.PumpsQueue => _pumpService,
|
||||
var q when q == _settings.RecordingQueue => _recordingService,
|
||||
var q when q == _settings.RecordingAlertQueue => _recordingAlertService,
|
||||
var q when q == _settings.AlarmObservationQueue => _alarmService,
|
||||
var q when q == _settings.AppointmentsQueue => _appointmentService,
|
||||
_ => null
|
||||
};
|
||||
}
|
||||
|
||||
await _bus.SendReceive.ReceiveAsync<Error>(errorQueueName, async err =>
|
||||
/// <summary>
|
||||
/// Asynchronously reprocesses messages from the specified error queue by deriving the target queue name from the routing key (with "Key" removed), resolving the associated service, and resubmitting the message. Operations are skipped when the bus is uninitialized, the derived queue name is empty, or no service is resolved.
|
||||
/// </summary>
|
||||
/// <param name="errorQueueName">The name of the error queue from which to reprocess messages.</param>
|
||||
public async Task ProcessErrorQueue(string errorQueueName)
|
||||
{
|
||||
try
|
||||
if (_bus == null)
|
||||
{
|
||||
var queue = err.RoutingKey?.Replace("Key", "");
|
||||
|
||||
if (string.IsNullOrEmpty(queue))
|
||||
return;
|
||||
|
||||
var service = ResolveService(queue);
|
||||
|
||||
if (service == null)
|
||||
return;
|
||||
|
||||
var message = new Message<string>(
|
||||
err.Message,
|
||||
new MessageProperties()
|
||||
);
|
||||
|
||||
await Task.Run(() => TryToParseAndSend(message, service));
|
||||
_logger.LogError("Cannot process error queue - bus is null");
|
||||
return;
|
||||
}
|
||||
catch (Exception ex)
|
||||
|
||||
await _bus.SendReceive.ReceiveAsync<Error>(errorQueueName, async err =>
|
||||
{
|
||||
_logger.LogError(ex, "Error reprocessing message from {queue}", errorQueueName);
|
||||
throw;
|
||||
}
|
||||
});
|
||||
}
|
||||
try
|
||||
{
|
||||
var queue = err.RoutingKey?.Replace("Key", "");
|
||||
|
||||
if (string.IsNullOrEmpty(queue))
|
||||
return;
|
||||
|
||||
var service = ResolveService(queue);
|
||||
|
||||
if (service == null)
|
||||
return;
|
||||
|
||||
var message = new Message<string>(
|
||||
err.Message,
|
||||
new MessageProperties()
|
||||
);
|
||||
|
||||
await Task.Run(() => TryToParseAndSend(message, service));
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, "Error reprocessing message from {queue}", errorQueueName);
|
||||
throw;
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Sends EasyNetQ messages to a service implementing IApiRequestService
|
||||
|
||||
Reference in New Issue
Block a user