MES-ETL/ConsoleApp2/Services/MySqlDestination.cs

using System.Reflection.Metadata;
using System.Text;
using ConsoleApp2.Helpers;
using ConsoleApp2.Options;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using MySqlConnector;
using ServiceStack;

namespace ConsoleApp2.Services;

/// <summary>
/// Mysql导出
/// </summary>
public class MySqlDestination : IDisposable, IAsyncDisposable
{
    private readonly Dictionary<string, IList<DataRecord>> _recordCache;
    private readonly MySqlConnection _conn;
    private readonly ILogger _logger;
    private readonly bool _prettyOutput;
    private readonly int _maxAllowPacket;
    private readonly ProcessContext _context;

    public MySqlDestination(string connStr, ILogger logger, ProcessContext context,bool prettyOutput = false)
    {
        _conn = new MySqlConnection(connStr);
        _conn.Open();
        _recordCache = new Dictionary<string, IList<DataRecord>>();
        _logger = logger;
        _context = context;
        _prettyOutput = prettyOutput;



    }
    public Task WriteRecordAsync(DataRecord record)
    {
        _recordCache.AddOrUpdate(record.TableName, [record], (key, value) =>
        {
            value.Add(record);
            return value;
        });
        return Task.CompletedTask;
    }

    public async Task WriteRecordsAsync(IEnumerable<DataRecord> records)
    {
        foreach (var record in records)
        {
            await WriteRecordAsync(record);
        }
    }

    public async Task FlushAsync(int maxAllowPacket, IOptions<DataTransformOptions> transOptions)
    {
        if (_recordCache.Count == 0)
            return;

        var cmd = _conn.CreateCommand();
        cmd.CommandTimeout = 3 * 60;
        var excuseList = GetExcuseList(_recordCache, maxAllowPacket, transOptions, _prettyOutput);
        try
        {
            foreach (var insertSql in excuseList)
            {
                cmd.CommandText = insertSql;
                await cmd.ExecuteNonQueryAsync();
            }
            _recordCache.Clear();
        }
        catch (Exception e)
        {
            _logger.LogCritical(e, "Error when flushing records, sql: {Sql}", cmd.CommandText.Omit(1000));
            _context.AddException(e);
            throw;
        }
        finally
        {
            await cmd.DisposeAsync();
        }
    }

    public static IList<string> GetExcuseList(IDictionary<string, IList<DataRecord>> tableRecords,int maxAllowPacket, IOptions<DataTransformOptions> transOptions,
           bool prettyOutput = false)
    {
        var resultList = new List<string>();
        var headerSb = new StringBuilder();
        var recordSb = new StringBuilder();
        foreach (var (tableName, records) in tableRecords)
        {
            if (records.Count == 0)
                continue;
            headerSb.Append($"INSERT INTO `{tableName}`(");
            for (var i = 0; i < records[0].Headers.Length; i++)
            {
                var header = records[0].Headers[i];
                headerSb.Append($"`{header}`");
                if (i != records[0].Headers.Length - 1)
                    headerSb.Append(',');
            }

            headerSb.Append(") VALUES ");
            if (prettyOutput)
                headerSb.AppendLine();

            var sbList = new List<string>();
            var currentLength = headerSb.Length;
            for (var i = 0; i < records.Count; i++)
            {
                var record = records[i];
                recordSb.Append('(');
                for (var j = 0; j < record.Fields.Length; j++)
                {
                    var field = record.Fields[j];
                    var header = record.Headers[j];
                    if (transOptions.Value.GetColumnType(record.TableName, header) ==ColumnType.Blob)
                    {
                        if (string.IsNullOrEmpty(field))
                        {
                            recordSb.Append("NULL");
                        }
                        else
                        recordSb.Append("0x"+field);
                    }
                    else
                    recordSb.Append(field);
                    if (j != record.Fields.Length - 1)
                        recordSb.Append(',');
                }

                recordSb.Append(')');

                //if (i != records.Count - 1) // not last field
                //    recordSb.Append(',');
                if (prettyOutput) recordSb.AppendLine();

                if (currentLength + recordSb.Length >= maxAllowPacket)
                {

                    var insertSb = new StringBuilder(headerSb.ToString());

                    insertSb.Append(string.Join(",", sbList));
                    insertSb.Append(";");
                    resultList.Add(insertSb.ToString());
                    insertSb.Clear();
                    sbList.Clear();
                    currentLength = headerSb.Length;
                    sbList.Add(recordSb.ToString());
                }
                else
                {
                    sbList.Add(recordSb.ToString());
                }
                currentLength += recordSb.Length;
                recordSb.Clear();
            }
            if (sbList.Count > 0)
            {
                var insertSb = new StringBuilder(headerSb.ToString());
                insertSb.Append(string.Join(",", sbList));
                insertSb.Append(";");
                resultList.Add(insertSb.ToString());
                insertSb.Clear();
            }
            headerSb.Clear();
        }
        if (resultList.Count == 2)
        {
            var a = 1;
        }
        return resultList;
    }


    public void Dispose()
    {
        _conn.Close();
        _conn.Dispose();
    }

    public async ValueTask DisposeAsync()
    {
        await _conn.CloseAsync();
        await _conn.DisposeAsync();
    }
}