diff --git a/src/SJZY.Track.Application/CAP/EventHandlers/TrackPushEventHandler.cs b/src/SJZY.Track.Application/CAP/EventHandlers/TrackPushEventHandler.cs index 07f0380d..fc9c5e03 100644 --- a/src/SJZY.Track.Application/CAP/EventHandlers/TrackPushEventHandler.cs +++ b/src/SJZY.Track.Application/CAP/EventHandlers/TrackPushEventHandler.cs @@ -1,3 +1,4 @@ +using System.Collections.Immutable; using System.Text.Json; using DotNetCore.CAP; using Microsoft.Extensions.Options; @@ -11,6 +12,7 @@ using SJZY.Track.Application.Helpers; using SJZY.Track.Application.Options; using SJZY.Track.Application.Track2InterfaceRelay; using SJZY.Track.Domain.Aggregates.OrderAggregate; +using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate; using SJZY.Track.Domain.Aggregates.TrackPushAggregate; using SJZY.Track.Domain.Enums; using Volo.Abp.DependencyInjection; @@ -25,6 +27,7 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe private readonly IOrderRepository _orderRepository; private readonly ITrackPushCacheService _trackPushCacheService; private readonly ITrackPushRecordEsRepository _trackPushRecordEsRepository; + private readonly ITrackDetailsEsRepository _trackDetailsEsRepository; private readonly IBroadcastingService _broadcastingService; private List _orders = null!; private List _pushConfigs = null!; @@ -34,7 +37,8 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe ITrackPushCacheService trackPushCacheService, ITrackPushRecordEsRepository trackPushRecordEsRepository, IOptionsMonitor options, - IBroadcastingService broadcastingService) + IBroadcastingService broadcastingService, + ITrackDetailsEsRepository trackDetailsEsRepository) { _httpClientProxy = httpClientProxy; _orderRepository = orderRepository; @@ -42,6 +46,7 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe _trackPushRecordEsRepository = trackPushRecordEsRepository; _options = options; _broadcastingService = broadcastingService; + _trackDetailsEsRepository = trackDetailsEsRepository; } [CapSubscribe(GlobalConst.TrackPushSubscribe, Group = GlobalConst.TrackPushSubscribe)] @@ -96,130 +101,21 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe private async Task DoPushAsync(TrackPushEvent @event) { var items = new List(); - TrackPushContentConfigDto? config; - var isSystemPush = @event.OperationMan == null; - if (isSystemPush && !@event.ForcedPush) + if (@event.OperationMan == null) { - config = _pushConfigs.FirstOrDefault(x => x.TrackCode == @event.TrackCode); - foreach (var order in _orders) + if (@event.OptionType == OptionType.新增) { - if (config == null) - { - TrackPushFailedPublicTopic($"没有在推送映射配置中找到关于轨迹代码[{@event.TrackCode}]的配置", @event, order); - return; - } - - // 系统不推异常单，只能人工操作 - if (order.QuestionType == (int) TrackProblemType.Error) - { - TrackPushFailedPublicTopic("系统不推异常单，只能人工操作", @event, order); - continue; - } - - // 推送新增，需要根据上一次的推送 推断 本次推送 是否可行，为了约束顺序 - var (resultConfig, nextCode) = GetValidPushConfigAsync(_pushConfigs, order.LatestPushTrackCode, @event.TrackCode); - if (resultConfig == null) - { - TrackPushFailedPublicTopic($"推送顺序异常，下一个要推送的应为[{nextCode}]", @event, order); - continue; - } - - config = resultConfig; - - // 是否自动推送 - if (!config.IsAutoPush) - { - TrackPushFailedPublicTopic($"没有设置为自动推送，如果要推送请手动推送", @event, order); - continue; - } - - // 配置的类型如何和订单类型不一致则不允许推送 - if (config.OrderType != order.OrderType) - { - TrackPushFailedPublicTopic($"订单类型{order.OrderType}和配置中的订单类型[{config.OrderType}]不一致，不能推送", @event, order); - continue; - } - - var log = new TrackPushRecordIndex - { - OptionType = @event.OptionType.ToString(), - OrderId = order.Id, - OrderNo = order.OrderNo, - TrackStatus = @event.SystemPushInfo.TrackStatus, - TrackCode = @event.TrackCode, - TrackContent = @event.SystemPushInfo.TrackContent, - TrackPlace = @event.SystemPushInfo.TrackPlace, - TrackTime = @event.SystemPushInfo.TrackTime, - TrackTimeZone = @event.SystemPushInfo.TrackTimeZone, - CreateUser = GlobalConst.SystemUser, - SubSystemCode = @event.SubSystemCode, - IsToMq = @event.IsToMq, - Extra = @event.SystemPushInfo.Extra - }; - var item = await BuiltPushItem(config, log, @event.SystemPushInfo.CompareFields); - item.Log = JsonConvert.SerializeObject(log); - items.Add(item); + await BuiltSystemPushItemsAsync(@event, items); + } + else if (@event.OptionType == OptionType.修改) + { + await BuiltManualPushItemsAsync(@event, items); } } - // 手工或强制推送，不会校验已推和顺序问题，因为可能是更新 else - { - config = _pushConfigs.FirstOrDefault(x => x.TrackCode == @event.TrackCode); - foreach (var order in _orders) - { - if (config == null) - { - TrackPushFailedPublicTopic($"没有在推送映射配置中找到关于轨迹代码[{@event.TrackCode}]的配置", @event, order); - return; - } - - if (config is { IsAllPush: true }) - { - // 非强制推送需要 推断 本次推送 是否可行 - if (!@event.ForcedPush) - { - var (resultConfig, nextCode) = GetValidPushConfigAsync(_pushConfigs, order.LatestPushTrackCode, @event.TrackCode); - if (resultConfig == null) - { - TrackPushFailedPublicTopic($"推送顺序异常，下一个要推送的应为[{nextCode}]", @event, order); - continue; - } - - config = resultConfig; - } - - // tips:强制推送直接使用配置 - } - - - //配置的类型如何和订单类型不一致则不允许推送 - if (config.OrderType != order.OrderType) - { - TrackPushFailedPublicTopic($"订单类型{order.OrderType}和配置中的订单类型[{config.OrderType}]不一致，不能推送", @event, order); - continue; - } - - var log = new TrackPushRecordIndex - { - OptionType = @event.OptionType.ToString(), - OrderId = order.Id, - OrderNo = order.OrderNo, - TrackStatus = @event.SystemPushInfo.TrackStatus, - TrackCode = @event.TrackCode, - TrackContent = @event.SystemPushInfo.TrackContent, - TrackPlace = @event.SystemPushInfo.TrackPlace, - TrackTime = @event.SystemPushInfo.TrackTime, - TrackTimeZone = @event.SystemPushInfo.TrackTimeZone, - CreateUser = @event.OperationMan, - SubSystemCode = @event.SubSystemCode, - IsToMq = @event.IsToMq, - Extra = @event.SystemPushInfo.Extra - }; - var item = await BuiltPushItem(config, log, @event.SystemPushInfo.CompareFields); - item.Log = JsonConvert.SerializeObject(log); - items.Add(item); - } + // 手工推送，不会校验已推和顺序问题，因为可能是更新 + await BuiltManualPushItemsAsync(@event, items); } if (items.Count == 0) return; @@ -236,12 +132,133 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe if (r.Code != GlobalConst.SuccessCode) { - TrackPushFailedPublicTopic($"轨迹中台调用接口通失败,RequestId: {r.RequestId}; Msg: {r.Msg}", @event, r.RequestId, items.Cast().ToList()); - throw new BusinessException(message: $"requestId: {r.RequestId}; serviceProviderCode:{serviceProviderCode} Msg: {r.Msg}"); + TrackPushFailedPublicTopic($"轨迹中台调用接口通失败,RequestId: {r.RequestId}; Msg: {r.Msg}", @event, r.RequestId, + items.Cast().ToList()); + throw new BusinessException( + message: $"requestId: {r.RequestId}; serviceProviderCode:{serviceProviderCode} Msg: {r.Msg}"); } } } + private async Task BuiltManualPushItemsAsync(TrackPushEvent @event, List items) + { + // tips: 手工强制推送直接使用配置 + var config = _pushConfigs.FirstOrDefault(x => x.TrackCode == @event.TrackCode); + foreach (var order in _orders) + { + if (config is { IsAllPush: true }) + { + // 非强制推送需要 推断 本次推送 是否可行 + if (!@event.ForcedPush) + { + config = GetValidPushConfigAsync(_pushConfigs, order.LatestPushTrackCode, @event); + } + } + + if (config == null) + { + TrackPushFailedPublicTopic("推送顺序异常，找不到有效推送配置", @event, order); + return; + } + + + //配置的类型如何和订单类型不一致则不允许推送 + if (config.OrderType != order.OrderType) + { + TrackPushFailedPublicTopic($"订单类型{order.OrderType}和配置中的订单类型[{config.OrderType}]不一致，不能推送", @event, + order); + continue; + } + + var log = new TrackPushRecordIndex + { + OptionType = @event.OptionType.ToString(), + OrderId = order.Id, + OrderNo = order.OrderNo, + TrackStatus = @event.SystemPushInfo.TrackStatus, + TrackCode = @event.TrackCode, + TrackContent = @event.SystemPushInfo.TrackContent, + TrackPlace = @event.SystemPushInfo.TrackPlace, + TrackTime = @event.SystemPushInfo.TrackTime, + TrackTimeZone = @event.SystemPushInfo.TrackTimeZone, + CreateUser = @event.OperationMan, + SubSystemCode = @event.SubSystemCode, + IsToMq = @event.IsToMq, + Extra = @event.SystemPushInfo.Extra + }; + var item = await BuiltPushItem(config, log, @event.SystemPushInfo.CompareFields); + item.Log = JsonConvert.SerializeObject(log); + items.Add(item); + } + } + + private async Task BuiltSystemPushItemsAsync(TrackPushEvent @event, List items) + { + var config = _pushConfigs.FirstOrDefault(x => x.TrackCode == @event.TrackCode); + foreach (var order in _orders) + { + if (config == null) + { + TrackPushFailedPublicTopic($"没有在推送映射配置中找到关于轨迹代码[{@event.TrackCode}]的配置", @event, order); + return; + } + + // 系统不推异常单，只能人工操作 + if (order.QuestionType == (int) TrackProblemType.Error) + { + TrackPushFailedPublicTopic("系统不推异常单，只能人工操作", @event, order); + continue; + } + + // 推送新增，需要根据上一次的推送 推断 本次推送 是否可行，为了约束顺序 + // 如果当前推送code与订单上的最后的code不一致，则考虑推送映射顺序 + if (!string.Equals(order.LatestPushTrackCode, @event.TrackCode)) + { + config = GetValidPushConfigAsync(_pushConfigs, order.LatestPushTrackCode, @event); + if (config == null) + { + TrackPushFailedPublicTopic("推送顺序异常，找不到有效推送配置", @event, order); + continue; + } + } + + // 是否自动推送 + if (!config.IsAutoPush) + { + TrackPushFailedPublicTopic("没有设置为自动推送，如果要推送请手动推送", @event, order); + continue; + } + + // 配置的类型如何和订单类型不一致则不允许推送 + if (config.OrderType != order.OrderType) + { + TrackPushFailedPublicTopic($"订单类型{order.OrderType}和配置中的订单类型[{config.OrderType}]不一致，不能推送", @event, + order); + continue; + } + + var log = new TrackPushRecordIndex + { + OptionType = @event.OptionType.ToString(), + OrderId = order.Id, + OrderNo = order.OrderNo, + TrackStatus = @event.SystemPushInfo.TrackStatus, + TrackCode = @event.TrackCode, + TrackContent = @event.SystemPushInfo.TrackContent, + TrackPlace = @event.SystemPushInfo.TrackPlace, + TrackTime = @event.SystemPushInfo.TrackTime, + TrackTimeZone = @event.SystemPushInfo.TrackTimeZone, + CreateUser = GlobalConst.SystemUser, + SubSystemCode = @event.SubSystemCode, + IsToMq = @event.IsToMq, + Extra = @event.SystemPushInfo.Extra + }; + var item = await BuiltPushItem(config, log, @event.SystemPushInfo.CompareFields); + item.Log = JsonConvert.SerializeObject(log); + items.Add(item); + } + } + private async Task DoUpdateAsync(TrackPushEvent @event) { var config = _pushConfigs.FirstOrDefault(x => x.TrackCode == @event.TrackCode); @@ -253,7 +270,8 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe //配置的类型如何和订单类型不一致则不允许推送 if (config.OrderType != order.OrderType) { - TrackPushFailedPublicTopic($"订单类型{order.OrderType}和配置中的订单类型[{config.OrderType}]不一致，不能推送", @event, order); + TrackPushFailedPublicTopic($"订单类型{order.OrderType}和配置中的订单类型[{config.OrderType}]不一致，不能推送", @event, + order); continue; } @@ -292,8 +310,10 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe if (r.Code != GlobalConst.SuccessCode) { - TrackPushFailedPublicTopic($"轨迹中台调用接口通失败,RequestId: {r.RequestId}; Msg: {r.Msg}", @event, r.RequestId, items.Cast().ToList()); - throw new BusinessException(message: $"requestId: {r.RequestId}; serviceProviderCode:{serviceProviderCode} Msg: {r.Msg}"); + TrackPushFailedPublicTopic($"轨迹中台调用接口通失败,RequestId: {r.RequestId}; Msg: {r.Msg}", @event, r.RequestId, + items.Cast().ToList()); + throw new BusinessException( + message: $"requestId: {r.RequestId}; serviceProviderCode:{serviceProviderCode} Msg: {r.Msg}"); } } } @@ -351,104 +371,69 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe if (config.IsAllPush) { - var history = await _trackPushRecordEsRepository.GetLatestTrackPushRecordAsync(log.OrderId); - //找到历史中推送成功的轨迹 - var trackPushRecords = await _trackPushRecordEsRepository.GetHistorySuccessTrackPushRecordAsync(log.OrderId); - //构建标识符，code+比较字段的值(如果提供的比较字段,否则使用轨迹时间) - var pushIdentitySuccess = trackPushRecords.Select(x => - { - //寻找映射配置 - var code = _pushConfigs.First(y => y.TrackCode == x.TrackCode).TrackContentCodeValue; - if (string.IsNullOrWhiteSpace(compareFields)) - return code; - - //尝试从扩展中提取目标值 - var extraValue = TextJsonHelper.ExtractValue(x.Extra, compareFields); - return code + extraValue; - }).ToHashSet(); - + var history = await _trackPushRecordEsRepository.GetLatestTrackPushRecordAsync(log.OrderId, true); if (history != default) { body = TextJsonHelper.Deserialize(history.PushContent); - //todo-test:在历史选出推送成功的历史轨迹 + + //在推送历史中，选出当前轨迹明细一致的轨迹 + var tracks = await _trackDetailsEsRepository.GetListAsync(log.OrderId); + //确定轨迹标识符 + var trackIdentifies = tracks.Select(x => + { + //寻找映射配置(转换后的code) + var code = _pushConfigs.First(y => y.TrackCode == x.TrackCode).TrackContentCodeValue; + var timeStr = DateTimeFormat(x.TrackTime, config.TrackTimeZone ?? log.TrackTimeZone, config); + return string.IsNullOrWhiteSpace(compareFields) + ? $"{code}{timeStr}" + : $"{code}{TextJsonHelper.ExtractValue(x.Extra, compareFields)}"; + } + ).ToImmutableHashSet(); + //在历史中选取在订单当前轨迹明细中的记录 body!.Data = body.Data.Where(x => { var code = ((JsonElement) x["code"]!).GetString(); - if (string.IsNullOrWhiteSpace(compareFields)) - return pushIdentitySuccess.Contains(code); + var timeStr = ((JsonElement) x["date"]!).GetString(); - var isMatch = ((JsonElement) x["extra"]!).TryGetProperty(compareFields, out var jsonElement); - var extraValue = isMatch ? TextJsonHelper.GetString(jsonElement) : null; - return pushIdentitySuccess.Contains(code + extraValue); + if (string.IsNullOrWhiteSpace(compareFields)) + return trackIdentifies.Contains($"{code}{timeStr}"); + + if (((JsonElement) x["extra"]!).TryGetProperty(compareFields, out var comparerValueJsonElement)) + return trackIdentifies.Contains($"{code}{TextJsonHelper.GetString(comparerValueJsonElement)}"); + + return trackIdentifies.Contains($"{code}{null}"); }).ToList(); - //添加当前要推送的轨迹 - body.Data.Add(cur); if (config.IsAllPushWithLatest) { - //历史轨迹中同一个code+比较字段联合作为key最新的一个 - var dict = new Dictionary(); - //todo-test:根据code和比较字段值分组，在每组里选出一个创建时间最新的 - foreach (var x in body.Data) + //同code同比较字段保证唯一 + var old = body!.Data + .Where(x => + { + //反序列化出的JsonElement和字典字符串类型做比较 + var a = ((JsonElement) x["code"]!).GetString(); + var b = cur["code"]! + .ToString(); //tips: 需要要将object类型的字符串转为string类型，否则使用的是引用比较，反序列化出的string一定和本地string引用不同 + return a == b; + }) + .WhereIf(!compareFields.IsNullOrWhiteSpace(), x => + { + if (x["extra"] == null || log.Extra == null) return false; + return TextJsonHelper.EqByJsonElementField(x["extra"]!, + TextJsonHelper.Deserialize(log.Extra), compareFields!); + }) + .FirstOrDefault(); + + if (old != null) { - var code = string.Empty; - var dateStr = string.Empty; - var key = string.Empty; - if (x["code"] is JsonElement codeEl) code = codeEl.GetString()!; - else code = x["code"]!.ToString(); - if (x["date"] is JsonElement dateEl) dateStr = dateEl.GetString()!; - else dateStr = x["date"]!.ToString(); - - if (string.IsNullOrWhiteSpace(compareFields)) - key = code; - else - { - var isMatch = ((JsonElement) x["extra"]!).TryGetProperty(compareFields, out var jsonElement); - var extraValue = isMatch ? TextJsonHelper.GetString(jsonElement) : null; - key = code + extraValue; - } - - //只保留最新的一条数据 - if (dict.TryGetValue(key, out var tuple)) - { - //如果当前值创建时间大于历史值，则替换历史值 - if (DateTime.Parse(dateStr) > DateTime.Parse(tuple.dateStr)) - dict[key] = (dateStr, x); - } - else - dict[key] = (dateStr, x); + body.Data.Remove(old); } - - // //同code同比较字段保证唯一 - // var old = body!.Data - // .Where(x => - // { - // //反序列化出的JsonElement和字典字符串类型做比较 - // var a = ((JsonElement) x["code"]!).GetString(); - // var b = cur["code"]!.ToString(); //tips: 需要要将object类型的字符串转为string类型，否则使用的是引用比较，反序列化出的string一定和本地string引用不同 - // return a == b; - // }).WhereIf(!compareFields.IsNullOrWhiteSpace(), x => - // { - // if (x["extra"] == null || log.Extra == null) return false; - // return TextJsonHelper.EqByJsonElementField(x["extra"]!, TextJsonHelper.Deserialize(log.Extra), compareFields!); - // }) - // .FirstOrDefault(); - // - // if (old != null) - // { - // body.Data.Remove(old); - // } - - //选出每个组最新的轨迹 - body.Data = dict.Values.Select(x => x.dict).Cast>().ToList(); } - // //添加当前要推送的轨迹 - // body.Data.Add(cur); + // 添加当前要推送的轨迹 + body.Data.Add(cur); } - - //todo-test:按推送映射顺序排序 } log.PushContent = TextJsonHelper.Serialize(body)!; @@ -460,24 +445,28 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe /// 根据轨迹代码推断下一个（升序）要推送的内容，如果找不到或者已经是最后一个了 /// 则查询有没有可交换的推送，没有就返回null /// - private static (TrackPushContentConfigDto? config, string nextCode) GetValidPushConfigAsync(List cache, - string? latestPushTrackCode, string? trackCode) + private static TrackPushContentConfigDto? GetValidPushConfigAsync(List cache, + string? latestPushTrackCode, TrackPushEvent @event) { + var trackCode = @event.TrackCode; var idx = cache.FindIndex(x => x.TrackCode == latestPushTrackCode); + if (@event.ForcedPush && trackCode == latestPushTrackCode) + { + //强制推送，允许重复推送当前轨迹 + return cache[idx]; + } + if (idx + 1 >= cache.Count) return default; // 推完了 // find next var matched = cache[idx + 1].TrackCode == trackCode ? cache[idx + 1] - : // find swap - cache.FirstOrDefault(x => x.TrackCode == cache[idx + 1].SwapTrackCode); + : cache.FirstOrDefault(x => x.TrackCode == cache[idx + 1].SwapTrackCode); //没推完，可能不符合推送顺序 - return matched != null && matched.TrackCode == trackCode - ? (matched, matched.TrackCode) - : (default, matched?.TrackCode ?? cache[idx + 1].TrackCode); + return matched != null && matched.TrackCode == trackCode ? matched : default; } /// @@ -501,10 +490,7 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe } //轨迹推送失败(被拦截) - private void TrackPushFailedPublicTopic( - string failedMessage, - TrackPushEvent @event, - OrderEntity order) + private void TrackPushFailedPublicTopic(string failedMessage, TrackPushEvent @event, OrderEntity order) { _broadcastingService.TrackPushFailedPubilcTopic(new TrackPushRecordDto { diff --git a/src/SJZY.Track.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs b/src/SJZY.Track.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs index 79f29ff8..135bb0bd 100644 --- a/src/SJZY.Track.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs +++ b/src/SJZY.Track.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs @@ -144,6 +144,8 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe return false; } + // todo: 在这里，判断存在性 + return true; } diff --git a/src/SJZY.Track.Application/Services/ProvideServices/TrackService.cs b/src/SJZY.Track.Application/Services/ProvideServices/TrackService.cs index db991a76..7e8206f8 100644 --- a/src/SJZY.Track.Application/Services/ProvideServices/TrackService.cs +++ b/src/SJZY.Track.Application/Services/ProvideServices/TrackService.cs @@ -3,6 +3,7 @@ using Microsoft.AspNetCore.Http; using SJZY.Track.Application.Contract.Dtos.ProvideServices; using SJZY.Track.Application.Contract.Events; using SJZY.Track.Application.Contract.Services.ProvideServices; +using SJZY.Track.Application.Helpers; using SJZY.Track.Domain.Aggregates.OrderAggregate; using SJZY.Track.Domain.Aggregates.TrackContentAggregate; using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate; @@ -55,6 +56,7 @@ public partial class TrackService : ApplicationService, ITrackService [HttpPost("Track/Upsert")] public async Task UpsertTrackAsync(UpsertTrackInput input) { + // todo: 代码太长了 var order = await _orderRepository.GetFirstAsync(x => x.SubSystemId == input.SubSystemId && x.OrderType == input.OrderType && x.OrderNo == input.OrderNo, x => new @@ -119,6 +121,15 @@ public partial class TrackService : ApplicationService, ITrackService .Where(x => x.TrackTime == input.TrackTime) .ToList(); + //根据比较字段选出所需的轨迹 + if (!string.IsNullOrWhiteSpace(input.CompareFields)) + { + var a = input.TrackCode + TextJsonHelper.ExtractValue(input.Extra, input.CompareFields); + tracks = tracks.Where(x => string.Equals(a, + x.TrackCode + TextJsonHelper.ExtractValue(x.Extra, input.CompareFields!))).ToList(); + } + + // todo：es落盘延迟是不确定的 //当为更新操作时发现更新的数目为0进行重试操作 if (!input.IsAdd && tracks.Count == 0) { @@ -133,6 +144,7 @@ public partial class TrackService : ApplicationService, ITrackService .ToList(); if (tracks.Count > 0) success = true; + // todo: 最大重试次数应该是可配置的，当出现的时候应该有告警，而不是直接给调用方，指数退避 2^N await Task.Delay(300); } finally diff --git a/src/SJZY.Track.Domain/Aggregates/OrderAggregate/IOrderOperationRecordEsRepository.cs b/src/SJZY.Track.Domain/Aggregates/OrderAggregate/IOrderOperationRecordEsRepository.cs index eb97ddae..18485b5a 100644 --- a/src/SJZY.Track.Domain/Aggregates/OrderAggregate/IOrderOperationRecordEsRepository.cs +++ b/src/SJZY.Track.Domain/Aggregates/OrderAggregate/IOrderOperationRecordEsRepository.cs @@ -16,6 +16,14 @@ public interface IOrderOperationRecordEsRepository /// Task AddManyAsync(List indexes); + /// + /// 根据id删除订单操作记录 + /// + /// + /// + /// + Task DeleteOperationRecordByIdAsync(string id, string indexName); + /// /// 获取订单操作记录 /// diff --git a/src/SJZY.Track.Domain/Aggregates/TrackPushAggregate/ITrackPushRecordEsRepository.cs b/src/SJZY.Track.Domain/Aggregates/TrackPushAggregate/ITrackPushRecordEsRepository.cs index 261599f3..de1f0fa3 100644 --- a/src/SJZY.Track.Domain/Aggregates/TrackPushAggregate/ITrackPushRecordEsRepository.cs +++ b/src/SJZY.Track.Domain/Aggregates/TrackPushAggregate/ITrackPushRecordEsRepository.cs @@ -11,6 +11,14 @@ public interface ITrackPushRecordEsRepository Task AddManyAsync(List index); + /// + /// 根据id删除轨迹推送记录 + /// + /// + /// + /// + Task DeletePushRecordByIdAsync(string id, string indexName); + /// /// 获取轨迹推送记录 /// @@ -22,9 +30,9 @@ public interface ITrackPushRecordEsRepository /// 获取最新轨迹推送记录 /// /// + /// 只选择成功的记录 /// - Task GetLatestTrackPushRecordAsync(int orderId); - + Task GetLatestTrackPushRecordAsync(int orderId, bool isSuccess = false); /// /// 获取历史中推送成功的记录 diff --git a/src/SJZY.Track.Repository/Repositories/OrderOperationRecordEsRepository.cs b/src/SJZY.Track.Repository/Repositories/OrderOperationRecordEsRepository.cs index 0c30c860..7230d375 100644 --- a/src/SJZY.Track.Repository/Repositories/OrderOperationRecordEsRepository.cs +++ b/src/SJZY.Track.Repository/Repositories/OrderOperationRecordEsRepository.cs @@ -47,6 +47,18 @@ public class OrderOperationRecordEsRepository : ElasticSearchContextAbstract(id, u => u + .Index(indexName) + ); + + if (!r.IsValid) + { + throw new InvalidOperationException($"轨迹操作记录失败，id：{indexName}，原因：{r.DebugInformation}。"); + } + } + public async Task> GetOrderOperationsByIdAsync(int orderId) { var data = await Context.SearchAsync(x => x @@ -57,6 +69,6 @@ public class OrderOperationRecordEsRepository : ElasticSearchContextAbstract s.Field(f => f.CreateTime, SortOrder.Descending))); - return data.Documents.ToList(); + return data.ToIdList(); } } diff --git a/src/SJZY.Track.Repository/Repositories/TrackPushRecordEsRepository.cs b/src/SJZY.Track.Repository/Repositories/TrackPushRecordEsRepository.cs index b606bc29..3cbef97d 100644 --- a/src/SJZY.Track.Repository/Repositories/TrackPushRecordEsRepository.cs +++ b/src/SJZY.Track.Repository/Repositories/TrackPushRecordEsRepository.cs @@ -45,6 +45,18 @@ public class TrackPushRecordEsRepository : ElasticSearchContextAbstract(id, u => u + .Index(indexName) + ); + + if (!r.IsValid) + { + throw new InvalidOperationException($"轨迹推送记录失败，id：{indexName}，原因：{r.DebugInformation}。"); + } + } + public async Task> GetTrackPushRecordsAsync(int orderId) { @@ -56,15 +68,24 @@ public class TrackPushRecordEsRepository : ElasticSearchContextAbstract s.Field(f => f.CreateTime, SortOrder.Descending))); - return data.Documents.ToList(); + return data.ToIdList(); } - public async Task GetLatestTrackPushRecordAsync(int orderId) + public async Task GetLatestTrackPushRecordAsync(int orderId, bool isSuccess = false) { + var musts = new List, QueryContainer>> + { + mu => mu.Term(t => t.OrderId, orderId), + }; + + if (isSuccess) + musts.Add(mu => mu.Term(t => t.IsSuccess, true)); + var data = await Context.SearchAsync(x => x .Index(GlobalConst.TrackPushRecordIndex) .Query(q => - q.Term(t => t.OrderId, orderId) + q.Bool(b => + b.Must(musts)) ) .Size(1000) .Sort(s => s.Field(f => f.CreateTime, SortOrder.Descending)));