DMS/软件设计文档/05-数据处理链设计.md

# 软件开发文档 - 数据处理链设计

本文档专门阐述当系统从S7 PLC等设备读取到变量值后，如何通过一个可扩展的、链式的处理流程，对数据进行加工、存储和转发。

## 1. 设计目标

*   **解耦**：每个处理步骤应是独立的、可复用的。
*   **可扩展**：可以轻松地添加新的处理环节（例如：数据校验、报警判断、线性转换等）。
*   **高性能**：处理流程应高效，避免阻塞数据采集线程。
*   **可配置**：能够根据不同变量的需求，动态构建处理链。

## 2. 核心概念：`VariableContext`

为了在处理链中传递数据，我们定义一个上下文对象 `VariableContext`。它将携带变量的原始值、当前值以及处理状态，在整个处理流程中流动。

```csharp
// 文件: DMS.Core/Models/VariableContext.cs
using System;

namespace DMS.Core.Models;

public class VariableContext
{
    public Variable Variable { get; init; }
    public object RawValue { get; init; } // 从设备读取的原始值
    public object CurrentValue { get; set; } // 当前处理后的值
    public DateTime Timestamp { get; init; }
    public bool IsProcessed { get; set; } // 标记是否已处理（短路机制）
    public bool IsSuccess { get; set; } // 标记处理是否成功
    public bool IsError { get; set; } // 标记处理是否发生错误

    public VariableContext(Variable variable, object rawValue)
    {
        Variable = variable;
        RawValue = rawValue;
        CurrentValue = rawValue;
        Timestamp = DateTime.UtcNow;
        IsProcessed = false;
        IsSuccess = false;
        IsError = false;
    }
}
```

## 3. 处理器接口

我们定义一个统一的处理器接口 `IVariableProcessor`。

### 3.1. `IVariableProcessor` 接口

```csharp
// 文件: DMS.Application/Interfaces/IVariableProcessor.cs
using System.Threading.Tasks;

namespace DMS.Application.Interfaces;

public interface IVariableProcessor
{
    Task ProcessAsync(VariableContext context);
}
```

## 4. 具体处理器实现

以下是处理链中几个核心处理器的设计。

### 4.1. `ChangeDetectionProcessor` - 变化检测处理器

**职责**：检测本次读取的值与上一次的值是否相同。如果相同，则标记 `IsProcessed` 为 `true`，以短路后续处理。

```csharp
// 文件: DMS.Infrastructure/Services/Processors/ChangeDetectionProcessor.cs
using DMS.Application.Interfaces;
using DMS.Core.Models;
using Microsoft.Extensions.Caching.Memory;
using System;
using System.Threading.Tasks;

namespace DMS.Infrastructure.Services.Processing;

public class ChangeDetectionProcessor : IVariableProcessor
{
    private readonly IMemoryCache _cache; // 使用内存缓存来存储上一次的值

    public ChangeDetectionProcessor(IMemoryCache cache)
    {
        _cache = cache;
    }

    public Task ProcessAsync(VariableContext context)
    {
        if (context.IsProcessed || context.IsError) return Task.CompletedTask; // 短路

        try
        {
            var lastValue = _cache.Get(context.Variable.Id);
            if (lastValue != null && lastValue.Equals(context.RawValue))
            {
                context.IsProcessed = true; // 值未变化，标记为已处理，短路后续操作
            }
            else
            {
                _cache.Set(context.Variable.Id, context.RawValue, TimeSpan.FromDays(1));
            }
        }
        catch (Exception ex)
        {
            context.IsError = true; // 标记错误
            // 记录日志
        }
        return Task.CompletedTask;
    }
}
```

### 4.2. `HistoryStorageProcessor` - 历史存储处理器

**职责**：将变化后的值存入数据库历史记录表。

