From 82bbca3002e0685b437b28742ec029c169ac7476 Mon Sep 17 00:00:00 2001 From: hardstylewyl <827778732@qq.com> Date: Wed, 15 Jan 2025 17:41:06 +0800 Subject: [PATCH] =?UTF-8?q?update:=E8=B0=83=E6=95=B4=E8=BD=A8=E8=BF=B9?= =?UTF-8?q?=E6=8E=A8=E9=80=81=E9=80=BB=E8=BE=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../EventHandlers/TrackPushEventHandler.cs | 98 +++++++++++++------ .../Services/ProvideServices/TrackService.cs | 9 ++ .../IOrderOperationRecordEsRepository.cs | 8 ++ .../ITrackPushRecordEsRepository.cs | 12 ++- .../OrderOperationRecordEsRepository.cs | 14 ++- .../TrackPushRecordEsRepository.cs | 27 ++++- 6 files changed, 132 insertions(+), 36 deletions(-) diff --git a/src/SJZY.Track.Application/CAP/EventHandlers/TrackPushEventHandler.cs b/src/SJZY.Track.Application/CAP/EventHandlers/TrackPushEventHandler.cs index 09d58a78..a7cc3d36 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)] @@ -206,11 +211,15 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe } // 推送新增，需要根据上一次的推送 推断 本次推送 是否可行，为了约束顺序 - config = GetValidPushConfigAsync(_pushConfigs, order.LatestPushTrackCode, @event); - if (config == null) + // 如果当前推送code与订单上的最后的code不一致，则考虑推送映射顺序 + if (!string.Equals(order.LatestPushTrackCode, @event.TrackCode)) { - TrackPushFailedPublicTopic("推送顺序异常，找不到有效推送配置", @event, order); - continue; + config = GetValidPushConfigAsync(_pushConfigs, order.LatestPushTrackCode, @event); + if (config == null) + { + TrackPushFailedPublicTopic("推送顺序异常，找不到有效推送配置", @event, order); + continue; + } } // 是否自动推送 @@ -362,38 +371,67 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe if (config.IsAllPush) { - var history = await _trackPushRecordEsRepository.GetLatestTrackPushRecordAsync(log.OrderId); + var history = await _trackPushRecordEsRepository.GetLatestTrackPushRecordAsync(log.OrderId, true); if (history != default) { body = TextJsonHelper.Deserialize(history.PushContent); - if (config.IsAllPushWithLatest) - { - //同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 tracks = await _trackDetailsEsRepository.GetListAsync(log.OrderId); + //确定轨迹标识符 + var trackIdentifies = tracks.Select(x => { - body.Data.Remove(old); + //寻找映射配置(转换后的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(); + var timeStr = ((JsonElement) x["date"]!).GetString(); - // //添加当前要推送的轨迹 - body!.Data.Add(cur); + 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(); + + // if (config.IsAllPushWithLatest) + // { + // //同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.Add(cur); } } diff --git a/src/SJZY.Track.Application/Services/ProvideServices/TrackService.cs b/src/SJZY.Track.Application/Services/ProvideServices/TrackService.cs index 92ce9710..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; @@ -120,6 +121,14 @@ 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) 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)));