azaion/annotations
1
0
Fork 0
mirror of https://github.com/azaion/annotations.git synced 2026-04-22 10:46:30 +00:00
Code Issues Packages Projects Releases Wiki Activity

log queue errors

Browse Source
This commit is contained in:
Alex Bezdieniezhnykh
2025-05-17 19:38:07 +03:00
parent dae342b70e
commit cf563571c8
1 changed files with 43 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Azaion.Common/Services/AnnotationService.cs
+43 -32
View File
@@ -13,6 +13,7 @@ using LinqToDB;
using LinqToDB.Data;
using MediatR;
using MessagePack;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using Newtonsoft.Json;
using RabbitMQ.Stream.Client;
@@ -27,6 +28,7 @@ public class AnnotationService : IAnnotationService, INotificationHandler<Annota
private readonly IGalleryService _galleryService;
private readonly IMediator _mediator;
private readonly IAzaionApi _api;
private readonly ILogger<AnnotationService> _logger;
private readonly QueueConfig _queueConfig;
private Consumer _consumer = null!;
private readonly UIConfig _uiConfig;
@@ -42,13 +44,15 @@ public class AnnotationService : IAnnotationService, INotificationHandler<Annota
IOptions<DirectoriesConfig> directoriesConfig,
IGalleryService galleryService,
IMediator mediator,
IAzaionApi api)
IAzaionApi api,
ILogger<AnnotationService> logger)
{
_dbFactory = dbFactory;
_producer = producer;
_galleryService = galleryService;
_mediator = mediator;
_api = api;
_logger = logger;
_queueConfig = queueConfig.Value;
_uiConfig = uiConfig.Value;
_dirConfig = directoriesConfig.Value;
@@ -76,40 +80,47 @@ public class AnnotationService : IAnnotationService, INotificationHandler<Annota
OffsetSpec = new OffsetTypeOffset(offsets.AnnotationsOffset + 1),
MessageHandler = async (_, _, context, message) =>
{
var email = (string)message.ApplicationProperties[nameof(User.Email)]!;
if (email == _api.CurrentUser.Email) //Don't process messages by yourself
return;
var annotationStatus = (AnnotationStatus)message.ApplicationProperties[nameof(AnnotationStatus)];
if (annotationStatus.In(AnnotationStatus.Created, AnnotationStatus.ValidatedEdited))
try
{
var msg = MessagePackSerializer.Deserialize<AnnotationMessage>(message.Data.Contents);
await SaveAnnotationInner(
msg.CreatedDate,
msg.OriginalMediaName,
msg.Time,
JsonConvert.DeserializeObject<List<Detection>>(msg.Detections) ?? [],
msg.Source,
new MemoryStream(msg.Image),
msg.Role,
msg.Email,
fromQueue: true,
token: cancellationToken);
}
else
{
var msg = MessagePackSerializer.Deserialize<AnnotationBulkMessage>(message.Data.Contents);
if (annotationStatus == AnnotationStatus.Validated)
await ValidateAnnotations(msg.AnnotationNames.ToList(), true, cancellationToken);
if (annotationStatus == AnnotationStatus.Deleted)
await _mediator.Publish(new AnnotationsDeletedEvent(msg.AnnotationNames.ToList(), fromQueue:true), cancellationToken);
}
var email = (string)message.ApplicationProperties[nameof(User.Email)]!;
if (email == _api.CurrentUser.Email) //Don't process messages by yourself
return;
var annotationStatus = (AnnotationStatus)message.ApplicationProperties[nameof(AnnotationStatus)];
if (annotationStatus.In(AnnotationStatus.Created, AnnotationStatus.ValidatedEdited))
{
var msg = MessagePackSerializer.Deserialize<AnnotationMessage>(message.Data.Contents);
await SaveAnnotationInner(
msg.CreatedDate,
msg.OriginalMediaName,
msg.Time,
JsonConvert.DeserializeObject<List<Detection>>(msg.Detections) ?? [],
msg.Source,
new MemoryStream(msg.Image),
msg.Role,
msg.Email,
fromQueue: true,
token: cancellationToken);
}
else
{
var msg = MessagePackSerializer.Deserialize<AnnotationBulkMessage>(message.Data.Contents);
if (annotationStatus == AnnotationStatus.Validated)
await ValidateAnnotations(msg.AnnotationNames.ToList(), true, cancellationToken);
if (annotationStatus == AnnotationStatus.Deleted)
await _mediator.Publish(new AnnotationsDeletedEvent(msg.AnnotationNames.ToList(), fromQueue:true), cancellationToken);
}
offsets.AnnotationsOffset = context.Offset;
ThrottleExt.Throttle(() =>
offsets.AnnotationsOffset = context.Offset;
ThrottleExt.Throttle(() =>
{
_api.UpdateOffsets(offsets);
return Task.CompletedTask;
}, SaveQueueOffsetTaskId, TimeSpan.FromSeconds(10), scheduleCallAfterCooldown: true);
}
catch (Exception e)
{
_api.UpdateOffsets(offsets);
return Task.CompletedTask;
}, SaveQueueOffsetTaskId, TimeSpan.FromSeconds(10), scheduleCallAfterCooldown: true);
_logger.LogError(e, e.Message);
}
}
});
}