Merge branch 'S07' into test
This commit is contained in:
hardstylewyl
commit
88f9ea94f3
@ -8,6 +8,8 @@ using SJZY.Track.Application.Options;
|
||||
using SJZY.Track.Application.Track2InterfaceRelay;
|
||||
using SJZY.Track.Domain.Aggregates.WarningAggregate;
|
||||
using SJZY.Track.Domain.Aggregates.WarningAggregate.Return;
|
||||
using SJZY.Track.Domain.Dtos.Warning;
|
||||
using SJZY.Track.Domain.Enums;
|
||||
using SJZY.Track.Domain.Extensions;
|
||||
using SJZY.UserCenter.HttpApi.Services.Message;
|
||||
using SJZY.UserCenter.SharedModel.Dtos.Message;
|
||||
@ -22,7 +24,7 @@ public interface IWarningTaskService
|
||||
[AutomaticRetry(Attempts = 3)]
|
||||
Task CallNotifyAsync(NotifyConfig config, CancellationToken ct);
|
||||
|
||||
[AutomaticRetry(Attempts = 0)]
|
||||
[AutomaticRetry(Attempts = 3, ExceptOn = new[] { typeof(BusinessException) })]
|
||||
Task CallTrackRetryPushAsync(ReturnWarningRetryPushTaskEntity entity, CancellationToken ct);
|
||||
}
|
||||
|
||||
@ -32,12 +34,14 @@ public class WarningTaskService : IWarningTaskService, ITransientDependency
|
||||
private readonly IHttpClientProxy<IInterfaceForwardServiceV2> _interfaceForwardServiceV2;
|
||||
private readonly IOptionsMonitor<InterfaceRelayParamsConfigOptions> _options;
|
||||
private readonly IWarningDetailsEsRepository _warningDetailsEsRepository;
|
||||
private readonly IWarningTaskRepository _warningTaskRepository;
|
||||
|
||||
public WarningTaskService(IHttpClientProxy<IInterfaceForwardServiceV2> interfaceForwardServiceV2, IOptionsMonitor<InterfaceRelayParamsConfigOptions> options, IHttpClientProxy<IRequestSendMessageService> requestSendMessageService, IWarningDetailsEsRepository warningDetailsEsRepository)
|
||||
public WarningTaskService(IHttpClientProxy<IInterfaceForwardServiceV2> interfaceForwardServiceV2, IOptionsMonitor<InterfaceRelayParamsConfigOptions> options, IHttpClientProxy<IRequestSendMessageService> requestSendMessageService, IWarningDetailsEsRepository warningDetailsEsRepository, IWarningTaskRepository warningTaskRepository)
|
||||
{
|
||||
_interfaceForwardServiceV2 = interfaceForwardServiceV2;
|
||||
_requestSendMessageService = requestSendMessageService;
|
||||
_warningDetailsEsRepository = warningDetailsEsRepository;
|
||||
_warningTaskRepository = warningTaskRepository;
|
||||
_options = options;
|
||||
}
|
||||
|
||||
@ -77,7 +81,7 @@ public class WarningTaskService : IWarningTaskService, ITransientDependency
|
||||
|
||||
#region 接口通调用，轨迹推送请求（一次）
|
||||
|
||||
[AutomaticRetry(Attempts = 0)] //不管失败与否，不进行重试
|
||||
[AutomaticRetry(Attempts = 3, ExceptOn = new[] { typeof(BusinessException) })] //遇到BusinessException就不重试了
|
||||
public async Task CallTrackRetryPushAsync(ReturnWarningRetryPushTaskEntity task, CancellationToken ct)
|
||||
{
|
||||
ct.ThrowIfCancellationRequested();
|
||||
@ -117,13 +121,9 @@ public class WarningTaskService : IWarningTaskService, ITransientDependency
|
||||
var msg = $"接口通调用失败：BatchForwardThenPushToQueueAsync" +
|
||||
$"RequestId: {r.RequestId};SubSystemCode:{task.SubSystemCode};" +
|
||||
$" serviceProviderCode:{task.ServiceProviderCode} Msg: {r.Msg}";
|
||||
await _warningDetailsEsRepository.InsertReturnMessageRecordAsync(new ReturnMessageRecordIndex
|
||||
{
|
||||
WarningDetailsIds = task.WarningDetailIds,
|
||||
IsSuccess = false,
|
||||
Message = msg,
|
||||
CreateTime = DateTime.Now
|
||||
});
|
||||
|
||||
// 处理调用接口通失败，直接处于完结状态
|
||||
await HandlerCallBusinessException(task, msg);
|
||||
throw new BusinessException(message: msg);
|
||||
}
|
||||
}
|
||||
@ -141,5 +141,60 @@ public class WarningTaskService : IWarningTaskService, ITransientDependency
|
||||
}
|
||||
}
|
||||
|
||||
private async Task HandlerCallBusinessException(ReturnWarningRetryPushTaskEntity task, string msg)
|
||||
{
|
||||
var detailIds = task.WarningDetailIds.Split(',').ToHashSet();
|
||||
// 1.查询处理（重试中/人工介入）的预警明细
|
||||
var warningDetails = await _warningDetailsEsRepository.GetReturnWarningDetailsAsync(new GetReturnWarningDetailsVO
|
||||
{
|
||||
DetailsIds = detailIds
|
||||
});
|
||||
|
||||
// 由于后续都不会在成功了，所以直接根据是否命中kpi考核
|
||||
var addNotifyTasks = new List<WarningNotifyTaskEntity>();
|
||||
foreach (var detail in warningDetails)
|
||||
{
|
||||
if (detail.Status == ReturnWarningStatus.已关闭.GetHashCode())
|
||||
continue;
|
||||
|
||||
// 存在kpi考核
|
||||
if (detail.IsExistsKpi)
|
||||
{
|
||||
detail.Status = ReturnWarningStatus.人工介入.GetHashCode();
|
||||
addNotifyTasks.Add(BuildWarningNotifyTask(detail));
|
||||
continue;
|
||||
}
|
||||
|
||||
// 不存在kpi考核直接关闭
|
||||
detail.Status = ReturnWarningStatus.已关闭.GetHashCode();
|
||||
}
|
||||
|
||||
// 更新预警明细，添加回传报文记录
|
||||
await _warningDetailsEsRepository.BulkUpdateReturnWarningDetailsAsync(warningDetails);
|
||||
await _warningDetailsEsRepository.InsertReturnMessageRecordAsync(new ReturnMessageRecordIndex
|
||||
{
|
||||
WarningDetailsIds = task.WarningDetailIds,
|
||||
IsSuccess = false,
|
||||
Message = msg,
|
||||
CreateTime = DateTime.Now
|
||||
});
|
||||
|
||||
// 添加通知任务
|
||||
if (addNotifyTasks.Count > 0)
|
||||
await _warningTaskRepository.BulkInsertWarningNotifyTaskAsync(addNotifyTasks);
|
||||
}
|
||||
|
||||
private static WarningNotifyTaskEntity BuildWarningNotifyTask(ReturnWarningDetailsIndex detail)
|
||||
{
|
||||
return new WarningNotifyTaskEntity(
|
||||
detail.OrderId,
|
||||
detail.WarningId!.Value,
|
||||
detail.ReturnWarningRuleItemId!.Value,
|
||||
detail.Id,
|
||||
WarningRuleType.回传时效预警.GetHashCode(),
|
||||
detail.NotifyUsers!
|
||||
);
|
||||
}
|
||||
|
||||
#endregion
|
||||
}
|
||||
|
||||
@ -373,7 +373,8 @@ public class TransportWarningHandler : ITransportWarningHandler, ITransientDepen
|
||||
// 明细上的规则校验为完成，修改为完成态（使用终止轨迹状态最早的轨迹填充截止轨迹信息）
|
||||
if (ruleItem.CanCompleted(trackStatusIds))
|
||||
{
|
||||
var earliestTrack = trackStatusIdEarliestTrackDict[detail.EndTrackStatusId];
|
||||
var startTrackStatusEarliestTrack = trackStatusIdEarliestTrackDict[detail.StartTrackStatusId];
|
||||
var endTrackStatusEarliestTrack = trackStatusIdEarliestTrackDict[detail.EndTrackStatusId];
|
||||
|
||||
// 修改为完成态
|
||||
if (detail.CalcStatus() == TransportWarningStatus.已超时.GetHashCode())
|
||||
@ -382,14 +383,25 @@ public class TransportWarningHandler : ITransportWarningHandler, ITransientDepen
|
||||
detail.Status = TransportWarningStatus.已完成.GetHashCode();
|
||||
|
||||
// 填充终止状态轨迹信息
|
||||
detail.EndTrackContentId = earliestTrack.TrackContentId;
|
||||
detail.EndTrackContent = earliestTrack.TrackContent;
|
||||
detail.EndTrackTime = earliestTrack.TrackTime;
|
||||
detail.EndTrackTimeZone = earliestTrack.TrackTimeZone;
|
||||
detail.EndTrackContentId = endTrackStatusEarliestTrack.TrackContentId;
|
||||
detail.EndTrackContent = endTrackStatusEarliestTrack.TrackContent;
|
||||
detail.EndTrackTime = endTrackStatusEarliestTrack.TrackTime;
|
||||
detail.EndTrackTimeZone = endTrackStatusEarliestTrack.TrackTimeZone;
|
||||
|
||||
// 维护累计运输时间（由于变更为终止态，此时为最后一次变更）
|
||||
detail.CumulativeTimeSpan = detail.CalcCumulativeTimeSpan();
|
||||
|
||||
// 默认判断超时完成是根据当前时间判断的，但可能截止轨迹发生时间在时效要求范围内，故需要二次校验
|
||||
if (detail.Status == TransportWarningStatus.超时完成.GetHashCode())
|
||||
{
|
||||
var starTrackTime = startTrackStatusEarliestTrack.GetUtc0TrackTime();
|
||||
var endTrackTime = endTrackStatusEarliestTrack.GetUtc0TrackTime();
|
||||
// 如果添加的截止轨迹最早轨迹发生时间，小于起始轨迹状态最早轨迹发生时间+时效则认为是正常完成
|
||||
// tips: 可能出现的原因，轨迹已经产生但没有及时加到系统中去
|
||||
if (endTrackTime < starTrackTime.AddHours(detail.TransportTimeSpan))
|
||||
detail.Status = TransportWarningStatus.已完成.GetHashCode();
|
||||
}
|
||||
|
||||
updateDetails.Add(detail);
|
||||
closeNotifyDetailIds.Add(detail.Id);
|
||||
continue;
|
||||
|
||||
@ -197,10 +197,14 @@ public class TrackDetailsIndex : IdVersionIndex
|
||||
/// </summary>
|
||||
public string? Extra { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// 如果时区为空，则兜底为utc+8
|
||||
/// </summary>
|
||||
/// <returns></returns>
|
||||
public DateTime GetUtc0TrackTime()
|
||||
{
|
||||
return Math.Abs(TrackTimeZone - GlobalConst.NoneTrackTimeZone) < 0.01f
|
||||
? TrackTime
|
||||
? TrackTime.AddHours(-8.0f)
|
||||
: TrackTime.AddHours(-TrackTimeZone);
|
||||
}
|
||||
}
|
||||
|
||||
@ -79,10 +79,10 @@ public sealed class TransportWarningDetailsIndex : IdBaseIndex
|
||||
{
|
||||
// 已完成/超时完成，截止轨迹时间 - 起始轨迹时间
|
||||
if (Status is (int) TransportWarningStatus.已完成 or (int) TransportWarningStatus.超时完成)
|
||||
return (long)EndTrackTime!.Value.Subtract(StartTrackTime).TotalMilliseconds;
|
||||
return (long)GetUtc0EndTrackTime()!.Value.Subtract(GetUtc0StartTrackTime()).TotalMilliseconds;
|
||||
|
||||
// 其他状态，当前时间 - 起始轨迹时间
|
||||
return (long)DateTime.Now.Subtract(StartTrackTime).TotalMilliseconds;
|
||||
return (long)DateTime.UtcNow.Subtract(GetUtc0StartTrackTime()).TotalMilliseconds;
|
||||
}
|
||||
|
||||
// tips:计算并设置状态，后续维护到vo当中
|
||||
|
||||
@ -34,7 +34,7 @@ public class WarningDetailsEsRepository : IWarningDetailsEsRepository, ITransien
|
||||
musts.AddIf(x => x.OrderType > 0, c => c.Term(f => f.OrderType, input.OrderType));
|
||||
musts.AddIf(x => x.Status > 0, c => c.Term(f => f.Status, input.Status));
|
||||
musts.AddIf(x => x.SubSystemId > 0, c => c.Term(f => f.SubSystemId, input.SubSystemId));
|
||||
musts.AddIf(x => !string.IsNullOrWhiteSpace(x.WarningName), c => c.Term(f => f.WarningName.Suffix("keyword"), input.WarningName));
|
||||
musts.AddIf(x => !string.IsNullOrWhiteSpace(x.WarningName), c => c.Match(f => f.Field(d => d.WarningName).Query(input.WarningName)));
|
||||
musts.AddIf(x => x.CreateTimeStart.HasValue, c => c.DateRange(r => r.Field(f => f.CreateTime).GreaterThanOrEquals(input.CreateTimeStart)));
|
||||
musts.AddIf(x => x.CreateTimeEnd.HasValue, c => c.DateRange(r => r.Field(f => f.CreateTime).LessThanOrEquals(input.CreateTimeEnd)));
|
||||
|
||||
|
||||
Loading…
Reference in New Issue
Block a user