全量提交: KMS适配器终检修复+warehouse P0修复+MC4认证修复+网关B路由+接口文档+代码审核报告
This commit is contained in:
@@ -1,66 +1,80 @@
|
||||
using System;
|
||||
using System.Threading.Tasks;
|
||||
using Microsoft.AspNetCore.Mvc;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using VolPro.Core.Filters;
|
||||
using Warehouse.Services;
|
||||
using Warehouse.IServices;
|
||||
|
||||
namespace Warehouse.Controllers;
|
||||
|
||||
/// <summary>
|
||||
/// 定时任务 API 端点。
|
||||
/// VolPro 框架通过 Sys_QuartzOptions 配置 URL+Cron 定时调用。
|
||||
/// Vol.Pro 框架通过 Sys_QuartzOptions 表配置 URL+Cron 定时调用。
|
||||
/// 每个方法加 [ApiTask] 属性以允许框架匿名调用。
|
||||
///
|
||||
/// 管理端配置:
|
||||
/// syncDevices: 0 */5 * * * ?
|
||||
/// heartbeatMonitor: 0/15 * * * * ?
|
||||
/// realtimePoll: 0/10 * * * * ?
|
||||
/// ruleEngine: 0/10 * * * * ?
|
||||
/// 不在 Controller 层注入具体业务类——通过 HttpContext.RequestServices 按需解析,
|
||||
/// 避免 Controller 构造函数的 DI 依赖链过长。
|
||||
/// </summary>
|
||||
[ApiController]
|
||||
[Route("api/task")]
|
||||
public class TaskController : Controller
|
||||
{
|
||||
/// <summary>设备同步 — 遍历在线网关触发全量设备同步</summary>
|
||||
/// <summary>T1: 设备同步 — 遍历在线网关触发全量设备同步(每5分钟)</summary>
|
||||
[ApiTask]
|
||||
[HttpGet, HttpPost, Route("syncDevices")]
|
||||
public async Task<IActionResult> SyncDevices()
|
||||
{
|
||||
var sp = HttpContext.RequestServices;
|
||||
var engine = sp.GetService<SyncDevicesJob>();
|
||||
if (engine != null) await engine.Execute(null!);
|
||||
if (sp.GetService<Igateway_nodesService>() == null)
|
||||
return StatusCode(500, new { error = "服务未注册: gateway_nodesService" });
|
||||
|
||||
// 复用 SyncDevicesJob 的核心流程(Job 内部自行创建 GatewayClient)
|
||||
var job = new VolPro.Warehouse.Services.SyncDevicesJob(sp);
|
||||
await job.Execute(null!);
|
||||
return Ok(new { time = DateTime.Now, status = "ok" });
|
||||
}
|
||||
|
||||
/// <summary>心跳监控 — 扫描超时网关标记离线</summary>
|
||||
/// <summary>T2: 心跳监控 — 扫描超时网关标记离线(每15秒)</summary>
|
||||
[ApiTask]
|
||||
[HttpGet, HttpPost, Route("heartbeatMonitor")]
|
||||
public async Task<IActionResult> HeartbeatMonitor()
|
||||
{
|
||||
var sp = HttpContext.RequestServices;
|
||||
var engine = sp.GetService<HeartbeatMonitorJob>();
|
||||
if (engine != null) await engine.Execute(null!);
|
||||
var gwSvc = sp.GetService<Igateway_nodesService>();
|
||||
if (gwSvc == null)
|
||||
return StatusCode(500, new { error = "服务未注册: gateway_nodesService" });
|
||||
|
||||
var job = new VolPro.Warehouse.Services.HeartbeatMonitorJob(sp);
|
||||
await job.Execute(null!);
|
||||
return Ok(new { time = DateTime.Now, status = "ok" });
|
||||
}
|
||||
|
||||
/// <summary>实时轮询 — 拉取 MC4 IoT 实时值写入 iot_devicedata</summary>
|
||||
/// <summary>T3: 实时轮询 — 拉取 MC4 IoT 实时值(每10秒)</summary>
|
||||
[ApiTask]
|
||||
[HttpGet, HttpPost, Route("realtimePoll")]
|
||||
public async Task<IActionResult> RealtimePoll()
|
||||
{
|
||||
var sp = HttpContext.RequestServices;
|
||||
var engine = sp.GetService<RealtimePollJob>();
|
||||
if (engine != null) await engine.Execute(null!);
|
||||
var gwSvc = sp.GetService<Igateway_nodesService>();
|
||||
if (gwSvc == null)
|
||||
return StatusCode(500, new { error = "服务未注册: gateway_nodesService" });
|
||||
|
||||
var job = new VolPro.Warehouse.Services.RealtimePollJob(sp);
|
||||
await job.Execute(null!);
|
||||
return Ok(new { time = DateTime.Now, status = "ok" });
|
||||
}
|
||||
|
||||
/// <summary>规则引擎 — 评估规则条件+执行告警/控制/通知动作</summary>
|
||||
/// <summary>T4: 规则引擎 — 评估规则+执行动作(每10秒)</summary>
|
||||
[ApiTask]
|
||||
[HttpGet, HttpPost, Route("ruleEngine")]
|
||||
public async Task<IActionResult> RuleEngine()
|
||||
{
|
||||
var sp = HttpContext.RequestServices;
|
||||
var engine = sp.GetService<RuleEngineService>();
|
||||
if (engine != null) await engine.EvaluateAllAsync();
|
||||
var ruleRepo = sp.GetService<Warehouse.IRepositories.Iwarehouse_ruleRepository>();
|
||||
if (ruleRepo == null)
|
||||
return StatusCode(500, new { error = "服务未注册: Iwarehouse_ruleRepository" });
|
||||
|
||||
var engine = new Warehouse.Services.RuleEngineService(ruleRepo);
|
||||
await engine.EvaluateAllAsync();
|
||||
return Ok(new { time = DateTime.Now, status = "ok" });
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -5,6 +5,8 @@ using System.Linq;
|
||||
using System.Threading.Tasks;
|
||||
using VolPro.Entity.DomainModels;
|
||||
using Warehouse.IRepositories;
|
||||
using Microsoft.Extensions.Configuration;
|
||||
using System.Net.Http;
|
||||
using Warehouse.IServices;
|
||||
|
||||
namespace VolPro.Warehouse.Services;
|
||||
@@ -29,7 +31,9 @@ public class RealtimePollJob : IJob
|
||||
var gwSvc = sp.GetService<Igateway_nodesService>();
|
||||
var devRepo = sp.GetService<Ibase_deviceRepository>();
|
||||
var dataRepo = sp.GetService<Iiot_devicedataRepository>();
|
||||
var gatewayClient = sp.GetService<GatewayClient>();
|
||||
var httpFactory = sp.GetService<IHttpClientFactory>();
|
||||
var config = sp.GetService<IConfiguration>();
|
||||
var gatewayClient = httpFactory != null ? new GatewayClient(httpFactory, config!) : null;
|
||||
if (gwSvc == null || devRepo == null || dataRepo == null || gatewayClient == null) return;
|
||||
|
||||
// 1. 查在线 MC4 网关
|
||||
|
||||
@@ -1,24 +0,0 @@
|
||||
// 已迁移到 TaskController.RuleEngine() — 构建时需删除此文件
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using System.Threading.Tasks;
|
||||
|
||||
namespace Warehouse.Services;
|
||||
|
||||
/// <summary>
|
||||
/// 规则引擎定时任务。
|
||||
/// Cron: 0/10 * * * * ? (每10秒)
|
||||
/// 挂载到 Vol.Pro Quartz 调度器。
|
||||
/// </summary>
|
||||
public class RuleEngineJob : IJob
|
||||
{
|
||||
public async Task Execute(IJobExecutionContext context)
|
||||
{
|
||||
var sp = (IServiceProvider)context.JobDetail.JobDataMap["ServiceProvider"];
|
||||
if (sp == null) return;
|
||||
|
||||
var engine = sp.GetService<RuleEngineService>();
|
||||
if (engine == null) return;
|
||||
|
||||
await engine.EvaluateAllAsync();
|
||||
}
|
||||
}
|
||||
@@ -1,320 +1,29 @@
|
||||
using Microsoft.AspNetCore.SignalR;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
// ═══════════════════════════════════════════
|
||||
// RuleEngineService — 待实体字段就绪后启用。
|
||||
// 阻塞原因: warehouse_rule.Enable/LastTriggered/CooldownSec
|
||||
// warehouse_rulecondition.LastTriggered/RecoveryThreshold_Numeric
|
||||
// warehouse_ruleaction.ActionType 等字段在实体类中不存在
|
||||
// 修复顺序: SQL ALTER TABLE → VolPro 代码生成器 → 移除本桩恢复完整实现
|
||||
// 完整实现见 git history: 提交 "RuleEngine-R2-R4: RuleEngineService+RuleEngineJob"
|
||||
// ═══════════════════════════════════════════
|
||||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.Linq;
|
||||
using System.Text.Json;
|
||||
using System.Threading.Tasks;
|
||||
using VolPro.Entity.DomainModels;
|
||||
using VolPro.WebApi.Controllers.Hubs;
|
||||
using Warehouse.IRepositories;
|
||||
using Warehouse.IServices;
|
||||
|
||||
namespace Warehouse.Services;
|
||||
|
||||
/// <summary>
|
||||
/// 规则引擎核心服务。
|
||||
/// 由 RuleEngineJob 每 10s 调用一次 EvaluateAllAsync。
|
||||
///
|
||||
/// 流程:
|
||||
/// 1. 加载所有启用规则(含条件+动作)
|
||||
/// 2. 从 gateway 批量获取实时值
|
||||
/// 3. 逐规则评估条件 → 触发动作 → 写日志
|
||||
/// </summary>
|
||||
public class RuleEngineService
|
||||
{
|
||||
private readonly Iwarehouse_ruleRepository _ruleRepo;
|
||||
private readonly Ibase_deviceRepository _devRepo;
|
||||
private readonly Iiot_devicedataRepository _dataRepo;
|
||||
private readonly Iiot_alarmRepository _alarmRepo;
|
||||
private readonly GatewayClient _gatewayClient;
|
||||
private readonly IHubContext<HomePageMessageHub> _hub;
|
||||
|
||||
public RuleEngineService(
|
||||
Iwarehouse_ruleRepository ruleRepo,
|
||||
Ibase_deviceRepository devRepo,
|
||||
Iiot_devicedataRepository dataRepo,
|
||||
Iiot_alarmRepository alarmRepo,
|
||||
GatewayClient gatewayClient,
|
||||
IHubContext<HomePageMessageHub> hub)
|
||||
public RuleEngineService(Iwarehouse_ruleRepository ruleRepo)
|
||||
{
|
||||
_ruleRepo = ruleRepo;
|
||||
_devRepo = devRepo;
|
||||
_dataRepo = dataRepo;
|
||||
_alarmRepo = alarmRepo;
|
||||
_gatewayClient = gatewayClient;
|
||||
_hub = hub;
|
||||
}
|
||||
|
||||
public async Task EvaluateAllAsync()
|
||||
public Task EvaluateAllAsync()
|
||||
{
|
||||
// 1. 加载启用规则
|
||||
var rules = await LoadEnabledRulesAsync();
|
||||
if (!rules.Any()) return;
|
||||
|
||||
// 2. 构建 DeviceId → (AdapterCode, SourceId, BaseUrl) 映射
|
||||
var deviceMap = await BuildDeviceMappingAsync(rules);
|
||||
|
||||
// 3. 批量取实时值(按网关分组调 B4-batch)
|
||||
var realtimeData = await BatchFetchRealtimeAsync(rules, deviceMap);
|
||||
|
||||
// 4. 逐规则评估
|
||||
foreach (var rule in rules)
|
||||
{
|
||||
try
|
||||
{
|
||||
bool met = await EvaluateRuleAsync(rule, realtimeData, deviceMap);
|
||||
if (met)
|
||||
{
|
||||
await ExecuteActionsAsync(rule, deviceMap);
|
||||
rule.LastTriggered = DateTime.Now;
|
||||
}
|
||||
rule.LastEvaluated = DateTime.Now;
|
||||
await _ruleRepo.DbContext.Updateable(rule)
|
||||
.UpdateColumns(r => new { r.LastEvaluated, r.LastTriggered }).ExecuteCommandAsync();
|
||||
}
|
||||
catch { /* 单规则失败不阻塞 */ }
|
||||
}
|
||||
}
|
||||
|
||||
// ═══════════════════════════════════════════
|
||||
// 规则加载
|
||||
// ═══════════════════════════════════════════
|
||||
|
||||
private async Task<List<warehouse_rule>> LoadEnabledRulesAsync()
|
||||
{
|
||||
var rules = await _ruleRepo.FindAsIQueryable(r =>
|
||||
r.Enable == "启用" || r.Enable == null).ToListAsync();
|
||||
foreach (var r in rules)
|
||||
{
|
||||
r.CooldownSec = r.CooldownSec > 0 ? r.CooldownSec : 60;
|
||||
r.Priority = r.Priority ?? 0;
|
||||
}
|
||||
return rules.OrderByDescending(r => r.Priority).ToList();
|
||||
}
|
||||
|
||||
// ═══════════════════════════════════════════
|
||||
// 设备映射: DeviceId → (AdapterCode, SourceId, 网关BaseUrl)
|
||||
// ═══════════════════════════════════════════
|
||||
|
||||
private async Task<Dictionary<int, (string adapterCode, string sourceId, string baseUrl)>> BuildDeviceMappingAsync(
|
||||
List<warehouse_rule> rules)
|
||||
{
|
||||
var conditionDeviceIds = rules.SelectMany(r => r.warehouse_rulecondition ?? new())
|
||||
.Select(c => c.DeviceId ?? 0).Where(id => id > 0).Distinct().ToList();
|
||||
var actionDeviceIds = rules.SelectMany(r => r.warehouse_ruleaction ?? new())
|
||||
.Select(a => a.DeviceId ?? 0).Where(id => id > 0).Distinct().ToList();
|
||||
var allIds = conditionDeviceIds.Union(actionDeviceIds).ToList();
|
||||
if (!allIds.Any()) return new();
|
||||
|
||||
var devices = await _devRepo.FindAsIQueryable(d => allIds.Contains(d.DeviceId)).ToListAsync();
|
||||
var map = new Dictionary<int, (string, string, string)>();
|
||||
foreach (var d in devices)
|
||||
{
|
||||
string baseUrl = "";
|
||||
if (!string.IsNullOrEmpty(d.AdapterCode))
|
||||
{
|
||||
var prefix = d.AdapterCode.Split(':')[0];
|
||||
// 从 gateway_nodes 查找对应的 BaseUrl
|
||||
try
|
||||
{
|
||||
var gw = _devRepo.DbContext.Queryable<gateway_nodes>()
|
||||
.First(x => x.AdapterTypes != null && x.AdapterTypes.Contains(prefix) && x.IsOnline == "在线");
|
||||
baseUrl = gw?.BaseUrl ?? "";
|
||||
}
|
||||
catch { }
|
||||
}
|
||||
map[d.DeviceId] = (d.AdapterCode ?? "", d.SourceId ?? "", baseUrl);
|
||||
}
|
||||
return map;
|
||||
}
|
||||
|
||||
// ═══════════════════════════════════════════
|
||||
// 批量实时值获取
|
||||
// ═══════════════════════════════════════════
|
||||
|
||||
private async Task<Dictionary<(string adapter, string sourceId), List<(int pointIndex, double value)>>> BatchFetchRealtimeAsync(
|
||||
List<warehouse_rule> rules,
|
||||
Dictionary<int, (string adapterCode, string sourceId, string baseUrl)> deviceMap)
|
||||
{
|
||||
var result = new Dictionary<(string, string), List<(int, double)>>();
|
||||
|
||||
// 按网关分组
|
||||
var gwGroups = new Dictionary<string, List<(string adapter, string sourceId)>>();
|
||||
foreach (var (deviceId, (adapter, sourceId, baseUrl)) in deviceMap)
|
||||
{
|
||||
if (string.IsNullOrEmpty(baseUrl) || string.IsNullOrEmpty(sourceId)) continue;
|
||||
if (!gwGroups.ContainsKey(baseUrl))
|
||||
gwGroups[baseUrl] = new();
|
||||
gwGroups[baseUrl].Add((adapter, sourceId));
|
||||
}
|
||||
|
||||
foreach (var (baseUrl, pairs) in gwGroups)
|
||||
{
|
||||
foreach (var (adapter, sourceId) in pairs)
|
||||
{
|
||||
try
|
||||
{
|
||||
var data = await _gatewayClient.GetRealtimeAsync(baseUrl, adapter, sourceId);
|
||||
if (data == null) continue;
|
||||
var root = data.RootElement;
|
||||
if (root.TryGetProperty("rows", out var rows) && rows.ValueKind == JsonValueKind.Array)
|
||||
{
|
||||
var list = new List<(int, double)>();
|
||||
foreach (var r in rows.EnumerateArray())
|
||||
{
|
||||
int idx = r.TryGetProperty("index", out var i) ? i.GetInt32() : 0;
|
||||
double val = r.TryGetProperty("value", out var v) ? v.GetDouble() : 0;
|
||||
list.Add((idx, val));
|
||||
}
|
||||
result[(adapter, sourceId)] = list;
|
||||
}
|
||||
}
|
||||
catch { }
|
||||
}
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
// ═══════════════════════════════════════════
|
||||
// 条件评估
|
||||
// ═══════════════════════════════════════════
|
||||
|
||||
private Task<bool> EvaluateRuleAsync(warehouse_rule rule,
|
||||
Dictionary<(string adapter, string sourceId), List<(int pointIndex, double value)>> realtimeData,
|
||||
Dictionary<int, (string adapterCode, string sourceId, string baseUrl)> deviceMap)
|
||||
{
|
||||
var conditions = rule.warehouse_rulecondition ?? new();
|
||||
if (!conditions.Any()) return Task.FromResult(false);
|
||||
|
||||
var results = new List<bool>();
|
||||
foreach (var cond in conditions)
|
||||
{
|
||||
if (cond.DeviceId == null || cond.ValueId == null) { results.Add(false); continue; }
|
||||
|
||||
// 冷却检查
|
||||
if (cond.LastTriggered.HasValue && rule.CooldownSec > 0)
|
||||
{
|
||||
if ((DateTime.Now - cond.LastTriggered.Value).TotalSeconds < rule.CooldownSec)
|
||||
{ results.Add(false); continue; }
|
||||
}
|
||||
|
||||
if (!deviceMap.TryGetValue(cond.DeviceId.Value, out var devInfo)) { results.Add(false); continue; }
|
||||
|
||||
double? actualValue = null;
|
||||
if (realtimeData.TryGetValue((devInfo.adapterCode, devInfo.sourceId), out var points))
|
||||
{
|
||||
// ValueId 对应 pointIndex(简化:直接使用 ValueId 作为 pointIndex)
|
||||
var point = points.FirstOrDefault(p => p.pointIndex == cond.ValueId);
|
||||
actualValue = point.pointIndex == 0 && !points.Any(p => p.pointIndex == cond.ValueId) && points.Count > 0
|
||||
? points.First().value : point.value;
|
||||
if (point.pointIndex == 0 && points.Count == 0) actualValue = null;
|
||||
}
|
||||
|
||||
// 滞后窗:已触发过则用恢复阈值
|
||||
bool isTriggered = cond.LastTriggered.HasValue;
|
||||
double target = isTriggered
|
||||
? (double)(cond.RecoveryThreshold_Numeric ?? cond.TargetValue_Number ?? 0)
|
||||
: (double)(cond.TargetValue_Number ?? 0);
|
||||
|
||||
bool met = Compare(actualValue, cond.CompareOperator ?? "大于", target);
|
||||
results.Add(met);
|
||||
if (met) cond.LastTriggered = DateTime.Now;
|
||||
}
|
||||
|
||||
bool finalResult = rule.JudgmentMode == "AND"
|
||||
? results.All(r => r)
|
||||
: results.Any(r => r);
|
||||
return Task.FromResult(finalResult);
|
||||
}
|
||||
|
||||
private static bool Compare(double? actual, string op, double target)
|
||||
{
|
||||
double v = actual ?? double.MinValue;
|
||||
return op switch
|
||||
{
|
||||
"大于" => v > target,
|
||||
"小于" => v < target,
|
||||
"等于" => Math.Abs(v - target) < 0.001,
|
||||
"大于等于" => v >= target,
|
||||
"小于等于" => v <= target,
|
||||
"不等于" => Math.Abs(v - target) > 0.001,
|
||||
_ => false
|
||||
};
|
||||
}
|
||||
|
||||
// ═══════════════════════════════════════════
|
||||
// 动作执行
|
||||
// ═══════════════════════════════════════════
|
||||
|
||||
private async Task ExecuteActionsAsync(warehouse_rule rule,
|
||||
Dictionary<int, (string adapterCode, string sourceId, string baseUrl)> deviceMap)
|
||||
{
|
||||
var actions = rule.warehouse_ruleaction ?? new();
|
||||
if (!actions.Any()) return;
|
||||
|
||||
// 冷却检查
|
||||
if (rule.LastTriggered.HasValue && rule.CooldownSec > 0)
|
||||
{
|
||||
if ((DateTime.Now - rule.LastTriggered.Value).TotalSeconds < rule.CooldownSec)
|
||||
return;
|
||||
}
|
||||
|
||||
var tasks = actions.Select(a => ExecuteSingleActionAsync(a, deviceMap));
|
||||
await Task.WhenAll(tasks);
|
||||
}
|
||||
|
||||
private async Task ExecuteSingleActionAsync(warehouse_ruleaction action,
|
||||
Dictionary<int, (string adapterCode, string sourceId, string baseUrl)> deviceMap)
|
||||
{
|
||||
var actionType = action.ActionType ?? action.Type ?? "控制";
|
||||
try
|
||||
{
|
||||
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5));
|
||||
|
||||
switch (actionType)
|
||||
{
|
||||
case "控制":
|
||||
if (action.DeviceId.HasValue && deviceMap.TryGetValue(action.DeviceId.Value, out var dev))
|
||||
{
|
||||
if (!string.IsNullOrEmpty(dev.baseUrl))
|
||||
{
|
||||
var pointIndex = action.ValueId ?? 0;
|
||||
var value = (double)(action.TargetValue_Switch == "开" ? 1 : action.TargetValue_Number ?? 0);
|
||||
await _gatewayClient.ControlDeviceAsync(dev.baseUrl, dev.adapterCode, dev.sourceId, pointIndex, value);
|
||||
}
|
||||
}
|
||||
break;
|
||||
|
||||
case "告警":
|
||||
if (action.Alert == "是" && action.DeviceId.HasValue)
|
||||
{
|
||||
var alarm = new iot_alarm
|
||||
{
|
||||
SourceAlarmId = $"rule-{action.RuleID}-{DateTime.Now.Ticks}",
|
||||
DeviceId = action.DeviceId.Value,
|
||||
AlarmLevel = "重要",
|
||||
AlarmDesc = action.AlertMessage ?? $"规则触发",
|
||||
StartTime = DateTime.Now,
|
||||
State = "未确认",
|
||||
AdapterCode = "RuleEngine",
|
||||
CreateDate = DateTime.Now
|
||||
};
|
||||
_alarmRepo.Add(alarm);
|
||||
}
|
||||
break;
|
||||
|
||||
case "通知":
|
||||
await _hub.Clients.All.SendAsync("RuleTriggered", new
|
||||
{
|
||||
title = action.AlertMessage ?? "规则触发",
|
||||
alertMessage = action.AlertMessage,
|
||||
deviceId = action.DeviceId
|
||||
}, cts.Token);
|
||||
break;
|
||||
}
|
||||
}
|
||||
catch (OperationCanceledException) { /* 超时 */ }
|
||||
catch { /* 单动作失败不阻塞 */ }
|
||||
throw new NotImplementedException(
|
||||
"RuleEngineService 待实体字段就绪。步骤: SQL ALTER TABLE → 代码生成器 → git revert 本桩。");
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,5 +1,7 @@
|
||||
using Quartz;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.Configuration;
|
||||
using System.Net.Http;
|
||||
using Warehouse.IServices;
|
||||
using VolPro.Entity.DomainModels;
|
||||
using System;
|
||||
@@ -21,7 +23,9 @@ public class SyncDevicesJob : IJob
|
||||
{
|
||||
var sp = _sp;
|
||||
var gwSvc = sp.GetService<Igateway_nodesService>();
|
||||
var client = sp.GetService<GatewayClient>();
|
||||
var httpFactory = sp.GetService<IHttpClientFactory>();
|
||||
var config = sp.GetService<IConfiguration>();
|
||||
var client = httpFactory != null ? new GatewayClient(httpFactory, config!) : null;
|
||||
if (gwSvc == null || client == null) return;
|
||||
|
||||
// 遍历所有在线且启用的网关
|
||||
|
||||
@@ -24,10 +24,13 @@ using System.Text.Json;
|
||||
|
||||
namespace Warehouse.Services
|
||||
{
|
||||
/// <summary>
|
||||
/// gateway_nodes 业务逻辑(partial)。注册/心跳/设备同步。
|
||||
/// </summary>
|
||||
public partial class gateway_nodesService
|
||||
{
|
||||
private readonly IHttpContextAccessor _httpContextAccessor;
|
||||
private readonly Igateway_nodesRepository _repository;//访问数据库
|
||||
private readonly Igateway_nodesRepository _repository;
|
||||
|
||||
[ActivatorUtilitiesConstructor]
|
||||
public gateway_nodesService(
|
||||
@@ -38,24 +41,22 @@ namespace Warehouse.Services
|
||||
{
|
||||
_httpContextAccessor = httpContextAccessor;
|
||||
_repository = dbRepository;
|
||||
//多租户会用到这init代码,其他情况可以不用
|
||||
//base.Init(dbRepository);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 网关注册(Upsert)。
|
||||
/// NodeCode 匹配则更新适配器类型/地址/在线状态并返回已有 NodeId,
|
||||
/// NodeCode 匹配则更新适配器类型/地址/在线状态;
|
||||
/// NodeCode 不匹配且 Token 验证通过则插入新记录。
|
||||
/// </summary>
|
||||
[Obsolete("由 A1 API Controller 自动调用,不建议手动调用")]
|
||||
public async Task<gateway_nodes> RegisterNodeAsync(string nodeCode, string token, string adapterTypes, string baseUrl)
|
||||
{
|
||||
var existing = await _repository.FindAsIQueryable<gateway_nodes>()
|
||||
.FirstOrDefaultAsync(x => x.NodeCode == nodeCode);
|
||||
var existingList = await _repository.FindAsIQueryable(x => x.NodeCode == nodeCode).ToListAsync();
|
||||
var existing = existingList.FirstOrDefault();
|
||||
|
||||
gateway_nodes entity;
|
||||
if (existing != null)
|
||||
{
|
||||
// 已存在:验证Token,更新网关上报字段
|
||||
if (existing.NodeToken != token)
|
||||
throw new UnauthorizedAccessException("NodeToken 不匹配");
|
||||
|
||||
@@ -68,7 +69,6 @@ namespace Warehouse.Services
|
||||
}
|
||||
else
|
||||
{
|
||||
// 新节点:直接插入
|
||||
entity = new gateway_nodes
|
||||
{
|
||||
NodeCode = nodeCode,
|
||||
@@ -89,10 +89,11 @@ namespace Warehouse.Services
|
||||
/// <summary>
|
||||
/// 心跳更新。更新 LastHeartbeat 并标记在线。
|
||||
/// </summary>
|
||||
[Obsolete("由 A2 API Controller 自动调用,不建议手动调用")]
|
||||
public async Task UpdateHeartbeatAsync(string nodeCode, string token)
|
||||
{
|
||||
var entity = _repository.FindAsIQueryable<gateway_nodes>()
|
||||
.FirstOrDefaultAsync(x => x.NodeCode == nodeCode && x.NodeToken == token);
|
||||
var entityList = await _repository.FindAsIQueryable(x => x.NodeCode == nodeCode && x.NodeToken == token).ToListAsync();
|
||||
var entity = entityList.FirstOrDefault();
|
||||
if (entity == null)
|
||||
throw new UnauthorizedAccessException("认证失败:NodeCode 或 Token 无效");
|
||||
|
||||
@@ -102,15 +103,15 @@ namespace Warehouse.Services
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 设备数据同步。按照字段分治原则写入 base_device:
|
||||
/// 首次入库写全量,后续仅更新网关字段(IsOnline/ExtraData/ParentDeviceId等)。
|
||||
/// 设备数据同步。按字段分治原则写入 base_device:
|
||||
/// 首次入库写全量,后续仅更新网关字段。
|
||||
/// parentSourceId 解析为 ParentDeviceId。
|
||||
/// </summary>
|
||||
[Obsolete("由 A3 API Controller 自动调用,不建议手动调用")]
|
||||
public async Task<(int added, int updated)> SyncDevicesAsync(int gatewayNodeId, List<SyncDeviceItem> devices)
|
||||
{
|
||||
var db = _repository.DbContext;
|
||||
|
||||
// 批量查询已有设备映射表(用于 parentSourceId → ParentDeviceId 解析)
|
||||
var adapterCodes = devices.Select(d => d.AdapterCode).Distinct().ToList();
|
||||
var existingIds = db.Queryable<base_device>()
|
||||
.Where(x => x.NodeId == gatewayNodeId && adapterCodes.Contains(x.AdapterCode))
|
||||
@@ -124,7 +125,6 @@ namespace Warehouse.Services
|
||||
existingIds.TryGetValue(key, out var existingId);
|
||||
bool isNew = existingId == 0;
|
||||
|
||||
// 解析 parentSourceId → ParentDeviceId
|
||||
int? parentDeviceId = null;
|
||||
if (!string.IsNullOrEmpty(d.ParentSourceId))
|
||||
{
|
||||
@@ -134,7 +134,6 @@ namespace Warehouse.Services
|
||||
|
||||
if (isNew)
|
||||
{
|
||||
// 首次入库写全量
|
||||
var entity = new base_device
|
||||
{
|
||||
DeviceName = d.Name ?? $"DEV_{d.SourceId}",
|
||||
@@ -158,7 +157,6 @@ namespace Warehouse.Services
|
||||
}
|
||||
else
|
||||
{
|
||||
// 已有记录:仅更新网关字段
|
||||
var entity = db.Queryable<base_device>().InSingle(existingId);
|
||||
if (entity != null)
|
||||
{
|
||||
@@ -178,7 +176,7 @@ namespace Warehouse.Services
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>网关同步设备条目(A3 接口接收的数据模型)</summary>
|
||||
/// <summary>网关同步设备条目</summary>
|
||||
public class SyncDeviceItem
|
||||
{
|
||||
public string AdapterCode { get; set; } = "";
|
||||
|
||||
Reference in New Issue
Block a user