diff --git a/src/SJZY.Track.Application.Contract/Dtos/Warning/Return/ReturnPushInput.cs b/src/SJZY.Track.Application.Contract/Dtos/Warning/Return/ReturnPushInput.cs new file mode 100644 index 00000000..d6d637b7 --- /dev/null +++ b/src/SJZY.Track.Application.Contract/Dtos/Warning/Return/ReturnPushInput.cs @@ -0,0 +1,12 @@ +using System.ComponentModel.DataAnnotations; + +namespace SJZY.Track.Application.Contract.Dtos.Warning.Return; + +public sealed class ReturnPushInput +{ + /// + /// 预警明细ids 最多100个 + /// + [MaxLength(100)] + public List WarningDetailIds { get; set; } = new(); +} diff --git a/src/SJZY.Track.Application/Hangfire/WarningTaskService.cs b/src/SJZY.Track.Application/Hangfire/WarningTaskService.cs index bfd1db69..928dce84 100644 --- a/src/SJZY.Track.Application/Hangfire/WarningTaskService.cs +++ b/src/SJZY.Track.Application/Hangfire/WarningTaskService.cs @@ -1,5 +1,6 @@ using Hangfire; using Microsoft.Extensions.Options; +using SJZY.Expand.ABP.Core.Logger; using SJZY.InterfaceRelay.Application.Contracts; using SJZY.Track.Application.Contract.Dtos.TrackPush; using SJZY.Track.Application.Helpers; @@ -41,6 +42,7 @@ public class WarningTaskService : IWarningTaskService, ITransientDependency } #region 用户中心调用，通知（一次） + [AutomaticRetry(Attempts = 3)] public async Task CallNotifyAsync(NotifyConfig config, CancellationToken ct) { @@ -112,16 +114,23 @@ public class WarningTaskService : IWarningTaskService, ITransientDependency if (r.Code != GlobalConst.SuccessCode) { - throw new BusinessException( - message: $"接口通调用失败：BatchForwardThenPushToQueueAsync" + - $"RequestId: {r.RequestId};SubSystemCode:{task.SubSystemCode};" + - $" serviceProviderCode:{task.ServiceProviderCode} Msg: {r.Msg}" - ); + 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 + }); + throw new BusinessException(message: msg); } } } - catch (Exception) + catch (Exception e) { + LogHelper.Error("[CallTrackRetryPushAsync]" + e, true); throw; } finally diff --git a/src/SJZY.Track.Application/Services/Warning/Handlers/IReturnWarningHandler.cs b/src/SJZY.Track.Application/Services/Warning/Handlers/IReturnWarningHandler.cs index b02eb6b0..bd5621fd 100644 --- a/src/SJZY.Track.Application/Services/Warning/Handlers/IReturnWarningHandler.cs +++ b/src/SJZY.Track.Application/Services/Warning/Handlers/IReturnWarningHandler.cs @@ -12,7 +12,7 @@ public interface IReturnWarningHandler Task TrackUpdatedOrDeletedHandleAsync(List updateOrDeleteSignals); // 收到轨迹推送记录（接口通回调）（人工介入/重试 阶段的回传） - Task ReceiveTrackPushRecordHandleAsync(bool isSuccess, string requestParams, List logs, List pushInputs, List updateInputs); + Task ReceiveTrackPushRecordHandleAsync(bool isSuccess, string requestParams, List records, List pushInputs, List updateInputs); // 收到轨迹推送记录（接口通回调）（人工介入/重试 阶段的回传） // Task ReceiveTrackPushRecordHandleAsync(TrackPushRecordIndex record); diff --git a/src/SJZY.Track.Application/Services/Warning/Handlers/ReturnWarningHandler.cs b/src/SJZY.Track.Application/Services/Warning/Handlers/ReturnWarningHandler.cs index cd669efd..f21895cd 100644 --- a/src/SJZY.Track.Application/Services/Warning/Handlers/ReturnWarningHandler.cs +++ b/src/SJZY.Track.Application/Services/Warning/Handlers/ReturnWarningHandler.cs @@ -40,7 +40,7 @@ public sealed class ReturnWarningHandler : IReturnWarningHandler, ITransientDepe public async Task ReceiveTrackPushRecordHandleAsync( bool isSuccess, string requestParams, - List logs, + List records, List pushInputs, List updateInputs ) @@ -48,14 +48,14 @@ public sealed class ReturnWarningHandler : IReturnWarningHandler, ITransientDepe if (!isSuccess) { // 失败处理 - await HandleFailedAsync(requestParams, logs, pushInputs, updateInputs); + await HandleFailedAsync(requestParams, records, pushInputs, updateInputs); return; } // 1.获取预警明细（重试中/人工介入） var warningDetails = await _warningDetailsEsRepository.GetReturnWarningDetailsAsync(new GetReturnWarningDetailsVO { - TrackEsIds = logs.Select(x => x.TrackEsId).ToHashSet() + TrackEsIds = records.Select(x => x.TrackEsId).ToHashSet() }); if (warningDetails.Count == 0) return; @@ -76,6 +76,7 @@ public sealed class ReturnWarningHandler : IReturnWarningHandler, ITransientDepe detail.CalcReturnSuccessTimeSpan(trackReturnTime); //计算回传成功各项计算属性 }); await _warningDetailsEsRepository.BulkUpdateReturnWarningDetailsAsync(warningDetails); + await _warningDetailsEsRepository.InsertReturnMessageRecordAsync(BuildReturnMessageRecord(warningDetails, records.First())); } private async Task HandleFailedAsync( diff --git a/src/SJZY.Track.Application/Services/Warning/ReturnWarningService.cs b/src/SJZY.Track.Application/Services/Warning/ReturnWarningService.cs index 46f8d84e..1437fe24 100644 --- a/src/SJZY.Track.Application/Services/Warning/ReturnWarningService.cs +++ b/src/SJZY.Track.Application/Services/Warning/ReturnWarningService.cs @@ -1,7 +1,8 @@ +using DotNetCore.CAP; using SJZY.Expand.ABP.Core.PublicMethords; using SJZY.Track.Application.Contract.Dtos.Warning; using SJZY.Track.Application.Contract.Dtos.Warning.Return; -using SJZY.Track.Application.Helpers; +using SJZY.Track.Application.Contract.Events; using SJZY.Track.Application.Services.Warning.Handlers; using SJZY.Track.Application.Services.Warning.Model; using SJZY.Track.Domain.Aggregates.OrderAggregate; @@ -10,6 +11,8 @@ using SJZY.Track.Domain.Aggregates.WarningAggregate; using SJZY.Track.Domain.Dtos.Warning; using SJZY.Track.Domain.Enums; using SJZY.Track.Domain.Queries; +using SJZY.UserCenter.HttpApi.Services.User; +using Volo.Abp.Http.Client.DynamicProxying; namespace SJZY.Track.Application.Services.Warning; @@ -27,8 +30,11 @@ public class ReturnWarningService : ApplicationService private readonly IWarningRuleRepository _warningRuleRepository; private readonly IWarningDetailsEsRepository _warningDetailsEsRepository; private readonly ITrackStatusRepository _trackStatusRepository; + private readonly ICapPublisher _capPublisher; + private readonly IHttpClientProxy _requestCurrentUserService; - public ReturnWarningService(IWarningRuleRepository warningRuleRepository, IWarningDetailsEsRepository warningDetailsEsRepository, IReturnWarningHandler returnWarningHandler, ITrackStatusRepository trackStatusRepository, IOrderRepository orderRepository, IVExternalSubjectQueries vExternalSubjectQueries, IVSystemQueries vSystemQueries) + + public ReturnWarningService(IWarningRuleRepository warningRuleRepository, IWarningDetailsEsRepository warningDetailsEsRepository, IReturnWarningHandler returnWarningHandler, ITrackStatusRepository trackStatusRepository, IOrderRepository orderRepository, IVExternalSubjectQueries vExternalSubjectQueries, IVSystemQueries vSystemQueries, ICapPublisher capPublisher, IHttpClientProxy requestCurrentUserService) { _warningRuleRepository = warningRuleRepository; _warningDetailsEsRepository = warningDetailsEsRepository; @@ -37,6 +43,8 @@ public class ReturnWarningService : ApplicationService _orderRepository = orderRepository; _vExternalSubjectQueries = vExternalSubjectQueries; _vSystemQueries = vSystemQueries; + _capPublisher = capPublisher; + _requestCurrentUserService = requestCurrentUserService; } /// @@ -240,7 +248,7 @@ public class ReturnWarningService : ApplicationService // 要求规则项目中轨迹状态集合不允许存在交集 if (CheckForIntersection(bos.Select(x => x.TrackStatusIds).ToList())) { - throw new BusinessException("运输预警规则项目间，轨迹状态不允许存在交集！"); + throw new BusinessException(message: "回传预警规则项目间，轨迹状态不允许存在交集！"); } await _returnWarningHandler.HandleEditAsync(rule, bos); @@ -248,6 +256,92 @@ public class ReturnWarningService : ApplicationService return new UnifyResultDto(true); } + /// + /// 发起回传推送 + /// + /// + [HttpPost("ReturnPush")] + public async Task ReturnPush(ReturnPushInput input) + { + if (input.WarningDetailIds.Count == 0) + { + throw new BusinessException(message: "未选择任何回传预警详情，无法进行推送！"); + } + + // 1.查出预警明细 + var ids = input.WarningDetailIds.ToHashSet(); + var warningDetails = await _warningDetailsEsRepository.GetReturnWarningDetailsAsync(new GetReturnWarningDetailsVO() + { + DetailsIds = ids + }); + + if (warningDetails.Count == 0) + { + throw new BusinessException(message: "一个能推送的也没有"); + } + + // 2.选出可以进行推送的预警明细 + var filtered = warningDetails + .Where(x => x.Status == (int) ReturnWarningStatus.人工介入) + .ToList(); + if (filtered.Count == 0) + { + throw new BusinessException(message: "一个能推送的也没有"); + } + + // 获取当前用户信息 + var userInfo = await _requestCurrentUserService.Service.GetCurrentUserBasicInfoAsync(); + if (userInfo.Code != GlobalConst.SuccessCode || userInfo.Data is null) + { + throw new BusinessException(message: $"调用用户中心，获取用户信息失败:{userInfo.Msg}"); + } + + // 子系统id换code + var subSystemIds = filtered.Select(x => x.SubSystemId).ToHashSet(); + var subSystemCodeDict = (await _vSystemQueries + .GetListAsync(x => subSystemIds.Contains(x.Id), x => new + { + x.Id, + x.Key + })).ToDictionary(x => x.Id, x => x.Key); + + // 3.进行推送 + foreach (var detail in filtered) + { + var trackPushEvent = new TrackPushEvent + { + OptionType = OptionType.新增, + SubSystemId = detail.SubSystemId, + SubSystemCode = subSystemCodeDict.GetValueOrDefault(detail.SubSystemId, $"系统代码{detail.SubSystemId}不存在"), + CustomId = detail.CustomId, + OrderIdEsIdVersionDict = new Dictionary> + { + [detail.OrderId] = KeyValuePair.Create(detail.TrackEsId, detail.TrackVersion) + }, + // OrderIds = new[] { order.Id }, + TrackCode = detail.TrackCode, + OperationMan = userInfo.Data.UserName, + SystemPushInfo = new SystemPushStruct + { + TrackStatusId = detail.TrackStatusId, + TrackStatus = detail.TrackStatus, + TrackContentId = detail.TrackContentId, + TrackContent = detail.TrackContent, + TrackPlace = detail.TrackPlace, + TrackTime = detail.TrackTime, + TrackTimeZone = detail.TrackTimeZone, + CompareFields = detail.CompareFields, + Extra = detail.Extra + } + }; + + await _capPublisher.PublishAsync(GlobalConst.TrackManualPushSubscribe, trackPushEvent); + } + + return new UnifyResultDto(true, $"即将进行{filtered.Count}个回传推送！"); + } + + // 检查集合中是否存在交集 private static bool CheckForIntersection(List> sets) { var combinedSet = new HashSet();