MES-ETL/ConsoleApp2/HostedServices/OutputService.cs

using ConsoleApp2.Const;
using ConsoleApp2.HostedServices.Abstractions;
using ConsoleApp2.Options;
using ConsoleApp2.Services;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;

namespace ConsoleApp2.HostedServices;

/// <summary>
/// 数据导出服务，将数据导出至MySql服务
/// </summary>
public class OutputService : IOutputService
{
    private readonly ILogger _logger;
    private readonly DataRecordQueue _consumerQueue;
    private readonly IOptions<DatabaseOutputOptions> _outputOptions;
    private readonly IOptions<DataTransformOptions> _transformOptions;
    private readonly ProcessContext _context;
    private readonly TaskManager _taskManager;
    private readonly ErrorRecorder _errorRecorder;

    public OutputService(ILogger<OutputService> logger,
        [FromKeyedServices(ProcessStep.Consumer)] DataRecordQueue consumerQueue,
        IOptions<DatabaseOutputOptions> outputOptions,
        ProcessContext context,
        TaskManager taskManager, 
        IOptions<DataTransformOptions> transformOptions, 
        ErrorRecorder errorRecorder)
    {
        _logger = logger;
        _consumerQueue = consumerQueue;
        _outputOptions = outputOptions;
        _context = context;
        _taskManager = taskManager;
        _transformOptions = transformOptions;
        _errorRecorder = errorRecorder;
    }

    public async Task ExecuteAsync(CancellationToken cancellationToken)
    {
        _logger.LogInformation("***** Mysql output service started *****");
        _taskManager.CreateTasks(async () =>
        {
            var records = new List<DataRecord>();
            while (!_context.IsTransformCompleted || _consumerQueue.Count > 0)
            {
                if (!_consumerQueue.TryDequeue(out var record)) continue;
                records.Add(record);
                //_logger.LogInformation(@"*****OutputCount: {count} *****",count);
                if (records.Count >= _outputOptions.Value.FlushCount)
                {
                    await FlushAsync(records);
                    records.Clear();
                }
                if (_context.GetExceptions().Count>0)
                {
                    _logger.LogInformation("***** Csv output thread is canceled *****");
                    return;
                }
            }
            if (records.Count > 0)
            {
                await FlushAsync(records);
                records.Clear();
                _logger.LogInformation("***** Mysql output thread completed *****");
            }
        }, _outputOptions.Value.TaskCount);

        await _taskManager.WaitAll();
        //_context.CompleteOutput();
        _logger.LogInformation(@"***** Mysql output service completed *****");

    }

    private async Task FlushAsync(IEnumerable<DataRecord> records)
    {
        var count = 0;
        await using var output = new MySqlDestination(
            _outputOptions.Value.ConnectionString ?? throw new InvalidOperationException("Connection string is required"),
            _logger, _context, _transformOptions, _errorRecorder);
        //if (records == null || records.Count() == 0) return;
        //var dbName = $"cferp_test_1";
        //if (records != null && records.Count() > 0)
        //{
        //    dbName = $"cferp_test_{records.FirstOrDefault()?.CompanyID}";
        //}

        //await  using var output = new MySqlDestination(new MySqlConnectionStringBuilder
        //{
        //    Server = "127.0.0.1",
        //    Port = 34309,
        //    Database = dbName,
        //    UserID = "root",
        //    Password = "123456",
        //    MaximumPoolSize = 50,
        //}.ConnectionString, _logger,true);
        foreach (var record in records)
        {
            await output.WriteRecordAsync(record);
            count++;
        }
        await output.FlushAsync(_outputOptions.Value.MaxAllowedPacket);
        _context.AddOutput(count);
    }
}