Merge branch 'S07' into test
This commit is contained in:
commit
c4732e711c
@ -367,11 +367,7 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe
|
||||
|
||||
private void WarningCalc_DeletedTrack(List<TrackDetailsIndex> 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<IWarningSignal>().ToList());
|
||||
await handler3.TrackUpdatedOrDeletedHandleAsync(signals);
|
||||
}
|
||||
finally
|
||||
{
|
||||
|
||||
@ -511,12 +511,7 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe
|
||||
|
||||
private void WarningCalc_AddedTrack(List<TrackDetailsIndex> 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<IWarningSignal>().ToList());
|
||||
await handler2.TrackAddedOrUpdatedHandleAsync(signals);
|
||||
}
|
||||
finally
|
||||
{
|
||||
@ -540,11 +535,8 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe
|
||||
|
||||
private void WarningCalc_UpdatedTrack(List<TrackDetailsIndex> 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<IReturnWarningHandler>();
|
||||
|
||||
await Task.Delay(1500); //确保落盘
|
||||
await handler.TrackAddedOrUpdatedHandleAsync(signals.Cast<IWarningSignal>().ToList());
|
||||
await handler2.TrackUpdatedOrDeletedHandleAsync(signals.Cast<IWarningSignal>().ToList());
|
||||
await handler.TrackAddedOrUpdatedHandleAsync(signals);
|
||||
await handler2.TrackUpdatedOrDeletedHandleAsync(signals);
|
||||
}
|
||||
finally
|
||||
{
|
||||
|
||||
@ -27,6 +27,7 @@ public class OrderPullStatusScanTask : ApplicationService
|
||||
/// 定时任务对订单拉取状态进行扫描,计算更新拉取状态
|
||||
/// </summary>
|
||||
/// <returns></returns>
|
||||
/// todo:标识为AllowAnonymous进行数据操作时拿不到用户信息,出现异常
|
||||
[AllowAnonymous]
|
||||
[HttpPost("/TimerTask/OrderPullStatusScan")]
|
||||
public async Task<string> OrderPullStatusScanAsync()
|
||||
|
||||
@ -31,7 +31,7 @@ public sealed class CompleteWarningHandler : ICompleteWarningHandler, ITransient
|
||||
|
||||
#region 预警信号批处理-删除轨迹
|
||||
|
||||
public async Task TrackDeletedHandleAsync(List<TrackDeletedSignal> deletedSignals)
|
||||
public async Task TrackDeletedHandleAsync(List<TrackSignal> 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<TrackAddedSignal> addedSignals)
|
||||
public async Task TrackAddedHandleAsync(List<TrackSignal> addedSignals)
|
||||
{
|
||||
// 1.只处理未删除的订单
|
||||
var orderIdSet = addedSignals.Select(x => x.OrderId).ToHashSet();
|
||||
|
||||
@ -7,10 +7,10 @@ namespace SJZY.Track.Application.Services.Warning.Handlers;
|
||||
public interface ICompleteWarningHandler
|
||||
{
|
||||
// 批量处理轨迹 删除信号
|
||||
Task TrackDeletedHandleAsync(List<TrackDeletedSignal> deletedSignals);
|
||||
Task TrackDeletedHandleAsync(List<TrackSignal> deletedSignals);
|
||||
|
||||
// 批量处理轨迹 新增信号
|
||||
Task TrackAddedHandleAsync(List<TrackAddedSignal> addedSignals);
|
||||
Task TrackAddedHandleAsync(List<TrackSignal> addedSignals);
|
||||
|
||||
// 完整预警主体禁用
|
||||
Task WarningRuleToDisabledHandleAsync(WarningRuleEntity rule);
|
||||
|
||||
@ -9,7 +9,7 @@ namespace SJZY.Track.Application.Services.Warning.Handlers;
|
||||
public interface IReturnWarningHandler
|
||||
{
|
||||
// 批量处理轨迹 修改/删除
|
||||
Task TrackUpdatedOrDeletedHandleAsync(List<IWarningSignal> updateOrDeleteSignals);
|
||||
Task TrackUpdatedOrDeletedHandleAsync(List<TrackSignal> updateOrDeleteSignals);
|
||||
|
||||
// 收到轨迹推送记录(接口通回调)(人工介入/重试 阶段的回传)
|
||||
Task ReceiveTrackPushRecordHandleAsync(bool isSuccess, string requestParams, List<TrackPushRecordIndex> records, List<TrackPushInput> pushInputs, List<TrackUpdateInput> updateInputs);
|
||||
|
||||
@ -11,10 +11,10 @@ namespace SJZY.Track.Application.Services.Warning.Handlers;
|
||||
public interface ITransportWarningHandler
|
||||
{
|
||||
// 批量处理轨迹删除信号
|
||||
Task TrackDeletedHandleAsync(List<TrackDeletedSignal> deletedSignals);
|
||||
Task TrackDeletedHandleAsync(List<TrackSignal> deletedSignals);
|
||||
|
||||
// 批量处理轨迹新增/修改
|
||||
Task TrackAddedOrUpdatedHandleAsync(List<IWarningSignal> addOrUpdateSignals);
|
||||
Task TrackAddedOrUpdatedHandleAsync(List<TrackSignal> addOrUpdateSignals);
|
||||
|
||||
// 运输预警主体禁用
|
||||
Task WarningRuleToDisabledHandleAsync(WarningRuleEntity rule);
|
||||
|
||||
@ -384,8 +384,7 @@ public sealed class ReturnWarningHandler : IReturnWarningHandler, ITransientDepe
|
||||
|
||||
#region todo-test: 预警信号批处理-修改或删除轨迹
|
||||
|
||||
// todo:可以优化为直接通过轨迹的esId来寻找明细
|
||||
public async Task TrackUpdatedOrDeletedHandleAsync(List<IWarningSignal> updateOrDeleteSignals)
|
||||
public async Task TrackUpdatedOrDeletedHandleAsync(List<TrackSignal> 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<string>();
|
||||
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;
|
||||
|
||||
@ -42,7 +42,7 @@ public class TransportWarningHandler : ITransportWarningHandler, ITransientDepen
|
||||
|
||||
// tips:删除轨迹信号在运输预警处理只关注被删除的轨迹的状态
|
||||
// 可能影响【运输预警明细】【运输预警明细生成任务】
|
||||
public async Task TrackDeletedHandleAsync(List<TrackDeletedSignal> deletedSignals)
|
||||
public async Task TrackDeletedHandleAsync(List<TrackSignal> 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<IWarningSignal> addOrUpdateSignals)
|
||||
public async Task TrackAddedOrUpdatedHandleAsync(List<TrackSignal> addOrUpdateSignals)
|
||||
{
|
||||
// 1.查询订单
|
||||
var orderIdSet = addOrUpdateSignals.Select(x => x.OrderId).ToHashSet();
|
||||
|
||||
@ -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<string, object> Payload { get; set; }
|
||||
}
|
||||
@ -1,13 +0,0 @@
|
||||
namespace SJZY.Track.Domain.Aggregates.WarningAggregate.Signals;
|
||||
|
||||
/// <summary>
|
||||
/// 轨迹新增信号
|
||||
/// </summary>
|
||||
public sealed class TrackAddedSignal : IWarningSignal
|
||||
{
|
||||
public override WarningSignalType Type { get; set; } = WarningSignalType.轨迹新增;
|
||||
|
||||
public int TrackStatusId { get; set; }
|
||||
|
||||
public string EsId { get; set; } //新增轨迹的esId
|
||||
}
|
||||
@ -1,13 +0,0 @@
|
||||
using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate;
|
||||
|
||||
namespace SJZY.Track.Domain.Aggregates.WarningAggregate.Signals;
|
||||
|
||||
/// <summary>
|
||||
/// 轨迹删除信号
|
||||
/// </summary>
|
||||
public class TrackDeletedSignal : IWarningSignal
|
||||
{
|
||||
public override WarningSignalType Type { get; set; } = WarningSignalType.轨迹删除;
|
||||
|
||||
public TrackDetailsIndex DeletedTrack { get; set; } //已删除的轨迹
|
||||
}
|
||||
@ -0,0 +1,43 @@
|
||||
using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate;
|
||||
|
||||
namespace SJZY.Track.Domain.Aggregates.WarningAggregate.Signals;
|
||||
|
||||
/// <summary>
|
||||
/// 轨迹信号,操作轨迹时产生
|
||||
/// 用于进行预警计算
|
||||
/// </summary>
|
||||
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<TrackSignal> FromTrack(List<TrackDetailsIndex> tracks, WarningSignalType type)
|
||||
{
|
||||
return tracks
|
||||
.Select(x => new TrackSignal(
|
||||
type,
|
||||
x.OrderId,
|
||||
x.Id,
|
||||
x.TrackStatusId,
|
||||
x.TrackContentId
|
||||
))
|
||||
.ToList();
|
||||
}
|
||||
}
|
||||
@ -1,13 +0,0 @@
|
||||
using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate;
|
||||
|
||||
namespace SJZY.Track.Domain.Aggregates.WarningAggregate.Signals;
|
||||
|
||||
/// <summary>
|
||||
/// 轨迹更新信号
|
||||
/// </summary>
|
||||
public class TrackUpdatedSignal : IWarningSignal
|
||||
{
|
||||
public override WarningSignalType Type { get; set; } = WarningSignalType.轨迹更新;
|
||||
|
||||
public TrackDetailsIndex BeforeChangeTrack { get; set; } //更改前的轨迹明细
|
||||
}
|
||||
@ -0,0 +1,10 @@
|
||||
namespace SJZY.Track.Domain.Aggregates.WarningAggregate.Signals;
|
||||
|
||||
public enum WarningSignalType
|
||||
{
|
||||
轨迹新增 = 0,
|
||||
|
||||
轨迹删除 = 1,
|
||||
|
||||
轨迹更新 = 2,
|
||||
}
|
||||
Loading…
Reference in New Issue
Block a user