```csharp
// 文件: DMS.Infrastructure/Services/Processors/HistoryStorageProcessor.cs
using DMS.Application.Interfaces;
using DMS.Core.Interfaces;
using DMS.Core.Models;
using System;
using System.Threading.Tasks;

namespace DMS.Infrastructure.Services.Processing;

public class HistoryStorageProcessor : IVariableProcessor
{
    private readonly IVariableHistoryRepository _historyRepository;

    public HistoryStorageProcessor(IVariableHistoryRepository historyRepository)
    {
        _historyRepository = historyRepository;
    }

    public async Task ProcessAsync(VariableContext context)
    {
        if (context.IsProcessed || context.IsError) return; // 短路

        try
        {
            // 只有当值发生变化时才存储历史，这里需要根据ChangeDetectionProcessor的逻辑来判断
            // 由于ChangeDetectionProcessor现在只标记IsProcessed，我们需要重新判断是否需要存储历史
            // 简单起见，这里假设如果未被短路，且RawValue与CurrentValue不同，则认为值已变化
            // 更严谨的做法是ChangeDetectionProcessor设置一个IsValueChanged标志
            // 这里我们直接使用RawValue和CurrentValue的比较
            if (!context.RawValue.Equals(context.CurrentValue))
            {
                var history = new VariableHistory
                {
                    VariableId = context.Variable.Id,
                    Value = context.CurrentValue.ToString(),
                    Timestamp = context.Timestamp
                };
                await _historyRepository.AddAsync(history);
            }
        }
        catch (Exception ex)
        {
            context.IsError = true; // 标记错误
            // 记录日志
        }
    }
}
```

### 4.3. `MqttPublishProcessor` - MQTT发布处理器

**职责**：如果变量关联了MQTT服务器，则将值发布出去。

```csharp
// 文件: DMS.Infrastructure/Services/Processors/MqttPublishProcessor.cs
using DMS.Application.Interfaces;
using DMS.Core.Interfaces;
using DMS.Core.Models;
using DMS.Infrastructure.Services.Communication;
using System.Linq;
using System.Text.Json;
using System.Threading.Tasks;
using NLog;

namespace DMS.Infrastructure.Services.Processing;

/// <summary>
/// MQTT发布处理器，负责将变量值发布到关联的MQTT服务器，并使用专属别名。
/// </summary>
public class MqttPublishProcessor : IVariableProcessor
{
    private readonly IMqttPublishService _mqttService;
    private readonly IRepositoryManager _repoManager; // 使用 RepositoryManager 来获取仓储
    private static readonly ILogger _logger = LogManager.GetCurrentClassLogger();

    /// <summary>
    /// 构造函数。
    /// </summary>
    public MqttPublishProcessor(IMqttPublishService mqttService, IRepositoryManager repoManager)
    {
        _mqttService = mqttService;
        _repoManager = repoManager;
    }

    public async Task ProcessAsync(VariableContext context)
    {
        if (context.IsProcessed || context.IsError) return; // 短路

        try
        {
            // 1. 从仓储获取变量及其完整的别名关联列表
            var variableWithAliases = await _repoManager.Variables.GetVariableWithMqttAliasesAsync(context.Variable.Id);
            
            if (variableWithAliases?.MqttAliases == null || !variableWithAliases.MqttAliases.Any())
            {
                return; // 没有关联的MQTT服务器，无需发布
            }

            foreach (var aliasInfo in variableWithAliases.MqttAliases)
            {
                // 确保 MqttServer 导航属性已加载且激活
                var targetServer = aliasInfo.MqttServer;
                if (targetServer == null || !targetServer.IsActive)
                {
                    _logger.Warn($"MQTT发布失败：变量 {context.Variable.Name} 关联的MQTT服务器 {aliasInfo.MqttServerId} 不存在或未激活。");
                    continue;
                }

                // 使用别名构建Topic
                // 示例Topic格式：DMS/DeviceName/VariableAlias
                var topic = $"DMS/{context.Variable.VariableTable.Device.Name}/{aliasInfo.Alias}";
                var payload = JsonSerializer.Serialize(new { value = context.CurrentValue, timestamp = context.Timestamp });

                await _mqttService.PublishAsync(targetServer, topic, payload);
            }
        }
        catch (Exception ex)
        {
            context.IsError = true; // 标记错误
            _logger.Error(ex, $"MQTT发布失败：变量 {context.Variable.Name}");
        }
    }
}
```

## 5. 变量处理后台服务 (`VariableProcessingService`)

设计一个新的后台服务 `VariableProcessingService`，它将负责管理和执行所有注册的 `IVariableProcessor`。这个服务将从数据采集队列中读取 `VariableContext`，并按顺序调用所有处理器，实现短路机制。

