Ble.Framework.Driver.Core 0.0.0.3

Ble.Framework.Driver — 工业级多协议驱动框架

一份纯机制层 + 协议层分离的工业驱动框架。Core 负责"连接生命周期 / 故障自愈 / 编排 / 集群 / 可观测";各协议驱动只负责"协议数据单元(PDU)编解码 + 类型化方法"。

业务侧唯一入口是 IBleSessionManager —— 写入 / 删除 / 替换 / 探测连接 / 等待 / 流式枚举 / 标签增删 / 生命周期事件。

凭据走 BleSecret 抽象(引用 / 内联);描述符写入前 fast-fail 校验器链;BleSessionsHealthCheck 暴露 Kubernetes 存活探针;OpenTelemetry 全埋点;优雅停止;断路器;金丝雀探针;反压自动反映到连接池。

架构红线 11 条:见 CONTRIBUTING.md,任一拉取请求违反即拒绝合并。

1. 目录

1. 适用场景
2. 架构总览
3. 11 不变量
4. 7 个协议驱动
5. 业务前门:IBleSessionManager
6. 配置热更 / 端点增删自动重启
7. 凭据隔离 BleSecret
8. 描述符校验链 IBleDescriptorValidator
9. 传输安全 / 端点级传输配置
10. 断路器 / 优雅排空 / 探针 / 反压
11. 可观测性:OpenTelemetry + 健康检查
12. 集群部署
13. 测试基线
14. 添加新驱动:5 文件夹模板
15. 现场坑 → 不变量 反查表

1. 适用场景

2. 架构总览

7 层,逐层只依赖下层:

┌─────────────────────────────────────────────────────────────────┐
│ L7  应用层门面                                                    │
│   IBleSessionManager (业务唯一前门)                              │
│     · UpsertSessionAsync / RemoveSessionAsync(写入 / 删除)       │
│     · TestConnectionAsync(写入前探测一下)                       │
│     · ReplaceSessionAsync(强制优雅替换)                          │
│     · GetSession / GetSessionStatus / Count                      │
│     · EnumerateAsync(filter) — 异步流式枚举                      │
│     · WaitForSessionAsync(等连上再用)                            │
│     · UpdateTagsAsync / ApplyTagDeltaAsync(标签全量/增量)         │
│     · SessionLifecycleChanged 事件总线                           │
├─────────────────────────────────────────────────────────────────┤
│ L6  生命周期 / 集群                                              │
│   BleSessionValidator + IBleDescriptorValidator 链(快速失败)     │
│   BleSessionLifecycleBus(跨组件事件)                            │
│   BleClusterSessionManager(分片调度 + 优雅替换)                  │
│   IBleSessionRegistry(进程级注册表)                             │
│   IBleSessionStore(权威源:内存 / Redis)                        │
├─────────────────────────────────────────────────────────────────┤
│ L5  会话                                                         │
│   BleDriverSessionBase<TTag, TFrame>                             │
│     · StartAsync / StopAsync / DisposeAsync                      │
│     · ExecuteFrameAsync(执行单帧)/ WriteBatchAsync(批量写)      │
│     · 能力接口:控制 / 文件传输 / 浏览 / 订阅                     │
├─────────────────────────────────────────────────────────────────┤
│ L4  连接池 / 路由                                                │
│   BlePoolHost<TConn>(各驱动模板基类,30 行)                       │
│   BleConnectionPool<TConn>                                       │
│     · 每端点监督器(退避 + 抖动 + 探针)                          │
│     · 每槽位互斥                                                  │
│     · 每端点断路器(可选)                                        │
│     · 优雅停止(等待在飞请求排空)                                │
│     · 反压自动检查(饱和时快速拒绝)                              │
│     · 唯一业务入口 ExecuteAsync(lambda 风格)                     │
├─────────────────────────────────────────────────────────────────┤
│ L3  连接契约                                                     │
│   IBleXxxConnection(每协议 1 份)                                │
│     · 类型化方法(读 / 写 / 发布)                                │
│     · IsHealthy + Faulted 事件(一次性锁存)                      │
│     · 业务异常 → IBleNonFatalProtocolError                       │
│     · 传输 / 帧错误 → BleTransportException...                   │
├─────────────────────────────────────────────────────────────────┤
│ L2  成帧 / 编解码                                                │
│   IBleXxxFramingCodec(自实现协议用,如 Modbus MBAP / RTU)        │
├─────────────────────────────────────────────────────────────────┤
│ L1  传输                                                         │
│   ITransport: BleTcpTransport / BleTlsTcpTransport               │
│     · 端点级配置(同会话内主路 TLS、备路明文皆可)                 │
└─────────────────────────────────────────────────────────────────┘

           横向依赖              IBleSecretProvider
                                 (环境变量 / 配置 / 缓存装饰)

