ideship/track
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix:修复回传成功也增加回传报文

Browse Source 
fix:补充调用接口通失败也增加回传报文
add-interface:增加回传推送接口
This commit is contained in:
hardstylewyl 2025-03-25 11:03:34 +08:00
parent 3b799baccf
commit e61d225140
5 changed files with 129 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
src/SJZY.Track.Application.Contract/Dtos/Warning/Return/ReturnPushInput.cs Normal file
View File

@ -0,0 +1,12 @@
using System.ComponentModel.DataAnnotations;
namespace SJZY.Track.Application.Contract.Dtos.Warning.Return;
public sealed class ReturnPushInput
{
/// <summary>
/// 预警明细ids 最多100个
/// </summary>
[MaxLength(100)]
public List<string> WarningDetailIds { get; set; } = new();
}

21
src/SJZY.Track.Application/Hangfire/WarningTaskService.cs
View File

@ -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

2
src/SJZY.Track.Application/Services/Warning/Handlers/IReturnWarningHandler.cs
View File

@ -12,7 +12,7 @@ public interface IReturnWarningHandler
Task TrackUpdatedOrDeletedHandleAsync(List<IWarningSignal> updateOrDeleteSignals);
// 收到轨迹推送记录（接口通回调）（人工介入/重试 阶段的回传）
Task ReceiveTrackPushRecordHandleAsync(bool isSuccess, string requestParams, List<TrackPushRecordIndex> logs, List<TrackPushInput> pushInputs, List<TrackUpdateInput> updateInputs);
Task ReceiveTrackPushRecordHandleAsync(bool isSuccess, string requestParams, List<TrackPushRecordIndex> records, List<TrackPushInput> pushInputs, List<TrackUpdateInput> updateInputs);
// 收到轨迹推送记录（接口通回调）（人工介入/重试 阶段的回传）
// Task ReceiveTrackPushRecordHandleAsync(TrackPushRecordIndex record);

7
src/SJZY.Track.Application/Services/Warning/Handlers/ReturnWarningHandler.cs
View File

@ -40,7 +40,7 @@ public sealed class ReturnWarningHandler : IReturnWarningHandler, ITransientDepe
public async Task ReceiveTrackPushRecordHandleAsync(
bool isSuccess,
string requestParams,
List<TrackPushRecordIndex> logs,
List<TrackPushRecordIndex> records,
List<TrackPushInput> pushInputs,
List<TrackUpdateInput> 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(

100
src/SJZY.Track.Application/Services/Warning/ReturnWarningService.cs
View File

@ -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<IUserBaseInfoService> _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<IUserBaseInfoService> requestCurrentUserService)
{
_warningRuleRepository = warningRuleRepository;
_warningDetailsEsRepository = warningDetailsEsRepository;
@ -37,6 +43,8 @@ public class ReturnWarningService : ApplicationService
_orderRepository = orderRepository;
_vExternalSubjectQueries = vExternalSubjectQueries;
_vSystemQueries = vSystemQueries;
_capPublisher = capPublisher;
_requestCurrentUserService = requestCurrentUserService;
}
/// <summary>
@ -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);
}
/// <summary>
/// 发起回传推送
/// </summary>
/// <returns></returns>
[HttpPost("ReturnPush")]
public async Task<UnifyResultDto> 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<int, KeyValuePair<string, long>>
{
[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<HashSet<int>> sets)
{
var combinedSet = new HashSet<int>();