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

fixed bugs with queue handling. At least most of them

Browse Source
This commit is contained in:
Alex Bezdieniezhnykh
2025-05-18 20:11:19 +03:00
parent cf563571c8
commit c5e81ebcc6
15 changed files with 135 additions and 124 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Azaion.Common/Services/AnnotationService.cs
+35 -41
View File
@@ -77,40 +77,43 @@ public class AnnotationService : IAnnotationService, INotificationHandler<Annota
_consumer = await Consumer.Create(new ConsumerConfig(consumerSystem, Constants.MQ_ANNOTATIONS_QUEUE)
{
Reference = _api.CurrentUser.Email,
OffsetSpec = new OffsetTypeOffset(offsets.AnnotationsOffset + 1),
OffsetSpec = new OffsetTypeOffset(offsets.AnnotationsOffset),
MessageHandler = async (_, _, context, message) =>
{
try
{
var email = (string)message.ApplicationProperties[nameof(User.Email)]!;
if (email == _api.CurrentUser.Email) //Don't process messages by yourself
if (!Enum.TryParse<AnnotationStatus>((string)message.ApplicationProperties[nameof(AnnotationStatus)], out var annotationStatus))
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;
if (email != _api.CurrentUser.Email) //Don't process messages by yourself
{
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 + 1; //to consume on the next launch from the next message
ThrottleExt.Throttle(() =>
{
_api.UpdateOffsets(offsets);
@@ -154,12 +157,6 @@ public class AnnotationService : IAnnotationService, INotificationHandler<Annota
.LoadWith(x => x.Detections)
.FirstOrDefaultAsync(x => x.Name == fName, token: token);
if (userRole.IsValidator() && source == SourceEnum.Manual)
status = AnnotationStatus.Validated;
if (fromQueue && ann is { AnnotationStatus: AnnotationStatus.Validated })
return ann;
await db.Detections.DeleteAsync(x => x.AnnotationName == fName, token: token);
if (ann != null) //Annotation is already exists
@@ -169,10 +166,9 @@ public class AnnotationService : IAnnotationService, INotificationHandler<Annota
.Set(x => x.Source, source)
.Set(x => x.CreatedRole, userRole);
if (status == AnnotationStatus.Validated)
if (userRole.IsValidator() && source == SourceEnum.Manual)
{
if (status == AnnotationStatus.Validated)
status = AnnotationStatus.ValidatedEdited; //For further processing mark Annotations *edited* by Validator, not just simply Validated by button.
status = AnnotationStatus.ValidatedEdited;
annotationUpdatable = annotationUpdatable
.Set(x => x.ValidateDate, createdDate)
@@ -212,9 +208,6 @@ public class AnnotationService : IAnnotationService, INotificationHandler<Annota
return ann;
});
if (fromQueue && annotation is { AnnotationStatus: AnnotationStatus.Validated })
return annotation;
if (stream != null)
{
var img = System.Drawing.Image.FromStream(stream);
@@ -228,7 +221,7 @@ public class AnnotationService : IAnnotationService, INotificationHandler<Annota
await _mediator.Publish(new AnnotationCreatedEvent(annotation), token);
if (!fromQueue && !_uiConfig.SilentDetection) //Send to queue only if we're not getting from queue already
if (!fromQueue) //Send to queue only if we're not getting from queue already
await _producer.SendToInnerQueue([annotation.Name], status, token);
ThrottleExt.Throttle(async () =>
@@ -275,7 +268,8 @@ public class AnnotationService : IAnnotationService, INotificationHandler<Annota
File.Delete(Path.Combine(_dirConfig.ResultsDirectory, $"{name}{Constants.RESULT_PREFIX}{Constants.JPG_EXT}"));
}
if (!notification.FromQueue)
//Only validators can send Delete to the queue
if (!notification.FromQueue && _api.CurrentUser.Role.IsValidator())
await _producer.SendToInnerQueue(notification.AnnotationNames, AnnotationStatus.Deleted, ct);
ThrottleExt.Throttle(async () =>