GFramework/GFramework.Core/logging/appenders/AsyncLogAppender.cs

using System.Threading.Channels;
using GFramework.Core.Abstractions.logging;

namespace GFramework.Core.logging.appenders;

/// <summary>
///     异步日志输出器，使用 Channel 实现非阻塞日志写入
/// </summary>
public sealed class AsyncLogAppender : ILogAppender, IDisposable
{
    private readonly Channel<LogEntry> _channel;
    private readonly CancellationTokenSource _cts;
    private readonly ILogAppender _innerAppender;
    private readonly Task _processingTask;
    private bool _disposed;

    /// <summary>
    ///     创建异步日志输出器
    /// </summary>
    /// <param name="innerAppender">内部日志输出器</param>
    /// <param name="bufferSize">缓冲区大小（默认 10000）</param>
    public AsyncLogAppender(ILogAppender innerAppender, int bufferSize = 10000)
    {
        _innerAppender = innerAppender ?? throw new ArgumentNullException(nameof(innerAppender));

        if (bufferSize <= 0)
            throw new ArgumentException("Buffer size must be greater than 0.", nameof(bufferSize));

        // 创建有界 Channel
        var options = new BoundedChannelOptions(bufferSize)
        {
            FullMode = BoundedChannelFullMode.Wait, // 缓冲区满时等待
            SingleReader = true,
            SingleWriter = false
        };

        _channel = Channel.CreateBounded<LogEntry>(options);
        _cts = new CancellationTokenSource();

        // 启动后台处理任务
        _processingTask = Task.Run(() => ProcessLogsAsync(_cts.Token));
    }

    /// <summary>
    ///     获取当前缓冲区中的日志数量
    /// </summary>
    public int PendingCount => _channel.Reader.Count;

    /// <summary>
    ///     获取是否已完成处理
    /// </summary>
    public bool IsCompleted => _channel.Reader.Completion.IsCompleted;

    /// <summary>
    ///     释放资源
    /// </summary>
    public void Dispose()
    {
        if (_disposed) return;

        // 标记 Channel 为完成状态
        _channel.Writer.Complete();

        // 等待处理任务完成（最多等待 5 秒）
        if (!_processingTask.Wait(TimeSpan.FromSeconds(5)))
        {
            _cts.Cancel();
        }

        // 释放内部 Appender
        if (_innerAppender is IDisposable disposable)
        {
            disposable.Dispose();
        }

        _cts.Dispose();
        _disposed = true;
    }

    /// <summary>
    ///     追加日志条目（非阻塞）
    /// </summary>
    /// <param name="entry">日志条目</param>
    public void Append(LogEntry entry)
    {
        if (_disposed)
            throw new ObjectDisposedException(nameof(AsyncLogAppender));

        // 尝试非阻塞写入，如果失败则丢弃（避免阻塞调用线程）
        _channel.Writer.TryWrite(entry);
    }

    /// <summary>
    ///     刷新缓冲区，等待所有日志写入完成
    /// </summary>
    public void Flush()
    {
        if (_disposed) return;

        // 等待 Channel 中的所有消息被处理
        while (_channel.Reader.Count > 0)
        {
            Thread.Sleep(10);
        }

        _innerAppender.Flush();
    }

    /// <summary>
    ///     后台处理日志的异步方法
    /// </summary>
    private async Task ProcessLogsAsync(CancellationToken cancellationToken)
    {
        try
        {
            await foreach (var entry in _channel.Reader.ReadAllAsync(cancellationToken))
            {
                try
                {
                    _innerAppender.Append(entry);
                }
                catch (Exception ex)
                {
                    // 记录内部错误到控制台（避免递归）
                    await Console.Error.WriteLineAsync($"[AsyncLogAppender] Error processing log entry: {ex.Message}");
                }
            }
        }
        catch (OperationCanceledException)
        {
            // 正常取消，忽略
        }
        catch (Exception ex)
        {
            await Console.Error.WriteLineAsync($"[AsyncLogAppender] Fatal error in processing task: {ex}");
        }
        finally
        {
            // 确保最后刷新
            try
            {
                _innerAppender.Flush();
            }
            catch
            {
                // 忽略刷新错误
            }
        }
    }
}