using DotNetCore.CAP; using SJZY.Expand.ABP.Core.Logger; using SJZY.Expand.ABP.Core.PublicMethords; using SJZY.Track.Application.Contract.Events; using SJZY.Track.Domain.Aggregates.OrderAggregate; using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate; using SJZY.Track.Domain.Consts; using SJZY.Track.Domain.Queries; using Volo.Abp.DependencyInjection; namespace SJZY.Track.Application.EventHandlers; public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe { private readonly IOrderRepository _orderRepository; private readonly IProvideQueries _provideQueries; private readonly ITrackDetailsRepository _trackDetailsRepository; private readonly ICapPublisher _capPublisher; public TrackUpsertEventHandler(IOrderRepository orderRepository, ICapPublisher capPublisher, IProvideQueries provideQueries, ITrackDetailsRepository trackDetailsRepository) { _orderRepository = orderRepository; _capPublisher = capPublisher; _provideQueries = provideQueries; _trackDetailsRepository = trackDetailsRepository; } [CapSubscribe(GlobalConst.TrackUpsertSubscribe, Group = GlobalConst.TrackUpsertSubscribe)] [CapSubscribe(GlobalConst.TrackUpsertForRefreshSubscribe, Group = GlobalConst.TrackUpsertForRefreshSubscribe)] public async Task ConsumerAsync(TrackUpsertEvent @event) { var orderIds = @event.OrderIdentifiers.Select(x => x.OrderId).ToArray(); var dup = @event.Visited.Intersect(orderIds).ToArray(); if (dup.Any()) { LogHelper.Error($"删除轨迹订单时,发现循环依赖:{dup.JoinAsString(",")}"); return; } var canNextOrderIds = new List(); await HandleTrackAsync(@event, canNextOrderIds); // 根据订单关联关系异步向下刷(异步) var linkedOrders = await _orderRepository.GetLinkedOrdersAsync(canNextOrderIds.ToArray()); foreach (var order in @event.OrderIdentifiers) { // 理论上小包在这不会分组 foreach (var group in linkedOrders.Where(x => x.OwnerOrderId == order.OrderId) .GroupBy(x => x.SubSystemId)) { var nextEvent = @event.DeepCopy(); nextEvent.Visited.AddRange(orderIds); nextEvent.SubSystemId = group.Key; nextEvent.OrderIdentifiers = group.Select(x => new OrderIdentifier { OrderId = x.OrderId, OrderNo = x.OrderNo, OrderType = x.OrderType }).ToArray(); nextEvent.TrackRefreshData = new TrackRefreshStruct { ServiceProviderName = @event.TrackRefreshData.ServiceProviderName, OriginTrackContent = @event.TrackRefreshData.OriginTrackContent, OriginTrackData = @event.TrackRefreshData.OriginTrackData, // tips:一定要重置为false,父级订单的完成不会使得关联订单完成,关联单也必须由刷新来完成或手动设置完成!! IsCompleted = false }; // 不同子系统需要做内容映射 if (@event.SubSystemId != nextEvent.SubSystemId) { var mapCode = (await _provideQueries .MapTrackCodesAsync(@event.SubSystemId, nextEvent.SubSystemId, @event.TrackCode)) .FirstOrDefault(); if (mapCode == default) { // 跨子系统没做映射的轨迹不处理 return; } nextEvent.TrackCode = mapCode; } await _capPublisher.PublishAsync(GlobalConst.TrackUpsertSubscribe, nextEvent); } } } private async Task HandleTrackAsync(TrackUpsertEvent @event, List canNextOrderIds) { var orderIds = @event.OrderIdentifiers.Select(x => x.OrderId).ToArray(); var tracks = await _trackDetailsRepository.GetListAsync(orderIds, @event.TrackCode); TrackDetailsIndex firstTrack = null; var insertTracks = new List(); foreach (var order in @event.OrderIdentifiers) { var track = tracks.FirstOrDefault(x => x.OrderId == order.OrderId); if (@event.IsAdd != null) { if (track == null && !@event.IsAdd!.Value) continue; // 不存在不让改 if (track != null && @event.IsAdd!.Value) continue; // 存在不让创建 } if (track == default) { track = new TrackDetailsIndex { OrderId = order.OrderId, OrderNo = order.OrderNo, OrderType = order.OrderType, TrackCode = @event.TrackCode, TrackPlace = @event.TrackPlace, TrackTime = @event.TrackTime, TrackTimeZone = @event.TrackTimeZone, ServiceProviderName = @event.TrackRefreshData.ServiceProviderName ?? GlobalConst.EmptyString, OriginTrackContent = @event.TrackRefreshData.OriginTrackContent ?? GlobalConst.EmptyString, OriginTrackData = @event.TrackRefreshData.OriginTrackData ?? GlobalConst.EmptyString, IsCanDelete = false, CreateUser = @event.OperationMan ?? GlobalConst.SystemUser, CreateTime = DateTime.Now }; //todo:可以优化? // 根据轨迹code在订单模板内容中寻找匹配的 var trackContent = await _provideQueries.MatchTrackContentAsync(order.OrderId, @event.TrackCode); if (trackContent == default) { // 订单所绑定的模板,没有一条内容可以匹配发生的轨迹 continue; } track.TrackStatus = trackContent.Status; track.TrackCode = trackContent.TrackCode; track.TrackContent = trackContent.Content; track.TrackEnContent = trackContent.EnContent; track.IsCanDelete = trackContent.IsCanDelete; track.ZeroTrackTime = track.TrackTime.Subtract(TimeSpan.FromHours(track.TrackTimeZone)); insertTracks.Add(track); } else { track.TrackPlace = @event.TrackPlace; track.TrackTime = @event.TrackTime; track.TrackTimeZone = @event.TrackTimeZone; track.UpdateUser = @event.OperationMan ?? GlobalConst.SystemUser; track.UpdateTime = DateTime.Now; await _trackDetailsRepository.UpdateAsync(track); } // step2、触发轨迹推送事件(异步) var trackPushEvent = new TrackPushEvent { OptionType = track.Id.IsNullOrWhiteSpace() ? OptionType.新增 : OptionType.修改, OrderId = order.OrderId, TrackStatus = track.TrackStatus, TrackCode = track.TrackCode, TrackContent = track.TrackContent, TrackPlace = track.TrackPlace, TrackTime = track.TrackTime, TrackTimeZone = track.TrackTimeZone, }; await _capPublisher.PublishAsync(GlobalConst.TrackPushSubscribe, trackPushEvent); canNextOrderIds.Add(order.OrderId); firstTrack ??= track; } // 批量add to es await _trackDetailsRepository.AddManyAsync(insertTracks); if (firstTrack == null) return; // 触发订单主表更新事件(异步) await _capPublisher.PublishAsync(GlobalConst.TrackChangeAfterUpdateOrderSubscribe, new TrackChangeAfterUpdateOrderEvent { OrderIds = orderIds, IsCompleted = @event.TrackRefreshData.IsCompleted, TrackCode = @event.TrackCode, TrackContent = firstTrack.TrackContent, TrackEnContent = firstTrack.TrackEnContent, TrackPlace = @event.TrackPlace, TrackTime = @event.TrackTime, TrackTimeZone = @event.TrackTimeZone }); } }