3. 11 不变量

完整版见 CONTRIBUTING.md:

4. 7 个协议驱动

每驱动统一 5 文件夹模板(IEC 61850 因 4 类能力接口例外):

5. 业务前门:IBleSessionManager

public interface IBleSessionManager
{
    // 写入 / 删除 / 替换 / 探测
    Task UpsertSessionAsync(BleSessionDescriptor descriptor, CancellationToken ct = default);
    Task RemoveSessionAsync(string sessionId, CancellationToken ct = default);
    Task ReplaceSessionAsync(string sessionId, CancellationToken ct = default);
    Task<BleConnectionTestResult> TestConnectionAsync(BleSessionDescriptor descriptor, CancellationToken ct = default);

    // 查询
    IBleDriverSession? GetSession(string sessionId);
    BleSessionStatus?  GetSessionStatus(string sessionId);
    int Count { get; }
    IAsyncEnumerable<BleSessionStatus> EnumerateAsync(BleSessionFilter? filter = null, CancellationToken ct = default);

    // 等待
    Task<IBleDriverSession> WaitForSessionAsync(string sessionId, TimeSpan timeout, CancellationToken ct = default);

    // 标签原地增删
    Task UpdateTagsAsync(string sessionId, IEnumerable<IBleTagDefinition> tags, CancellationToken ct = default);
    Task ApplyTagDeltaAsync(string sessionId, BleTagDelta delta, CancellationToken ct = default);

    // 事件
    event EventHandler<BleSessionLifecycleEvent>? SessionLifecycleChanged;
}

典型业务流程:

// 依赖注入
services.AddBleDriverCore();
services.AddBleStandalone();         // 集群版换 AddBleClusterRedis(...)
services.AddBleModbusDriver();       // 协议按需添加,N 个并存
services.AddBleOpcUaDriver();
services.AddBleMqttDriver();

// 业务
var manager = sp.GetRequiredService<IBleSessionManager>();

var descriptor = new BleSessionDescriptor
{
    SessionId = "PLC-A1",
    Driver = BleDriverKind.ModbusTcp,
    Endpoints = [
        new() { Host = "192.168.1.10", Port = 502, ConnectionCount = 2 },
        new() { Host = "192.168.1.11", Port = 502 },
    ],
    DefaultFrequencyMs = 500,
    Properties = new Dictionary<string, object?>
    {
        [BleModbusSessionDescriptor.SlaveAddressKey] = (byte)1,
        [BleModbusSessionDescriptor.FramingKey] = BleModbusFraming.Tcp,
    },
    Tags = [ /* 点位定义 */ ],
};

// 1) 保存前探测一下
var probe = await manager.TestConnectionAsync(descriptor);
if (!probe.Ok) { throw new InvalidOperationException(probe.Reason); }

// 2) 写入存储 + 自动调度
await manager.UpsertSessionAsync(descriptor);

// 3) 等连上再用
var session = await manager.WaitForSessionAsync(descriptor.SessionId, TimeSpan.FromSeconds(5));

// 4) 读 / 写
var values = await session.ExecuteFrameAsync("frame-key");
var result = await session.WriteBatchAsync(writes);

// 5) 标签增量(不重启会话)
await manager.ApplyTagDeltaAsync(descriptor.SessionId, delta);

// 6) 强制刷一次(PLC 物理 reset 后)
await manager.ReplaceSessionAsync(descriptor.SessionId);

// 7) 移除
await manager.RemoveSessionAsync(descriptor.SessionId);

// 8) 仪表盘流式
await foreach (var status in manager.EnumerateAsync(new BleSessionFilter
{
    Drivers = new HashSet<BleDriverKind> { BleDriverKind.ModbusTcp },
    States = new HashSet<BleSessionState> { BleSessionState.Connected },
}))
{
    Console.WriteLine($"{status.SessionId}: {status.State}");
}

6. 配置热更 / 端点增删自动重启

// 改端口 / 加备路 / 调超时 → UpsertAsync 即可
var newDescriptor = oldDescriptor with
{
    Endpoints = [.. oldDescriptor.Endpoints, new() { Host = "192.168.1.12", Port = 502 }],
};
await manager.UpsertSessionAsync(newDescriptor);

自动机制:

1. BleSessionValidator + IBleDescriptorValidator 链快速失败(从站地址范围 / 凭据类型 / 端口范围 / 安全模式等)
2. IBleSessionStore.SaveAsync 持久化为权威源
3. BleClusterSessionManager 监听到 Upserted 事件
4. BleSessionDescriptorEquality.SemanticEquals 判定是否变化
5. 变化 → 优雅替换:起新会话 → WaitConnectedAsync 等连上 → 替换注册表引用 → 优雅停旧
6. SessionLifecycleChanged 触发 Replaced 事件

业务侧在飞的 ExecuteFrameAsync 不会被打断(连接池槽位锁 + 排空保证)。

标签增删走 manager.ApplyTagDeltaAsync 的原地无锁路径(64 分片 CAS-spin),<u>不</u>触发会话重启。

7. 凭据隔离 BleSecret

public abstract record BleSecret;
public sealed record BleSecretRef(string Id) : BleSecret;       // 生产推荐:密钥管理服务 / 环境变量 / Vault 引用
public sealed record BleInlineSecret(string Value) : BleSecret; // 仅测试用

ToString 永远输出引用名或 ***** —— 真值不进任何持久化层。

public interface IBleSecretProvider
{
    ValueTask<string?> ResolveAsync(BleSecret? secret, CancellationToken ct);
}

实现:

• BleEnvSecretProvider(默认):引用 mqtt-prod-pwd → 环境变量 BLE_SECRET_MQTT_PROD_PWD
• BleConfigurationSecretProvider:键 Ble:Secrets:{id} 走 IConfiguration 链(可接 Azure App Configuration / Vault)
• BleCachingSecretProvider 装饰器:按 id 设置 TTL(默认 5 分钟)缓存,密钥服务调用从每秒 N 次降到 5 分钟 N 次
• BleNoopSecretProvider:依赖注入默认兜底

用法(MQTT 凭据示例):

Properties = new Dictionary<string, object?>
{
    [BleMqttSessionDescriptor.UserNameKey] = "factory",
    [BleMqttSessionDescriptor.PasswordKey] = new BleSecretRef("mqtt-prod-pwd"),
}

描述符解析器见到非 BleSecret 类型直接抛 InvalidOperationException —— 杜绝明文字符串误用。

OPC UA / SNMP V3 同款接口。

8. 描述符校验链 IBleDescriptorValidator

每个驱动注册一个,BleSessionValidator 在 UpsertSessionAsync 写入存储前依次跑:

public sealed class BleModbusDescriptorValidator : IBleDescriptorValidator
{
    public void Validate(BleSessionDescriptor d, List<string> errors)
    {
        if (d.Driver != BleDriverKind.ModbusTcp) { return; }
        // 校验从站地址范围 / 帧格式 / 字节序枚举
    }
}

services.AddBleDescriptorValidator<BleModbusDescriptorValidator>();   // 自动跑在 6 个驱动里

错误聚合:BleDescriptorValidationException.Errors 一次性返回所有问题,不挤牙膏。

内置校验项:

• Modbus:从站地址 ∈ [0, 247],帧格式 / 字节序是合法枚举
• S7:CPU 型号 / 机架号 / 槽号范围,端口 ≤ Int16.MaxValue
• OPC UA:口令是 BleSecret,安全模式合法,非 opc.tcp 主机时端口 > 0
• SNMP:版本 / 安全级别合法,认证口令 / 加密口令是 BleSecret,Trap 监听端口范围
• MQTT:口令是 BleSecret,遗嘱 QoS 合法,心跳间隔 ∈ [1, 65535]
• IEC 61850:端口范围

9. 传输安全 / 端点级传输配置

端点级配置,同一会话主路 TLS、备路明文完全合法:

new BleEndpointDescriptor
{
    Host = "plc.example.com",
    Port = 8502,
    Properties = new Dictionary<string, string?>
    {
        [BleEndpointTransport.KeyTransport] = "tls",
        [BleEndpointTransport.KeyTlsTargetHost] = "plc.example.com",
        [BleEndpointTransport.KeyTlsProtocols] = "Tls12,Tls13",
        [BleEndpointTransport.KeyTlsPinnedSpkiSha256] = "abc...,def...",  // 公钥指纹白名单
    },
}

BleEndpointTransport.Build(endpoint, ipEndPoint) 在所有自实现协议(Modbus)的连接器内统一调用。

10. 断路器 / 优雅排空 / 探针 / 反压

断路器(每端点独立,默认禁用)

new BleConnectionPoolOptions<...>
{
    BreakerFailureThreshold = 5,                          // 连续 5 次致命 → 打开
    BreakerBreakDuration    = TimeSpan.FromSeconds(30),
}

断路器打开期请求直接抛 BleCircuitOpenException(或路由到下一端点);到期半开放一条试探。业务异常不计入。

