queue + local sqlite WIP

2026-04-22 10:56:31 +00:00 · 2024-12-17 18:46:33 +02:00
parent 626767469a
commit 5fa18aa514
47 changed files with 694 additions and 222 deletions
@@ -0,0 +1,125 @@
+using System.IO;
+using System.Net;
+using Azaion.Common.Database;
+using Azaion.Common.DTO;
+using Azaion.Common.DTO.Config;
+using Azaion.Common.DTO.Queue;
+using LinqToDB;
+using MessagePack;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Options;
+using RabbitMQ.Stream.Client;
+using RabbitMQ.Stream.Client.Reliable;
+
+namespace Azaion.Common.Services;
+
+public class FailsafeAnnotationsProducer
+{
+    private readonly ILogger<FailsafeAnnotationsProducer> _logger;
+    private readonly IDbFactory _dbFactory;
+    private readonly QueueConfig _queueConfig;
+
+    private Producer _annotationProducer = null!;
+    private Producer _annotationConfirmProducer = null!;
+
+
+    public FailsafeAnnotationsProducer(ILogger<FailsafeAnnotationsProducer> logger, IDbFactory dbFactory, IOptions<QueueConfig> queueConfig)
+    {
+        _logger = logger;
+        _dbFactory = dbFactory;
+        _queueConfig = queueConfig.Value;
+        Task.Run(async () => await ProcessQueue()).Wait();
+    }
+
+    private async Task<StreamSystem> GetProducerQueueConfig()
+    {
+        return await StreamSystem.Create(new StreamSystemConfig
+        {
+
+            Endpoints = new List<EndPoint> { new IPEndPoint(IPAddress.Parse(_queueConfig.Host), _queueConfig.Port) },
+            UserName = _queueConfig.ProducerUsername,
+            Password = _queueConfig.ProducerPassword
+        });
+    }
+
+    private async Task Init(CancellationToken cancellationToken = default)
+    {
+        _annotationProducer = await Producer.Create(new ProducerConfig(await GetProducerQueueConfig(), Constants.MQ_ANNOTATIONS_QUEUE));
+        //_annotationConfirmProducer = await Producer.Create(new ProducerConfig(await GetProducerQueueConfig(), Constants.MQ_ANNOTATIONS_CONFIRM_QUEUE));
+    }
+
+    private async Task ProcessQueue(CancellationToken cancellationToken = default)
+    {
+        await Init(cancellationToken);
+        while (!cancellationToken.IsCancellationRequested)
+        {
+            var messages = await GetFromQueue(cancellationToken);
+            foreach (var messagesChunk in messages.Chunk(10)) //Sending by 10
+            {
+                var sent = false;
+                while (!sent || cancellationToken.IsCancellationRequested) //Waiting for send
+                {
+                    try
+                    {
+                        var createdMessages = messagesChunk
+                            .Where(x => x.Status == AnnotationStatus.Created)
+                            .Select(x => new Message(MessagePackSerializer.Serialize(x)))
+                            .ToList();
+                        await _annotationProducer.Send(createdMessages, CompressionType.Gzip);
+
+                        var validatedMessages = messagesChunk
+                            .Where(x => x.Status == AnnotationStatus.Validated)
+                            .Select(x => new Message(MessagePackSerializer.Serialize(x)))
+                            .ToList();
+                        await _annotationConfirmProducer.Send(validatedMessages, CompressionType.Gzip);
+
+                        await _dbFactory.Run(async db =>
+                            await db.AnnotationsQueue.DeleteAsync(aq => messagesChunk.Any(x => aq.Name == x.Name), token: cancellationToken));
+                        sent = true;
+                    }
+                    catch (Exception e)
+                    {
+                        _logger.LogError(e, e.Message);
+                        await Task.Delay(TimeSpan.FromSeconds(30), cancellationToken);
+                    }
+                }
+            }
+        }
+    }
+
+    private async Task<List<AnnotationCreatedMessage>> GetFromQueue(CancellationToken cancellationToken = default)
+    {
+        return await _dbFactory.Run(async db =>
+        {
+            var annotations = await db.AnnotationsQueue.Join(db.Annotations, aq => aq.Name, a => a.Name, (aq, a) => a)
+                .ToListAsync(token: cancellationToken);
+
+            var messages = new List<AnnotationCreatedMessage>();
+            foreach (var annotation in annotations)
+            {
+                var image = await File.ReadAllBytesAsync(annotation.ImagePath, cancellationToken);
+                var label = await File.ReadAllTextAsync(annotation.LabelPath, cancellationToken);
+                var annCreateMessage = new AnnotationCreatedMessage
+                {
+                    Name = annotation.Name,
+
+                    CreatedRole = annotation.CreatedRole,
+                    CreatedEmail = annotation.CreatedEmail,
+                    CreatedDate = annotation.CreatedDate,
+
+                    Image = image,
+                    Label = label,
+                    Source = annotation.Source
+                };
+                messages.Add(annCreateMessage);
+            }
+            return messages;
+        });
+    }
+
+    public async Task SendToQueue(Annotation annotation, CancellationToken cancellationToken = default)
+    {
+        await _dbFactory.Run(async db =>
+            await db.InsertAsync(new AnnotationName { Name = annotation.Name }, token: cancellationToken));
+    }
+}