klhztrader/KLHZ.Trader.Core/Declisions/Services/ProcessedPricesLogger.cs

using KLHZ.Trader.Core.Contracts.Messaging.Dtos.Intarfaces;
using KLHZ.Trader.Core.Contracts.Messaging.Interfaces;
using KLHZ.Trader.Core.DataLayer;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using System.Threading.Channels;

namespace KLHZ.Trader.Core.Declisions.Services
{
    public class ProcessedPricesLogger : IHostedService
    {
        private readonly ILogger<ProcessedPricesLogger> _logger;
        private readonly IDataBus _dataBus;
        private readonly IDbContextFactory<TraderDbContext> _dbContextFactory;

        private readonly Channel<IProcessedPrice> _channel = Channel.CreateUnbounded<IProcessedPrice>();
        public ProcessedPricesLogger(IDataBus dataBus, IDbContextFactory<TraderDbContext> dbContextFactory, ILogger<ProcessedPricesLogger> logger)
        {
            _dataBus = dataBus;
            _dbContextFactory = dbContextFactory;
            _logger = logger;
            _dataBus.AddChannel(nameof(ProcessedPricesLogger), _channel);
            _ = ProcessMessages();
        }

        private async Task ProcessMessages()
        {
            using var context = await _dbContextFactory.CreateDbContextAsync();
            context.ChangeTracker.QueryTrackingBehavior = QueryTrackingBehavior.NoTracking;
            var buffer = new List<DataLayer.Entities.Prices.ProcessedPrice>();
            var lastWrite = DateTime.UtcNow;
            while (await _channel.Reader.WaitToReadAsync())
            {
                try
                {
                    var message = await _channel.Reader.ReadAsync();

                    buffer.Add(new DataLayer.Entities.Prices.ProcessedPrice()
                    {
                        Figi = message.Figi,
                        Processor = message.Processor,
                        Ticker = message.Ticker,
                        IsHistoricalData = message.IsHistoricalData,
                        Time = message.Time,
                        Value = message.Value,
                    });

                    if (buffer.Count > 10000 || (DateTime.UtcNow - lastWrite) > TimeSpan.FromSeconds(5) || _channel.Reader.Count == 0)
                    {
                        await context.AddRangeAsync(buffer);
                        await context.SaveChangesAsync();
                        buffer.Clear();
                        lastWrite = DateTime.UtcNow;
                    }
                }
                catch (Exception ex)
                {

                }
            }
        }

        public Task StartAsync(CancellationToken cancellationToken)
        {
            return Task.CompletedTask;
        }

        public Task StopAsync(CancellationToken cancellationToken)
        {
            return Task.CompletedTask;
        }
    }
}