优雅停止(默认 5 秒)

new BleConnectionPoolOptions<...> { StopDrainTimeout = TimeSpan.FromSeconds(10) }

StopAsync 先标记拒绝新请求,等在飞请求归零后才取消。

探针(每协议必须)

每驱动实现 ICanary<IBleXxxConnection>,周期发起协议层无副作用探测(Modbus FC03 读 1 寄存器 / S7 读 MB0 / OPC UA 读 Server.ServerStatus.State / SNMP 读 sysUpTime / IEC 61850 GetState / MQTT 检查 IsConnected)。连续 3 次失败 → 槽位致命 → 监督器重连。

反压

new BleConnectionPoolOptions<...> { BackpressureSignal = sp.GetRequiredService<IBleBackpressureSignal>() }

下游接收端上报 BlePressureLevel.Saturated(饱和)→ 连接池后续 ExecuteAsync 直接抛 BleBackpressureException 快速失败,不堆积在飞请求。

11. 可观测性:OpenTelemetry + 健康检查

OpenTelemetry

services.AddOpenTelemetry()
    .WithTracing(b => b.AddSource(BleDriverTelemetry.ActivitySourceName))
    .WithMetrics(b => b.AddMeter(BleDriverTelemetry.MeterName));

健康检查

services.AddBleSessionsHealthCheck();   // 全局,聚合所有会话
// 或按驱动过滤
services.AddBleSessionsHealthCheck(new BleSessionFilter
{
    Drivers = new HashSet<BleDriverKind> { BleDriverKind.ModbusTcp }
});

services.AddHealthChecks()
    .AddCheck<BleSessionsHealthCheck>("ble-driver-sessions");

返回 Healthy(全部已连)/ Degraded(部分在线)/ Unhealthy(任一已断 / 已故障),数据字段含已连 / 部分连 / 已断三分类计数 + 失败会话标识示例。

12. 集群部署

// 单机
services.AddBleStandalone();

// Redis 集群
services.AddBleClusterRedis(opts =>
{
    opts.ConnectionString = "redis-master:6379";
    opts.ShardCount = 64;
    opts.GracefulShutdownMs = 30_000;
});

IBleSessionManager 接口完全不变 —— UpsertSessionAsync 写到 Redis,各节点的 BleClusterSessionManager 通过监听接管自己负责的分片。节点崩 30 秒内,该节点持有的会话自动迁到其他节点。

13. 测试基线

cd components/Driver
dotnet test    # 当前 445/445 全绿

IEC 61850 无单元测试(libiec61850 跨进程互调用在 CI 跑不动);改动该协议必须本地实跑。

14. 添加新驱动:5 文件夹模板

Ble.Framework.Driver.Foo/
├── Connection/
│   ├── IBleFooConnection.cs                    ← 类型化方法,继承 IBleConnection(对应不变量 #10)
│   ├── BleFooProtocolException.cs              ← 非致命业务异常标记(对应不变量 #4)
│   ├── BleFooCanary.cs                         ← 协议层无副作用探针(对应不变量 #11)
│   └── Impl/
│       ├── BleFooConnection.cs                 ← SDK 包装,致命 / 非致命分流
│       └── BleFooConnectionPool.cs             ← 继承 BlePoolHost<IBleFooConnection>, 30 行
├── Sessions/
│   ├── BleFooSessionDescriptor.cs              ← Properties 键常量 + 强类型 Get*()
│   ├── BleFooConnectionOptions.cs              ← 强类型记录,口令用 BleSecret
│   ├── BleFooDescriptorValidator.cs            ← 实现 IBleDescriptorValidator
│   └── Impl/
│       ├── BleFooSession.cs                    ← 继承 BleDriverSessionBase
│       └── BleFooSessionFactory.cs             ← TestConnectionAsync 用 BleConnectionTestHelpers
├── Diagnostics/BleFooQualityMapper.cs          ← 异常 → 结构化诊断
├── Descriptors/Impl/BleFooDriverDescriptor.cs  ← IBleDriverDescriptor 元数据
├── Extensions/BleFooServiceCollectionExtensions.cs
│       AddBleFooDriver()
│         · AddBleDriverSessionFactory<BleFooSessionFactory>
│         · AddBleDriverDescriptor(new BleFooDriverDescriptor())
│         · AddBleDescriptorValidator<BleFooDescriptorValidator>
├── Module/BleFooModule.cs                      ← Furion 自动加载入口
└── _Imports.cs

骨架 ~600 行(含测试)。不变量都在 CONTRIBUTING.md 写死。

15. 现场坑 → 不变量 反查表

许可

© Ble Corporation. All rights reserved.