diff --git a/src/SJZY.Track.Application/CAP/EventHandlers/TrackDeleteEventHandler.cs b/src/SJZY.Track.Application/CAP/EventHandlers/TrackDeleteEventHandler.cs index 89207bee..83c5bf61 100644 --- a/src/SJZY.Track.Application/CAP/EventHandlers/TrackDeleteEventHandler.cs +++ b/src/SJZY.Track.Application/CAP/EventHandlers/TrackDeleteEventHandler.cs @@ -367,11 +367,7 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe private void WarningCalc_DeletedTrack(List delTracks) { - var signals = delTracks.Select(x => new TrackDeletedSignal - { - OrderId = x.OrderId, - DeletedTrack = x - }).ToList(); + var signals = TrackSignal.FromTrack(delTracks, WarningSignalType.轨迹删除); _ = Task.Run(async () => { @@ -385,7 +381,7 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe await Task.Delay(1500); //确保落盘 await handler.TrackDeletedHandleAsync(signals); await handler2.TrackDeletedHandleAsync(signals); - await handler3.TrackUpdatedOrDeletedHandleAsync(signals.Cast().ToList()); + await handler3.TrackUpdatedOrDeletedHandleAsync(signals); } finally { diff --git a/src/SJZY.Track.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs b/src/SJZY.Track.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs index e1f0402b..0ee94b21 100644 --- a/src/SJZY.Track.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs +++ b/src/SJZY.Track.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs @@ -511,12 +511,7 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe private void WarningCalc_AddedTrack(List tracks) { - var signals = tracks.Select(x => new TrackAddedSignal - { - EsId = x.Id, - OrderId = x.OrderId, - TrackStatusId = x.TrackStatusId - }).ToList(); + var signals = TrackSignal.FromTrack(tracks, WarningSignalType.轨迹新增); _ = Task.Run(async () => { @@ -528,7 +523,7 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe await Task.Delay(1500); //确保落盘 await handler.TrackAddedHandleAsync(signals); - await handler2.TrackAddedOrUpdatedHandleAsync(signals.Cast().ToList()); + await handler2.TrackAddedOrUpdatedHandleAsync(signals); } finally { @@ -540,11 +535,8 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe private void WarningCalc_UpdatedTrack(List tracks) { - var signals = tracks.Select(x => new TrackUpdatedSignal - { - OrderId = x.OrderId, - BeforeChangeTrack = x - }).ToList(); + var signals = TrackSignal.FromTrack(tracks, WarningSignalType.轨迹更新); + _ = Task.Run(async () => { var scope = _serviceProvider.CreateAsyncScope(); @@ -554,8 +546,8 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe var handler2 = scope.ServiceProvider.GetRequiredService(); await Task.Delay(1500); //确保落盘 - await handler.TrackAddedOrUpdatedHandleAsync(signals.Cast().ToList()); - await handler2.TrackUpdatedOrDeletedHandleAsync(signals.Cast().ToList()); + await handler.TrackAddedOrUpdatedHandleAsync(signals); + await handler2.TrackUpdatedOrDeletedHandleAsync(signals); } finally { diff --git a/src/SJZY.Track.Application/Services/TimerTaskss/OrderPullStatusScanTask.cs b/src/SJZY.Track.Application/Services/TimerTaskss/OrderPullStatusScanTask.cs index 8d3bdfd1..07a52858 100644 --- a/src/SJZY.Track.Application/Services/TimerTaskss/OrderPullStatusScanTask.cs +++ b/src/SJZY.Track.Application/Services/TimerTaskss/OrderPullStatusScanTask.cs @@ -27,6 +27,7 @@ public class OrderPullStatusScanTask : ApplicationService /// 定时任务对订单拉取状态进行扫描，计算更新拉取状态 /// /// + /// todo:标识为AllowAnonymous进行数据操作时拿不到用户信息，出现异常 [AllowAnonymous] [HttpPost("/TimerTask/OrderPullStatusScan")] public async Task OrderPullStatusScanAsync() diff --git a/src/SJZY.Track.Application/Services/Warning/Handlers/CompleteWarningHandler.cs b/src/SJZY.Track.Application/Services/Warning/Handlers/CompleteWarningHandler.cs index d7f69306..1d95537e 100644 --- a/src/SJZY.Track.Application/Services/Warning/Handlers/CompleteWarningHandler.cs +++ b/src/SJZY.Track.Application/Services/Warning/Handlers/CompleteWarningHandler.cs @@ -31,7 +31,7 @@ public sealed class CompleteWarningHandler : ICompleteWarningHandler, ITransient #region 预警信号批处理-删除轨迹 - public async Task TrackDeletedHandleAsync(List deletedSignals) + public async Task TrackDeletedHandleAsync(List deletedSignals) { // 1.只处理未删除的订单 var orderIdSet = deletedSignals.Select(x => x.OrderId).ToHashSet(); @@ -118,7 +118,7 @@ public sealed class CompleteWarningHandler : ICompleteWarningHandler, ITransient #region todo: 预警信号批处理-新增轨迹 - public async Task TrackAddedHandleAsync(List addedSignals) + public async Task TrackAddedHandleAsync(List addedSignals) { // 1.只处理未删除的订单 var orderIdSet = addedSignals.Select(x => x.OrderId).ToHashSet(); diff --git a/src/SJZY.Track.Application/Services/Warning/Handlers/ICompleteWarningHandler.cs b/src/SJZY.Track.Application/Services/Warning/Handlers/ICompleteWarningHandler.cs index 858805a1..a9444cf2 100644 --- a/src/SJZY.Track.Application/Services/Warning/Handlers/ICompleteWarningHandler.cs +++ b/src/SJZY.Track.Application/Services/Warning/Handlers/ICompleteWarningHandler.cs @@ -7,10 +7,10 @@ namespace SJZY.Track.Application.Services.Warning.Handlers; public interface ICompleteWarningHandler { // 批量处理轨迹 删除信号 - Task TrackDeletedHandleAsync(List deletedSignals); + Task TrackDeletedHandleAsync(List deletedSignals); // 批量处理轨迹 新增信号 - Task TrackAddedHandleAsync(List addedSignals); + Task TrackAddedHandleAsync(List addedSignals); // 完整预警主体禁用 Task WarningRuleToDisabledHandleAsync(WarningRuleEntity rule); diff --git a/src/SJZY.Track.Application/Services/Warning/Handlers/IReturnWarningHandler.cs b/src/SJZY.Track.Application/Services/Warning/Handlers/IReturnWarningHandler.cs index bd5621fd..91f8ee79 100644 --- a/src/SJZY.Track.Application/Services/Warning/Handlers/IReturnWarningHandler.cs +++ b/src/SJZY.Track.Application/Services/Warning/Handlers/IReturnWarningHandler.cs @@ -9,7 +9,7 @@ namespace SJZY.Track.Application.Services.Warning.Handlers; public interface IReturnWarningHandler { // 批量处理轨迹 修改/删除 - Task TrackUpdatedOrDeletedHandleAsync(List updateOrDeleteSignals); + Task TrackUpdatedOrDeletedHandleAsync(List updateOrDeleteSignals); // 收到轨迹推送记录（接口通回调）（人工介入/重试 阶段的回传） Task ReceiveTrackPushRecordHandleAsync(bool isSuccess, string requestParams, List records, List pushInputs, List updateInputs); diff --git a/src/SJZY.Track.Application/Services/Warning/Handlers/ITransportWarningHandler.cs b/src/SJZY.Track.Application/Services/Warning/Handlers/ITransportWarningHandler.cs index f54107b2..61b8c4b8 100644 --- a/src/SJZY.Track.Application/Services/Warning/Handlers/ITransportWarningHandler.cs +++ b/src/SJZY.Track.Application/Services/Warning/Handlers/ITransportWarningHandler.cs @@ -11,10 +11,10 @@ namespace SJZY.Track.Application.Services.Warning.Handlers; public interface ITransportWarningHandler { // 批量处理轨迹删除信号 - Task TrackDeletedHandleAsync(List deletedSignals); + Task TrackDeletedHandleAsync(List deletedSignals); // 批量处理轨迹新增/修改 - Task TrackAddedOrUpdatedHandleAsync(List addOrUpdateSignals); + Task TrackAddedOrUpdatedHandleAsync(List addOrUpdateSignals); // 运输预警主体禁用 Task WarningRuleToDisabledHandleAsync(WarningRuleEntity rule); diff --git a/src/SJZY.Track.Application/Services/Warning/Handlers/ReturnWarningHandler.cs b/src/SJZY.Track.Application/Services/Warning/Handlers/ReturnWarningHandler.cs index 954c77eb..bc9486d1 100644 --- a/src/SJZY.Track.Application/Services/Warning/Handlers/ReturnWarningHandler.cs +++ b/src/SJZY.Track.Application/Services/Warning/Handlers/ReturnWarningHandler.cs @@ -384,8 +384,7 @@ public sealed class ReturnWarningHandler : IReturnWarningHandler, ITransientDepe #region todo-test: 预警信号批处理-修改或删除轨迹 - // todo:可以优化为直接通过轨迹的esId来寻找明细 - public async Task TrackUpdatedOrDeletedHandleAsync(List updateOrDeleteSignals) + public async Task TrackUpdatedOrDeletedHandleAsync(List updateOrDeleteSignals) { // 1.查询订单 var orderIdSet = updateOrDeleteSignals.Select(x => x.OrderId).ToHashSet(); @@ -396,19 +395,13 @@ public sealed class ReturnWarningHandler : IReturnWarningHandler, ITransientDepe var orderIds = orders.Select(x => x.Id).ToHashSet(); // 1.1 构建订单和轨迹字典 - var orderIdTracksDict = updateOrDeleteSignals + var orderIdSignalDict = updateOrDeleteSignals .Where(x => orderIds.Contains(x.OrderId)) // tips:信号过滤 .GroupBy(x => x.OrderId) - .ToDictionary(x => x.Key, - y => y.Select(x => x.Type == WarningSignalType.轨迹更新 - ? ((TrackUpdatedSignal) x).BeforeChangeTrack - : ((TrackDeletedSignal) x).DeletedTrack - ).ToList() - ); - if (orderIdTracksDict.Count == 0) return; //无需进行处理，直接提交信号丢弃 + .ToDictionary(x => x.Key, y => y.ToList()); // 1.2 构建订单和轨迹内容id字典 - var orderIdTrackContentIdsDict = orderIdTracksDict.ToDictionary( + var orderIdTrackContentIdsDict = orderIdSignalDict.ToDictionary( x => x.Key, y => y.Value.Select(x => x.TrackContentId).ToHashSet() ); @@ -420,20 +413,13 @@ public sealed class ReturnWarningHandler : IReturnWarningHandler, ITransientDepe }); if (warningDetails.Count == 0) return; - // 3.轨迹匹配预警明细，如果有比较字段则使用比较字段，否则使用时间 + // 3.轨迹匹配预警明细，使用esId var closeDetailIds = new HashSet(); foreach (var group in warningDetails.GroupBy(x => x.OrderId)) { - var tracks = orderIdTracksDict[group.Key]; - foreach (var detail in group) - { - var isMatch = tracks.Any(t => string.IsNullOrWhiteSpace(detail.CompareFields) - ? detail.MatchTrackTime(t.TrackTime, t.TrackTimeZone) - : detail.MatchExtra(detail.CompareFields, t.Extra) - ); - if (isMatch) - closeDetailIds.Add(detail.Id); - } + var trackEsIds = orderIdSignalDict[group.Key].Select(x => x.EsId).ToHashSet(); + var detailIds = group.Where(x => trackEsIds.Contains(x.TrackEsId)).Select(x => x.Id); + closeDetailIds.UnionWith(detailIds); } if (closeDetailIds.Count == 0) return; diff --git a/src/SJZY.Track.Application/Services/Warning/Handlers/TransportWarningHandler.cs b/src/SJZY.Track.Application/Services/Warning/Handlers/TransportWarningHandler.cs index ecb9b572..d5cad8ca 100644 --- a/src/SJZY.Track.Application/Services/Warning/Handlers/TransportWarningHandler.cs +++ b/src/SJZY.Track.Application/Services/Warning/Handlers/TransportWarningHandler.cs @@ -42,7 +42,7 @@ public class TransportWarningHandler : ITransportWarningHandler, ITransientDepen // tips:删除轨迹信号在运输预警处理只关注被删除的轨迹的状态 // 可能影响【运输预警明细】【运输预警明细生成任务】 - public async Task TrackDeletedHandleAsync(List deletedSignals) + public async Task TrackDeletedHandleAsync(List deletedSignals) { // 1.查询订单 var orderIdSet = deletedSignals.Select(x => x.OrderId).ToHashSet(); @@ -57,7 +57,7 @@ public class TransportWarningHandler : ITransportWarningHandler, ITransientDepen var orderIdTrackStatusIdsDict = deletedSignals .Where(s => orderIds.Contains(s.OrderId)) // tips:过滤掉无需处理的信号 .GroupBy(x => x.OrderId) - .ToDictionary(x => x.Key, y => y.Select(z => z.DeletedTrack.TrackStatusId).ToHashSet()); + .ToDictionary(x => x.Key, y => y.Select(z => z.TrackStatusId).ToHashSet()); // 2.查询订单每个状态最早轨迹的字点(tips：由于已经是删除，查询出的轨迹不会包含当前已经删除的轨迹) var orderIdTrackStatusIdEarliestDict = await _trackDetailsEsRepository.GetTrackStatusEarliestDictAsync(orderIds); @@ -199,7 +199,7 @@ public class TransportWarningHandler : ITransportWarningHandler, ITransientDepen #region 预警信号批处理-新增修改轨迹 // tips:新增修改轨迹信号，在运输处理关注最新的轨迹 - public async Task TrackAddedOrUpdatedHandleAsync(List addOrUpdateSignals) + public async Task TrackAddedOrUpdatedHandleAsync(List addOrUpdateSignals) { // 1.查询订单 var orderIdSet = addOrUpdateSignals.Select(x => x.OrderId).ToHashSet(); diff --git a/src/SJZY.Track.Domain/Aggregates/WarningAggregate/Signals/IWarningSignal.cs b/src/SJZY.Track.Domain/Aggregates/WarningAggregate/Signals/IWarningSignal.cs deleted file mode 100644 index 4e6cf15e..00000000 --- a/src/SJZY.Track.Domain/Aggregates/WarningAggregate/Signals/IWarningSignal.cs +++ /dev/null @@ -1,25 +0,0 @@ -namespace SJZY.Track.Domain.Aggregates.WarningAggregate.Signals; - -public class IWarningSignal -{ - public int OrderId { get; set; } - public virtual WarningSignalType Type { get; set; } -} - -public enum WarningSignalType -{ - 轨迹新增 = 0, - - 轨迹删除 = 1, - - 轨迹更新 = 2, -} - -public class WarningSignal -{ - public string MessageId { get; set; } = Guid.NewGuid().ToString(); - public string OrderId { get; set; } // 分区键 - public WarningSignalType Type { get; set; } // 新增/修改/删除 - public DateTime Timestamp { get; set; } = DateTime.UtcNow; - public Dictionary Payload { get; set; } -} diff --git a/src/SJZY.Track.Domain/Aggregates/WarningAggregate/Signals/TrackAddedSignal.cs b/src/SJZY.Track.Domain/Aggregates/WarningAggregate/Signals/TrackAddedSignal.cs deleted file mode 100644 index 5537893a..00000000 --- a/src/SJZY.Track.Domain/Aggregates/WarningAggregate/Signals/TrackAddedSignal.cs +++ /dev/null @@ -1,13 +0,0 @@ -namespace SJZY.Track.Domain.Aggregates.WarningAggregate.Signals; - -/// -/// 轨迹新增信号 -/// -public sealed class TrackAddedSignal : IWarningSignal -{ - public override WarningSignalType Type { get; set; } = WarningSignalType.轨迹新增; - - public int TrackStatusId { get; set; } - - public string EsId { get; set; } //新增轨迹的esId -} diff --git a/src/SJZY.Track.Domain/Aggregates/WarningAggregate/Signals/TrackDeletedSignal.cs b/src/SJZY.Track.Domain/Aggregates/WarningAggregate/Signals/TrackDeletedSignal.cs deleted file mode 100644 index cb52ebae..00000000 --- a/src/SJZY.Track.Domain/Aggregates/WarningAggregate/Signals/TrackDeletedSignal.cs +++ /dev/null @@ -1,13 +0,0 @@ -using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate; - -namespace SJZY.Track.Domain.Aggregates.WarningAggregate.Signals; - -/// -/// 轨迹删除信号 -/// -public class TrackDeletedSignal : IWarningSignal -{ - public override WarningSignalType Type { get; set; } = WarningSignalType.轨迹删除; - - public TrackDetailsIndex DeletedTrack { get; set; } //已删除的轨迹 -} \ No newline at end of file diff --git a/src/SJZY.Track.Domain/Aggregates/WarningAggregate/Signals/TrackSignal.cs b/src/SJZY.Track.Domain/Aggregates/WarningAggregate/Signals/TrackSignal.cs new file mode 100644 index 00000000..00389cd9 --- /dev/null +++ b/src/SJZY.Track.Domain/Aggregates/WarningAggregate/Signals/TrackSignal.cs @@ -0,0 +1,43 @@ +using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate; + +namespace SJZY.Track.Domain.Aggregates.WarningAggregate.Signals; + +/// +/// 轨迹信号，操作轨迹时产生 +/// 用于进行预警计算 +/// +public readonly struct TrackSignal +{ + public WarningSignalType Type { get; } + + public int OrderId { get; } + // public string OrderNo { get; set; } + // public int OrderType { get; set; } + // public int SubSystemId { get; set; } + + public string EsId { get; } + public int TrackStatusId { get; } + public int TrackContentId { get; } + + public TrackSignal(WarningSignalType type, int orderId, string esId, int trackStatusId, int trackContentId) + { + Type = type; + OrderId = orderId; + EsId = esId; + TrackStatusId = trackStatusId; + TrackContentId = trackContentId; + } + + public static List FromTrack(List tracks, WarningSignalType type) + { + return tracks + .Select(x => new TrackSignal( + type, + x.OrderId, + x.Id, + x.TrackStatusId, + x.TrackContentId + )) + .ToList(); + } +} diff --git a/src/SJZY.Track.Domain/Aggregates/WarningAggregate/Signals/TrackUpdatedSignal.cs b/src/SJZY.Track.Domain/Aggregates/WarningAggregate/Signals/TrackUpdatedSignal.cs deleted file mode 100644 index 11de126b..00000000 --- a/src/SJZY.Track.Domain/Aggregates/WarningAggregate/Signals/TrackUpdatedSignal.cs +++ /dev/null @@ -1,13 +0,0 @@ -using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate; - -namespace SJZY.Track.Domain.Aggregates.WarningAggregate.Signals; - -/// -/// 轨迹更新信号 -/// -public class TrackUpdatedSignal : IWarningSignal -{ - public override WarningSignalType Type { get; set; } = WarningSignalType.轨迹更新; - - public TrackDetailsIndex BeforeChangeTrack { get; set; } //更改前的轨迹明细 -} \ No newline at end of file diff --git a/src/SJZY.Track.Domain/Aggregates/WarningAggregate/Signals/WarningSignalType.cs b/src/SJZY.Track.Domain/Aggregates/WarningAggregate/Signals/WarningSignalType.cs new file mode 100644 index 00000000..31f8dec1 --- /dev/null +++ b/src/SJZY.Track.Domain/Aggregates/WarningAggregate/Signals/WarningSignalType.cs @@ -0,0 +1,10 @@ +namespace SJZY.Track.Domain.Aggregates.WarningAggregate.Signals; + +public enum WarningSignalType +{ + 轨迹新增 = 0, + + 轨迹删除 = 1, + + 轨迹更新 = 2, +}