112 lines
4.4 KiB
C#
112 lines
4.4 KiB
C#
using System.Text.Json;
|
|
using Confluent.Kafka;
|
|
using Microsoft.Extensions.Options;
|
|
|
|
public class SepsisEngineService : BackgroundService
|
|
{
|
|
private static readonly JsonSerializerOptions EventJsonOptions = new()
|
|
{
|
|
PropertyNameCaseInsensitive = true
|
|
};
|
|
|
|
private readonly IServiceProvider _services;
|
|
private readonly KafkaOptions _kafkaOptions;
|
|
private readonly ILogger<SepsisEngineService> _logger;
|
|
|
|
public SepsisEngineService(
|
|
IServiceProvider services,
|
|
IOptions<KafkaOptions> kafkaOptions,
|
|
ILogger<SepsisEngineService> logger)
|
|
{
|
|
_services = services;
|
|
_kafkaOptions = kafkaOptions.Value;
|
|
_logger = logger;
|
|
}
|
|
|
|
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
|
|
{
|
|
var config = new ConsumerConfig
|
|
{
|
|
BootstrapServers = _kafkaOptions.BootstrapServers,
|
|
GroupId = "sepsis-engine",
|
|
AutoOffsetReset = AutoOffsetReset.Earliest,
|
|
EnableAutoCommit = false
|
|
};
|
|
|
|
using var consumer = new ConsumerBuilder<string, string>(config).Build();
|
|
// Subscribes to observation.recorded only.
|
|
// The es-indexer consumes all three topics; the sepsis engine only needs one.
|
|
// Subscribing to a superset of needed topics would waste CPU deserializing
|
|
// alert and encounter events that this engine discards immediately.
|
|
consumer.Subscribe(_kafkaOptions.Topics.ObservationRecorded);
|
|
|
|
_logger.LogInformation("SepsisEngineService started — consumer group: sepsis-engine");
|
|
|
|
try
|
|
{
|
|
while (!stoppingToken.IsCancellationRequested)
|
|
{
|
|
ConsumeResult<string, string>? result = null;
|
|
try
|
|
{
|
|
result = consumer.Consume(stoppingToken);
|
|
|
|
var evt = JsonSerializer.Deserialize<SepsisObservationEvent>(
|
|
result.Message.Value, EventJsonOptions)!;
|
|
|
|
// Create a scope per message — both detectors are scoped and
|
|
// each owns a fresh DbContext when creating alerts.
|
|
using var scope = _services.CreateScope();
|
|
var sirsDetector = scope.ServiceProvider.GetRequiredService<SirsDetector>();
|
|
var qsofaDetector = scope.ServiceProvider.GetRequiredService<QsofaDetector>();
|
|
|
|
var sirsOutcome = await sirsDetector.ProcessObservationAsync(
|
|
evt.EncounterId,
|
|
evt.PatientId,
|
|
evt.ObservationCode,
|
|
evt.Value,
|
|
stoppingToken);
|
|
|
|
var qsofaOutcome = await qsofaDetector.ProcessObservationAsync(
|
|
evt.EncounterId,
|
|
evt.PatientId,
|
|
evt.ObservationCode,
|
|
evt.Value,
|
|
stoppingToken);
|
|
|
|
if (sirsOutcome.Outcome == SirsOutcome.AlertCreated)
|
|
_logger.LogWarning(
|
|
"SEPSIS_WARNING created via SepsisEngine " +
|
|
"— encounter={EncounterId} code={Code} value={Value}",
|
|
evt.EncounterId, evt.ObservationCode, evt.Value);
|
|
|
|
if (qsofaOutcome.Outcome == QsofaOutcome.AlertCreated)
|
|
_logger.LogWarning(
|
|
"QSOFA_WARNING created via SepsisEngine " +
|
|
"— encounter={EncounterId} code={Code} value={Value}",
|
|
evt.EncounterId, evt.ObservationCode, evt.Value);
|
|
|
|
// Commit only after successful processing.
|
|
consumer.Commit(result);
|
|
}
|
|
catch (OperationCanceledException)
|
|
{
|
|
break;
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger.LogError(ex,
|
|
"SepsisEngine failed on topic={Topic} offset={Offset} — not committing",
|
|
result?.Topic, result?.Offset.Value);
|
|
// Back off before retrying so a persistent failure (e.g., Redis down)
|
|
// does not spin the loop at maximum throughput.
|
|
await Task.Delay(2000, stoppingToken);
|
|
}
|
|
}
|
|
}
|
|
finally
|
|
{
|
|
consumer.Close();
|
|
}
|
|
}
|
|
} |