79 lines
2.5 KiB
C#
79 lines
2.5 KiB
C#
using System.Text.Json;
|
|
using Femto.Modules.Blog.Data;
|
|
using Femto.Modules.Blog.Infrastructure.Integration;
|
|
using MediatR;
|
|
using Microsoft.Extensions.Logging;
|
|
using Quartz;
|
|
|
|
namespace Femto.Modules.Blog;
|
|
|
|
[DisallowConcurrentExecution]
|
|
internal class MailmanJob(
|
|
Outbox outbox,
|
|
BlogContext context,
|
|
ILogger<MailmanJob> logger,
|
|
IMediator mediator
|
|
) : IJob
|
|
{
|
|
public async Task Execute(IJobExecutionContext executionContext)
|
|
{
|
|
try
|
|
{
|
|
var messages = await outbox.GetPendingMessages(executionContext.CancellationToken);
|
|
|
|
logger.LogTrace("loaded {Count} outbox messages to process", messages.Count);
|
|
foreach (var message in messages)
|
|
{
|
|
try
|
|
{
|
|
var notificationType = OutboxMessageTypeRegistry.GetType(message.EventType);
|
|
|
|
if (notificationType is null)
|
|
{
|
|
logger.LogWarning(
|
|
"unmapped event type {Type}. skipping.",
|
|
message.EventType
|
|
);
|
|
continue;
|
|
}
|
|
|
|
var notification =
|
|
JsonSerializer.Deserialize(message.Payload, notificationType)
|
|
as INotification;
|
|
|
|
if (notification is null)
|
|
throw new Exception("notification is null");
|
|
|
|
logger.LogTrace(
|
|
"publishing outbox message {EventType}. Id: {Id}, AggregateId: {AggregateId}",
|
|
message.EventType,
|
|
message.Id,
|
|
message.AggregateId
|
|
);
|
|
|
|
await mediator.Publish(notification, executionContext.CancellationToken);
|
|
|
|
message.Succeed();
|
|
}
|
|
catch (Exception e)
|
|
{
|
|
logger.LogError(
|
|
e,
|
|
"Error processing event {EventId} for aggregate {AggregateId}",
|
|
message.Id,
|
|
message.AggregateId
|
|
);
|
|
|
|
message.Fail(e.ToString());
|
|
}
|
|
|
|
await context.SaveChangesAsync(executionContext.CancellationToken);
|
|
}
|
|
}
|
|
catch (Exception e)
|
|
{
|
|
logger.LogError(e, "Error while processing outbox");
|
|
throw;
|
|
}
|
|
}
|
|
}
|