```csharp
// 文件: DMS.Infrastructure/Services/VariableProcessingService.cs
using DMS.Application.Interfaces;
using DMS.Core.Models;
using DMS.WPF.Services;
using CommunityToolkit.Mvvm.Messaging;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using NLog;
using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Channels;
using System.Threading.Tasks;

namespace DMS.Infrastructure.Services;

/// <summary>
/// 变量数据处理后台服务，负责协调所有 IVariableProcessor 的执行。
/// </summary>
public class VariableProcessingService : BackgroundService
{
    private readonly ChannelReader<VariableContext> _queueReader;
    private readonly IServiceProvider _serviceProvider;
    private readonly IMessenger _messenger;
    private static readonly ILogger _logger = LogManager.GetCurrentClassLogger();

    public VariableProcessingService(
        IChannelBus channelBus,
        IServiceProvider serviceProvider,
        IMessenger messenger)
    {
        _queueReader = channelBus.GetReader<VariableContext>("DataProcessingQueue");
        _serviceProvider = serviceProvider;
        _messenger = messenger;
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        await foreach (var context in _queueReader.ReadAllAsync(stoppingToken))
        {
            // 获取所有注册的 IVariableProcessor 实例
            // 注意：这里假设处理器是 Transient 或 Scoped，每次处理都获取新的实例
            var processors = _serviceProvider.GetServices<IVariableProcessor>();

            context.IsSuccess = true; // 默认成功，任何处理器失败则设为false

            foreach (var processor in processors)
            {
                if (context.IsProcessed || context.IsError) // 短路机制
                {
                    context.IsSuccess = false; // 如果短路，则不认为是完全成功
                    break;
                }

                try
                {
                    await processor.ProcessAsync(context);
                }
                catch (Exception ex)
                {
                    context.IsError = true; // 标记错误
                    context.IsSuccess = false; // 标记失败
                    _logger.Error(ex, $"变量处理器 {processor.GetType().Name} 执行错误，变量ID: {context.Variable.Id}");
                    break; // 发生错误，中断后续处理
                }
            }

            // 处理完成后，发送消息通知UI更新变量值
            // 只有当处理未被短路且没有错误时才发送成功消息
            if (!context.IsProcessed && !context.IsError)
            {
                _messenger.Send(new VariableValueUpdatedMessage(context.Variable.Id, context.CurrentValue));
            }
            else if (context.IsError)
            {
                // 可以发送错误消息或进行其他错误处理
                _messenger.Send(new VariableProcessingErrorMessage(context.Variable.Id, context.CurrentValue, "处理失败"));
            }
        }
    }
}
```

## 6. 注册处理器和后台服务

在应用程序的启动配置中（例如 `Startup.cs` 或 `Program.cs`），需要注册所有的 `IVariableProcessor` 实现和 `VariableProcessingService`。

```csharp
// 在 ConfigureServices 方法中
public void ConfigureServices(IServiceCollection services)
{
    // ... 其他服务注册

    // 注册所有 IVariableProcessor 实现
    services.AddTransient<IVariableProcessor, ChangeDetectionProcessor>();
    services.AddTransient<IVariableProcessor, HistoryStorageProcessor>();
    services.AddTransient<IVariableProcessor, MqttPublishProcessor>();
    // ... 注册其他处理器

    // 注册后台服务
    services.AddHostedService<VariableProcessingService>();

    // 注册 IMemoryCache (如果尚未注册)
    services.AddMemoryCache();
}
```

## 7. 数据采集服务中的调用

数据采集服务（例如 `S7BackgroundService`）现在只需将采集到的 `VariableContext` 写入 `DataProcessingQueue` 即可，无需关心处理链的构建和执行。

```csharp
// 在 S7BackgroundService.cs 或类似服务中
public class S7BackgroundService : IHostedService
{
    private readonly ChannelWriter<VariableContext> _processingQueueWriter;
    // ... 其他依赖

    public S7BackgroundService(IRepositoryManager repo, IChannelBus bus, IMessenger msg)
    {
        // ...
        _processingQueueWriter = bus.GetWriter<VariableContext>("DataProcessingQueue");
    }

    // ... StartAsync 方法中
    private async Task PollingLoopAsync(List<Variable> varsToRead, int interval, CancellationToken token)
    {
        while (!token.IsCancellationRequested)
        {
            // ... 读取变量
            foreach (var variable in varsToRead)
            {
                var context = new VariableContext(variable, variable.DataValue);
                await _processingQueueWriter.WriteAsync(context, token);
            }
            // ...
        }
    }
}
```