Compare commits
13 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| db4c348288 | |||
| a65e926a53 | |||
| 6683234642 | |||
| 1471d9e597 | |||
| 6ddbb9479a | |||
| e978aa17f8 | |||
| 6c7349a0f8 | |||
| a31f75c5ef | |||
| e97fd6cd3f | |||
| 7f4246ac63 | |||
| 184c7c252a | |||
| dfc62c8fe1 | |||
| bfe8c458d6 |
@@ -1,4 +1,5 @@
|
|||||||
using LiteCharms.Features.Shop.Notifications;
|
using LiteCharms.Features.Shop.Notifications;
|
||||||
|
using LiteCharms.Features.Shop.Notifications.Events;
|
||||||
|
|
||||||
namespace LiteCharms.Features.Tests;
|
namespace LiteCharms.Features.Tests;
|
||||||
|
|
||||||
@@ -32,4 +33,14 @@ public class NotificationsFeatureTests(CommonFixture fixture, ITestOutputHelper
|
|||||||
|
|
||||||
foreach (var error in createResult.Errors) output.WriteLine(error.Message);
|
foreach (var error in createResult.Errors) output.WriteLine(error.Message);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task ProcessEmailNotificationsEvent_ShouldSucceed()
|
||||||
|
{
|
||||||
|
var notification = ProcessEmailNotificationsEvent.Create();
|
||||||
|
|
||||||
|
await fixture.Mediator.Publish(notification);
|
||||||
|
|
||||||
|
Assert.True(true);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -35,8 +35,9 @@ public class EmailService(IOptions<SmtpSettings> options) : IDisposable
|
|||||||
|
|
||||||
var bodyBuilder = new BodyBuilder();
|
var bodyBuilder = new BodyBuilder();
|
||||||
|
|
||||||
foreach (var attachment in message.Body?.Attachments!)
|
if (message.Body!.Properties.HasAttachments)
|
||||||
bodyBuilder.Attachments.Add(attachment.Name!, attachment.FileStream!, cancellationToken);
|
foreach (var attachment in message.Body?.Attachments!)
|
||||||
|
bodyBuilder.Attachments.Add(attachment.Name!, attachment.FileStream!, cancellationToken);
|
||||||
|
|
||||||
if (!message.Body.Properties.IsHtml) bodyBuilder.TextBody = message.Body.Message;
|
if (!message.Body.Properties.IsHtml) bodyBuilder.TextBody = message.Body.Message;
|
||||||
if (message.Body.Properties.IsHtml) bodyBuilder.HtmlBody = message.Body.Message;
|
if (message.Body.Properties.IsHtml) bodyBuilder.HtmlBody = message.Body.Message;
|
||||||
|
|||||||
@@ -34,7 +34,7 @@ public static class Quartz
|
|||||||
storage.UseClustering(cluster =>
|
storage.UseClustering(cluster =>
|
||||||
{
|
{
|
||||||
cluster.CheckinInterval = TimeSpan.FromSeconds(30);
|
cluster.CheckinInterval = TimeSpan.FromSeconds(30);
|
||||||
cluster.CheckinMisfireThreshold = TimeSpan.FromSeconds(20);
|
cluster.CheckinMisfireThreshold = TimeSpan.FromSeconds(90);
|
||||||
});
|
});
|
||||||
});
|
});
|
||||||
});
|
});
|
||||||
@@ -62,6 +62,8 @@ public static class Quartz
|
|||||||
config.UseDefaultThreadPool(options => options.MaxConcurrency = 1);
|
config.UseDefaultThreadPool(options => options.MaxConcurrency = 1);
|
||||||
config.UseTimeZoneConverter();
|
config.UseTimeZoneConverter();
|
||||||
|
|
||||||
|
config.SetProperty("quartz.jobStore.misfireThreshold", TimeSpan.FromMinutes(2).TotalMilliseconds.ToString());
|
||||||
|
|
||||||
config.UsePersistentStore(storage =>
|
config.UsePersistentStore(storage =>
|
||||||
{
|
{
|
||||||
storage.PerformSchemaValidation = false;
|
storage.PerformSchemaValidation = false;
|
||||||
@@ -74,7 +76,7 @@ public static class Quartz
|
|||||||
storage.UseClustering(cluster =>
|
storage.UseClustering(cluster =>
|
||||||
{
|
{
|
||||||
cluster.CheckinInterval = TimeSpan.FromSeconds(30);
|
cluster.CheckinInterval = TimeSpan.FromSeconds(30);
|
||||||
cluster.CheckinMisfireThreshold = TimeSpan.FromSeconds(20);
|
cluster.CheckinMisfireThreshold = TimeSpan.FromSeconds(90);
|
||||||
});
|
});
|
||||||
});
|
});
|
||||||
});
|
});
|
||||||
|
|||||||
@@ -50,9 +50,9 @@ public class JobOrchestrator(ISchedulerFactory schedulerFactory) : IJobOrchestra
|
|||||||
|
|
||||||
var trigger = global::Quartz.TriggerBuilder.Create()
|
var trigger = global::Quartz.TriggerBuilder.Create()
|
||||||
.WithIdentity(triggerKey)
|
.WithIdentity(triggerKey)
|
||||||
.WithDescription($"Scheduled via Main Job at {now:g} UTC")
|
.WithDescription($"Scheduled via Main Job at {now:g}")
|
||||||
.WithCronSchedule(cronExpression, cron => cron
|
.WithCronSchedule(cronExpression, cron => cron
|
||||||
.WithMisfireHandlingInstructionFireAndProceed())
|
.WithMisfireHandlingInstructionIgnoreMisfires())
|
||||||
.StartAt(now)
|
.StartAt(now)
|
||||||
.Build();
|
.Build();
|
||||||
|
|
||||||
|
|||||||
@@ -10,16 +10,28 @@ public class MediatorJob<TNotification>(IMediator mediator) : IJob where TNotifi
|
|||||||
{
|
{
|
||||||
var data = context.MergedJobDataMap["Payload"] as string;
|
var data = context.MergedJobDataMap["Payload"] as string;
|
||||||
|
|
||||||
if (string.IsNullOrWhiteSpace(data)) return;
|
if (string.IsNullOrWhiteSpace(data))
|
||||||
|
{
|
||||||
|
Trace.WriteLine("Job Payload missing, job ended");
|
||||||
|
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
var notification = JsonSerializer.Deserialize<TNotification>(data);
|
var notification = JsonSerializer.Deserialize<TNotification>(data);
|
||||||
|
|
||||||
if (notification is null) return;
|
if (notification is null)
|
||||||
|
{
|
||||||
|
Trace.WriteLine("Notification could not be JSon converted from data string, job ended");
|
||||||
|
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
using var activity = MediatorTelemetry.Source.StartActivity($"Quartz: {typeof(TNotification).Name}");
|
using var activity = MediatorTelemetry.Source.StartActivity($"Quartz: {typeof(TNotification).Name}");
|
||||||
|
|
||||||
activity?.SetTag("event.correlation_id", notification.CorrelationId);
|
activity?.SetTag("event.correlation_id", notification.CorrelationId);
|
||||||
|
|
||||||
await mediator.Publish(notification, context.CancellationToken);
|
await mediator.Publish(notification, context.CancellationToken);
|
||||||
|
|
||||||
|
Trace.WriteLine("Job published");
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
+13
-5
@@ -1,4 +1,5 @@
|
|||||||
using LiteCharms.Features.Email;
|
using k8s.KubeConfigModels;
|
||||||
|
using LiteCharms.Features.Email;
|
||||||
using LiteCharms.Features.Shop.Notifications.Models;
|
using LiteCharms.Features.Shop.Notifications.Models;
|
||||||
using LiteCharms.Features.Shop.Postgres;
|
using LiteCharms.Features.Shop.Postgres;
|
||||||
|
|
||||||
@@ -15,17 +16,20 @@ public class ProcessEmailNotificationsEventHandler(IDbContextFactory<ShopDbConte
|
|||||||
{
|
{
|
||||||
using var context = await contextFactory.CreateDbContextAsync(cancellationToken);
|
using var context = await contextFactory.CreateDbContextAsync(cancellationToken);
|
||||||
|
|
||||||
|
if (emailService.Status != EmailStatuses.Connected)
|
||||||
|
await emailService.ConnectAsync(cancellationToken);
|
||||||
|
|
||||||
var notifications = await context.Notifications
|
var notifications = await context.Notifications
|
||||||
.OrderByDescending(o => o.CreatedAt)
|
.OrderByDescending(o => o.CreatedAt)
|
||||||
.ThenBy(o => o.Priority)
|
.ThenBy(o => o.Priority)
|
||||||
.Where(n => n.CorrelationIdType == CorrelationIdTypes.Email)
|
.Where(n => n.Platform == NotificationPlatforms.Email &&
|
||||||
.Where(n => n.Direction == NotificationDirection.Outgoing)
|
n.Direction == NotificationDirection.Outgoing && n.Processed == false)
|
||||||
.Take(message.MaxRecords)
|
.Take(message.MaxRecords)
|
||||||
.ToListAsync(cancellationToken);
|
.ToListAsync(cancellationToken);
|
||||||
|
|
||||||
foreach (var notification in notifications)
|
foreach (var notification in notifications)
|
||||||
{
|
{
|
||||||
if (dropBatch || cancellationToken.IsCancellationRequested) break;
|
if (dropBatch) break;
|
||||||
|
|
||||||
var sendResult = await SendEmailAsync(notification,emailService, cancellationToken);
|
var sendResult = await SendEmailAsync(notification,emailService, cancellationToken);
|
||||||
|
|
||||||
@@ -46,12 +50,16 @@ public class ProcessEmailNotificationsEventHandler(IDbContextFactory<ShopDbConte
|
|||||||
notification.UpdatedAt = DateTime.UtcNow;
|
notification.UpdatedAt = DateTime.UtcNow;
|
||||||
}
|
}
|
||||||
|
|
||||||
await context.SaveChangesAsync(cancellationToken);
|
await context.SaveChangesAsync(cancellationToken);
|
||||||
}
|
}
|
||||||
catch (Exception ex)
|
catch (Exception ex)
|
||||||
{
|
{
|
||||||
logger.LogError(ex, ex.Message);
|
logger.LogError(ex, ex.Message);
|
||||||
}
|
}
|
||||||
|
finally
|
||||||
|
{
|
||||||
|
await emailService.DisconnectAsync(cancellationToken);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
private async Task<Result> SendEmailAsync(Notification notification, EmailService service, CancellationToken cancellationToken = default)
|
private async Task<Result> SendEmailAsync(Notification notification, EmailService service, CancellationToken cancellationToken = default)
|
||||||
|
|||||||
@@ -8,6 +8,8 @@ public class ProcessEmailNotificationsEvent : EventBase, IEvent
|
|||||||
|
|
||||||
public int MaxRecords { get; set; }
|
public int MaxRecords { get; set; }
|
||||||
|
|
||||||
|
public ProcessEmailNotificationsEvent() { MaxRecords = 1000; }
|
||||||
|
|
||||||
private ProcessEmailNotificationsEvent(int maxRecords = 1000) => MaxRecords = maxRecords;
|
private ProcessEmailNotificationsEvent(int maxRecords = 1000) => MaxRecords = maxRecords;
|
||||||
|
|
||||||
public static ProcessEmailNotificationsEvent Create(int maxRecords = 1000) => new(maxRecords);
|
public static ProcessEmailNotificationsEvent Create(int maxRecords = 1000) => new(maxRecords);
|
||||||
|
|||||||
Reference in New Issue
Block a user