using System.Collections.Generic;
using Microsoft.Extensions.Logging;
namespace CellularManagement.WebSocket.Pipeline;
///
/// 管道构建器
/// 用于构建消息处理管道
///
/// 输入消息类型
/// 输出消息类型
public class PipelineBuilder
{
private readonly ILoggerFactory _loggerFactory;
private readonly ILogger> _logger;
private readonly List> _steps = new();
public PipelineBuilder(ILoggerFactory loggerFactory)
{
_loggerFactory = loggerFactory;
_logger = loggerFactory.CreateLogger>();
_logger.LogInformation("初始化管道构建器,输入类型:{InputType},输出类型:{OutputType}",
typeof(TInput).Name, typeof(TOutput).Name);
}
///
/// 添加处理步骤
///
/// 处理步骤
/// 管道构建器
public PipelineBuilder AddStep(IPipelineStep step)
{
_logger.LogInformation("添加处理步骤,步骤类型:{StepType}", step.GetType().Name);
_steps.Add(step);
return this;
}
///
/// 构建处理管道
///
/// 处理管道
public IPipelineStep Build()
{
_logger.LogInformation("开始构建处理管道,步骤数量:{StepCount}", _steps.Count);
if (_steps.Count == 0)
{
_logger.LogWarning("处理管道没有添加任何步骤");
throw new InvalidOperationException("处理管道必须至少包含一个步骤");
}
// 构建处理链
var pipeline = _steps[0];
for (int i = 1; i < _steps.Count; i++)
{
var currentStep = _steps[i];
_logger.LogDebug("连接处理步骤:{CurrentStep} -> {NextStep}",
pipeline.GetType().Name, currentStep.GetType().Name);
pipeline = new ChainedPipelineStep(pipeline, currentStep, _loggerFactory);
}
_logger.LogInformation("处理管道构建完成,总步骤数:{StepCount}", _steps.Count);
return pipeline;
}
///
/// 链式处理步骤
/// 用于连接多个处理步骤
///
private class ChainedPipelineStep : IPipelineStep
{
private readonly IPipelineStep _first;
private readonly IPipelineStep _second;
private readonly ILogger _logger;
public ChainedPipelineStep(
IPipelineStep first,
IPipelineStep second,
ILoggerFactory loggerFactory)
{
_first = first;
_second = second;
_logger = loggerFactory.CreateLogger();
_logger.LogDebug("创建链式处理步骤,第一步:{FirstStep},第二步:{SecondStep}",
first.GetType().Name, second.GetType().Name);
}
public async Task ProcessAsync(TInput input, CancellationToken cancellationToken)
{
_logger.LogDebug("开始链式处理消息");
var intermediate = await _first.ProcessAsync(input, cancellationToken);
if (intermediate is TInput inputValue)
{
return await _second.ProcessAsync(inputValue, cancellationToken);
}
throw new InvalidOperationException($"无法将类型 {intermediate?.GetType().Name ?? "null"} 转换为 {typeof(TInput).Name}");
}
public async Task ProcessWithErrorHandlingAsync(TInput input, CancellationToken cancellationToken)
{
_logger.LogDebug("开始链式处理消息(带错误处理)");
try
{
var intermediate = await _first.ProcessWithErrorHandlingAsync(input, cancellationToken);
if (intermediate is TInput inputValue)
{
return await _second.ProcessWithErrorHandlingAsync(inputValue, cancellationToken);
}
throw new InvalidOperationException($"无法将类型 {intermediate?.GetType().Name ?? "null"} 转换为 {typeof(TInput).Name}");
}
catch (Exception ex)
{
_logger.LogError(ex, "链式处理发生错误");
throw;
}
}
}
}
public class Pipeline : IPipelineStep
{
private readonly IReadOnlyList> _steps;
private readonly ILogger> _logger;
public Pipeline(IReadOnlyList> steps, ILogger> logger)
{
_steps = steps ?? throw new ArgumentNullException(nameof(steps));
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
}
public async Task ProcessAsync(TInput input, CancellationToken cancellationToken)
{
TOutput? result = default;
var current = input;
foreach (var step in _steps)
{
result = await step.ProcessAsync(current, cancellationToken);
if (result is TInput nextInput)
{
current = nextInput;
}
else
{
break;
}
}
return result ?? throw new InvalidOperationException("管道处理结果不能为 null");
}
public async Task ProcessWithErrorHandlingAsync(TInput input, CancellationToken cancellationToken)
{
try
{
return await ProcessAsync(input, cancellationToken);
}
catch (Exception ex)
{
_logger.LogError(ex, "管道处理发生错误");
throw new PipelineException("管道处理失败", ex);
}
}
}