From f5b87695d92a599d06741ab5813e231aacb55361 Mon Sep 17 00:00:00 2001 From: xiaolipro <2357729423@qq.com> Date: Fri, 5 Sep 2025 10:42:15 +0800 Subject: [PATCH] fix cap uow --- ...TrackChangeAfterUpdateOrderEventHandler.cs | 2 + .../EventHandlers/TrackDeleteEventHandler.cs | 534 +++++++++--------- .../EventHandlers/TrackPushEventHandler.cs | 2 + .../EventHandlers/TrackUpsertEventHandler.cs | 2 + 4 files changed, 274 insertions(+), 266 deletions(-) diff --git a/src/TrackCenter.Application/CAP/EventHandlers/TrackChangeAfterUpdateOrderEventHandler.cs b/src/TrackCenter.Application/CAP/EventHandlers/TrackChangeAfterUpdateOrderEventHandler.cs index a6803eba..0c3dc3d5 100644 --- a/src/TrackCenter.Application/CAP/EventHandlers/TrackChangeAfterUpdateOrderEventHandler.cs +++ b/src/TrackCenter.Application/CAP/EventHandlers/TrackChangeAfterUpdateOrderEventHandler.cs @@ -1,4 +1,5 @@ using DotNetCore.CAP; +using Fake.UnitOfWork; using TrackCenter.Application.Events; using TrackCenter.Application.Services.Warning.Handlers; using TrackCenter.Domain.Aggregates.OrderAggregate; @@ -16,6 +17,7 @@ public class TrackChangeAfterUpdateOrderEventHandler : ITransientDependency, ICa _serviceProvider = serviceProvider; } + [UnitOfWork] [CapSubscribe(GlobalConst.TrackChangeAfterUpdateOrderSubscribe)] public async Task ConsumerAsync(TrackChangeAfterUpdateOrderEvent @event) { diff --git a/src/TrackCenter.Application/CAP/EventHandlers/TrackDeleteEventHandler.cs b/src/TrackCenter.Application/CAP/EventHandlers/TrackDeleteEventHandler.cs index fdcfe8ae..8cbbb65f 100644 --- a/src/TrackCenter.Application/CAP/EventHandlers/TrackDeleteEventHandler.cs +++ b/src/TrackCenter.Application/CAP/EventHandlers/TrackDeleteEventHandler.cs @@ -1,4 +1,5 @@ using DotNetCore.CAP; +using Fake.UnitOfWork; using InterfaceForward.Domain.Shared.Helpers; using TrackCenter.Application.Dtos.ProvideServices; using TrackCenter.Application.Events; @@ -41,301 +42,302 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe _warningSignalPublisher = warningSignalPublisher; } + [UnitOfWork] [CapSubscribe(GlobalConst.TrackDeleteSubscribe, Group = GlobalConst.TrackDeleteSubscribe)] - public async Task ConsumerAsync(TrackDeleteEvent @event) + public async Task ConsumerAsync(TrackDeleteEvent @event) + { + if (!CanDelete(@event)) return; + + // 处理删除轨迹 + await HandleDeleteTrackAsync(@event); + + // 根据订单关联关系异步向下刷（异步） + await DownByOrderLink(@event); + } + + private async Task DownByOrderLink(TrackDeleteEvent @event) + { + var linkedOrders = await _orderRepository.GetLinkedOrdersAsync(@event.OrderIds); + foreach (var orderId in @event.OrderIds) { - if (!CanDelete(@event)) return; - - // 处理删除轨迹 - await HandleDeleteTrackAsync(@event); - - // 根据订单关联关系异步向下刷（异步） - await DownByOrderLink(@event); - } - - private async Task DownByOrderLink(TrackDeleteEvent @event) - { - var linkedOrders = await _orderRepository.GetLinkedOrdersAsync(@event.OrderIds); - foreach (var orderId in @event.OrderIds) + foreach (var group in linkedOrders.Where(x => x.OwnerOrderId == orderId).GroupBy(x => x.AppKey)) { - foreach (var group in linkedOrders.Where(x => x.OwnerOrderId == orderId).GroupBy(x => x.AppKey)) + var nextEvent = @event.DeepCopy(); + nextEvent.Visited.AddRange(@event.OrderIds); + nextEvent.TrackIdentifiers = @event.TrackIdentifiers; + nextEvent.AppKey = group.Key; + + // 不同应用需要做内容映射 + if (@event.AppKey != nextEvent.AppKey) { - var nextEvent = @event.DeepCopy(); - nextEvent.Visited.AddRange(@event.OrderIds); - nextEvent.TrackIdentifiers = @event.TrackIdentifiers; - nextEvent.AppKey = group.Key; + var codeMaps = await _provideQueries.GetTrackCodeMapsAsync( + @event.AppKey, + nextEvent.AppKey, + @event.TrackIdentifiers.Select(x => x.TrackCode).ToArray() + ); - // 不同应用需要做内容映射 - if (@event.AppKey != nextEvent.AppKey) + if (codeMaps.Count == 0) { - var codeMaps = await _provideQueries.GetTrackCodeMapsAsync( - @event.AppKey, - nextEvent.AppKey, - @event.TrackIdentifiers.Select(x => x.TrackCode).ToArray() - ); - - if (codeMaps.Count == 0) - { - LogHelper.Warn($"订单跨应用，未配置内容映射，请联系管理员：{@event}", true); - continue; - } - - //映射轨迹code - foreach (var trackIdentity in nextEvent.TrackIdentifiers) - { - var match = codeMaps.FirstOrDefault(x => x.MapTrackContentCode == trackIdentity.TrackCode); - if (match == null) continue; - trackIdentity.TrackCode = match.TrackContentCode; - } + LogHelper.Warn($"订单跨应用，未配置内容映射，请联系管理员：{@event}", true); + continue; } - foreach (var subOrderIds in group.Select(x => x.OrderId).Chunk(GlobalConst.ChunkSize)) + //映射轨迹code + foreach (var trackIdentity in nextEvent.TrackIdentifiers) { - var subEvent = nextEvent.DeepCopy(); - subEvent.OrderIds = subOrderIds; - await _capPublisher.PublishAsync(GlobalConst.TrackUpsertSubscribe, subEvent); + var match = codeMaps.FirstOrDefault(x => x.MapTrackContentCode == trackIdentity.TrackCode); + if (match == null) continue; + trackIdentity.TrackCode = match.TrackContentCode; } } + + foreach (var subOrderIds in group.Select(x => x.OrderId).Chunk(GlobalConst.ChunkSize)) + { + var subEvent = nextEvent.DeepCopy(); + subEvent.OrderIds = subOrderIds; + await _capPublisher.PublishAsync(GlobalConst.TrackUpsertSubscribe, subEvent); + } } } + } - private static bool CanDelete(TrackDeleteEvent @event) - { - if (@event.OrderIds.Length == 0) - return true; - - // 订单循环关联检测 - var dup = @event.Visited.Intersect(@event.OrderIds).ToArray(); - if (dup.Any()) - { - LogHelper.Warn($"删除轨迹时，发现循环依赖：{dup.JoinAsString(",")}", true); - return false; - } - + private static bool CanDelete(TrackDeleteEvent @event) + { + if (@event.OrderIds.Length == 0) return true; + + // 订单循环关联检测 + var dup = @event.Visited.Intersect(@event.OrderIds).ToArray(); + if (dup.Any()) + { + LogHelper.Warn($"删除轨迹时，发现循环依赖：{dup.JoinAsString(",")}", true); + return false; } - private async Task HandleDeleteTrackAsync(TrackDeleteEvent @event) + return true; + } + + private async Task HandleDeleteTrackAsync(TrackDeleteEvent @event) + { + HashSet orderIds; + List oldTracks; + List deleteTracks; + // 优先根据EsId查询轨迹 + if (@event.EsIds.Length > 0) { - HashSet orderIds; - List oldTracks; - List deleteTracks; - // 优先根据EsId查询轨迹 - if (@event.EsIds.Length > 0) + deleteTracks = await _trackDetailsEsRepository.GetDataByIds(@event.EsIds); + + // 外部api获取的轨迹不允许删除 + if (deleteTracks.Any(m => m.Source == (int) TrackSourceEnum.ExternalServiceProvide)) return; + + // 从被删除轨迹寻找订单id，稍后会更新订单主表 + orderIds = deleteTracks.Select(m => m.OrderId).ToHashSet(); + + // 清理已经删除过的EsId,并设置用于下刷删除的其他信息 + @event.EsIds = Array.Empty(); + @event.OrderIds = orderIds.ToArray(); + @event.TrackIdentifiers = deleteTracks.Select(x => new TrackIdentifierDto { - deleteTracks = await _trackDetailsEsRepository.GetDataByIds(@event.EsIds); + TrackCode = x.TrackCode, + TrackTime = x.TrackTime + }).ToList(); - // 外部api获取的轨迹不允许删除 - if (deleteTracks.Any(m => m.Source == (int) TrackSourceEnum.ExternalServiceProvide)) return; + // 查出订单下的所有轨迹，以供后面回退使用 + oldTracks = await _trackDetailsEsRepository.GetListAsync(@event.OrderIds); + } + else + { + // 根据订单查询出所有的轨迹，查所有是因为主单上需要进行回退更新 + oldTracks = await _trackDetailsEsRepository.GetListAsync(@event.OrderIds); - // 从被删除轨迹寻找订单id，稍后会更新订单主表 - orderIds = deleteTracks.Select(m => m.OrderId).ToHashSet(); + // 构建轨迹标识符 + var trackIds = @event.TrackIdentifiers.Select(y => $"{y.TrackCode}{y.GetUtc0TrackTime():G}").ToHashSet(); - // 清理已经删除过的EsId,并设置用于下刷删除的其他信息 - @event.EsIds = Array.Empty(); - @event.OrderIds = orderIds.ToArray(); - @event.TrackIdentifiers = deleteTracks.Select(x => new TrackIdentifierDto + // 选出要删除的轨迹 + deleteTracks = oldTracks + .Where(x => trackIds.Contains($"{x.TrackCode}{x.GetUtc0TrackTime():G}")) + .WhereIf(!string.IsNullOrWhiteSpace(@event.CompareFields), y => { - TrackCode = x.TrackCode, - TrackTime = x.TrackTime - }).ToList(); + // 扩展信息不存在，跳过 + if (string.IsNullOrWhiteSpace(y.Extra)) return false; - // 查出订单下的所有轨迹，以供后面回退使用 - oldTracks = await _trackDetailsEsRepository.GetListAsync(@event.OrderIds); + // 从扩展信息提取比较值和传入值比较匹配 + var extraDict = TextJsonHelper.DeserializeToDictionary(y.Extra); + if (extraDict!.TryGetValue(@event.CompareFields!, out var compareValueJsonElement)) + return TextJsonHelper.EqJsonElement(compareValueJsonElement, @event.CompareValue); + + return false; + }) + .ToList(); + + //从被删除轨迹寻找订单id，稍后会更新订单主表 + orderIds = deleteTracks.Select(m => m.OrderId).ToHashSet(); + } + + if (deleteTracks.Count == 0) return; + + // 删除轨迹批量 + await _trackDetailsEsRepository.DeleteTrackByIdsAsync(deleteTracks.Select(x => x.Id).ToHashSet()); + + // 删除轨迹后，更新订单上的轨迹信息 + await TrackDeleteAfterUpdateOrderAsync(@event, oldTracks, deleteTracks, orderIds); + + // 删除轨迹广播 + await PublicTrackTopicAsync(@event, deleteTracks); + + // 发布预警信号，稍后将进行预警计算 + _warningSignalPublisher.PublishSignal(WarningSignalType.轨迹删除, deleteTracks); + } + + private async Task PublicTrackTopicAsync(TrackDeleteEvent @event, List delTracks) + { + var trackContentIds = delTracks.Select(x => x.TrackContentId).ToHashSet(); + var trackContents = await _provideQueries.GetTrackContentAsync(trackContentIds); + var contentDict = trackContents.ToDictionary(x => x.Id); + + var needPublicTracks = delTracks + // 轨迹内容已经不存在了，不需要广播 + .Where(x => contentDict.ContainsKey(x.TrackContentId)) + // 当删除操作没有标记广播时，根据轨迹状态/轨迹标记 + .WhereIf(@event.IsToMq == false, x => contentDict[x.TrackContentId].IsToMq) + .ToList(); + + if (needPublicTracks.Count == 0) return; + + //增加延时确保写盘成功 + await Task.Delay(1000); + + foreach (var track in needPublicTracks) + { + var eto = new TrackChangeEto + { + TrackEsId = track.Id, + Option = 2, + IsSuccess = true, + FailedReason = null, + EstimatedDeliveryDate = track.EstimatedDeliveryDate, + DestinationTimeZone = track.DestinationTimeZone, + TrackPlacePostcode = track.TrackPlacePostcode, + TrackPlaceNation = track.TrackPlaceNation, + TrackPlaceNationId = track.TrackPlaceNationId, + TrackPlaceProvince = track.TrackPlaceProvince, + TrackPlaceProvinceId = track.TrackPlaceProvinceId, + TrackPlaceCity = track.TrackPlaceCity, + TrackPlaceCityId = track.TrackPlaceCityId, + Files = track.Files?.Select(x => new FileItemDto + { + FileName = x.FileName, + OssKey = x.OssKey + }).ToList(), + Remarks = track.Remarks, + Source = track.Source, + IsExternal = track.IsExternal, + TrackTimeZone = track.TrackTimeZone, + TrackTime = track.TrackTime, + TrackPlace = track.TrackPlace, + TrackStatusCode = track.TrackStatusCode, + EnStatus = track.TrackEnContent, + Content = track.TrackContent, + TrackCode = track.TrackCode, + OperationMan = @event.OperationMan, + OrderNo = track.OrderNo, + OrderIdentifiers = new[] + { + new OrderIdentifier + { + OrderId = track.OrderId, + OrderNo = track.OrderNo, + OrderType = track.OrderType + } + }, + AppKey = track.AppKey, + Extra = track.Extra + }; + + var content = contentDict[track.TrackContentId]; + _broadcastingService.TrackChangePublicTopic(eto, content.AppKey); + } + } + + private async Task TrackDeleteAfterUpdateOrderAsync( + TrackDeleteEvent @event, + List oldTracks, + List deleteTracks, + HashSet orderIds + ) + { + // 排除掉已经删除的轨迹后的新的轨迹 + var deleteTrackIds = deleteTracks.Select(m => m.Id).ToList(); + var newTracks = oldTracks.Where(x => !deleteTrackIds.Contains(x.Id)).ToList(); + + var orders = await _orderRepository.GetListAsync(x => orderIds.Contains(x.Id), + x => new OrderEntity + { + Id = x.Id, + LatestTrackStatus = x.LatestTrackStatus, + LatestTrackContent = x.LatestTrackContent, + LatestTrackEnContent = x.LatestTrackEnContent, + LatestTrackPlace = x.LatestTrackPlace, + LatestTrackTime = x.LatestTrackTime, + LatestTrackTimeZone = x.LatestTrackTimeZone, + EstimatedDeliveryDate = x.EstimatedDeliveryDate, + DestinationTimeZone = x.DestinationTimeZone + }); + if (orders.Count == 0) return; + + foreach (var order in orders) + { + //找到最新的轨迹数据 + var track = newTracks?.OrderByDescending(x => x.ZeroTrackTime) + .FirstOrDefault(x => x.OrderId == order.Id); + + // 删除全部轨迹导致的重置行为 + if (track == null) + { + if (order.LatestTrackTime == null) continue; + + // tips：订单是支持从已完成态到未完成态的 + order.ResetTrackInfo(); } else { - // 根据订单查询出所有的轨迹，查所有是因为主单上需要进行回退更新 - oldTracks = await _trackDetailsEsRepository.GetListAsync(@event.OrderIds); - - // 构建轨迹标识符 - var trackIds = @event.TrackIdentifiers.Select(y => $"{y.TrackCode}{y.GetUtc0TrackTime():G}").ToHashSet(); - - // 选出要删除的轨迹 - deleteTracks = oldTracks - .Where(x => trackIds.Contains($"{x.TrackCode}{x.GetUtc0TrackTime():G}")) - .WhereIf(!string.IsNullOrWhiteSpace(@event.CompareFields), y => - { - // 扩展信息不存在，跳过 - if (string.IsNullOrWhiteSpace(y.Extra)) return false; - - // 从扩展信息提取比较值和传入值比较匹配 - var extraDict = TextJsonHelper.DeserializeToDictionary(y.Extra); - if (extraDict!.TryGetValue(@event.CompareFields!, out var compareValueJsonElement)) - return TextJsonHelper.EqJsonElement(compareValueJsonElement, @event.CompareValue); - - return false; - }) - .ToList(); - - //从被删除轨迹寻找订单id，稍后会更新订单主表 - orderIds = deleteTracks.Select(m => m.OrderId).ToHashSet(); + if (order.LatestTrackContent == track.TrackContent) continue; + // 删除引发的回退更新 + order.LatestTrackStatus = track.TrackStatus; + order.LatestTrackContent = track.TrackContent; + order.LatestTrackEnContent = track.TrackEnContent; + order.LatestTrackPlace = track.TrackPlace; + order.LatestTrackTime = track.TrackTime; + order.LatestTrackTimeZone = track.TrackTimeZone; + order.EstimatedDeliveryDate = track.EstimatedDeliveryDate; + order.DestinationTimeZone = track.DestinationTimeZone; } - - if (deleteTracks.Count == 0) return; - - // 删除轨迹批量 - await _trackDetailsEsRepository.DeleteTrackByIdsAsync(deleteTracks.Select(x => x.Id).ToHashSet()); - - // 删除轨迹后，更新订单上的轨迹信息 - await TrackDeleteAfterUpdateOrderAsync(@event, oldTracks, deleteTracks, orderIds); - - // 删除轨迹广播 - await PublicTrackTopicAsync(@event, deleteTracks); - - // 发布预警信号，稍后将进行预警计算 - _warningSignalPublisher.PublishSignal(WarningSignalType.轨迹删除, deleteTracks); } - private async Task PublicTrackTopicAsync(TrackDeleteEvent @event, List delTracks) + //更新订单上冗余轨迹明细的字段 + await _orderRepository.UpdateRangeAsync(orders, + x => new + { + x.LatestTrackStatus, + x.LatestTrackContent, + x.LatestTrackEnContent, + x.LatestTrackPlace, + x.LatestTrackTime, + x.LatestTrackTimeZone, + x.EstimatedDeliveryDate, + x.DestinationTimeZone + }); + + //批量写入订单操作日志 + _ = _orderOperationRecordEsRepository.AddManyAsync(orders.Select(order => new OrderOperationRecordIndex { - var trackContentIds = delTracks.Select(x => x.TrackContentId).ToHashSet(); - var trackContents = await _provideQueries.GetTrackContentAsync(trackContentIds); - var contentDict = trackContents.ToDictionary(x => x.Id); - - var needPublicTracks = delTracks - // 轨迹内容已经不存在了，不需要广播 - .Where(x => contentDict.ContainsKey(x.TrackContentId)) - // 当删除操作没有标记广播时，根据轨迹状态/轨迹标记 - .WhereIf(@event.IsToMq == false, x => contentDict[x.TrackContentId].IsToMq) - .ToList(); - - if (needPublicTracks.Count == 0) return; - - //增加延时确保写盘成功 - await Task.Delay(1000); - - foreach (var track in needPublicTracks) - { - var eto = new TrackChangeEto - { - TrackEsId = track.Id, - Option = 2, - IsSuccess = true, - FailedReason = null, - EstimatedDeliveryDate = track.EstimatedDeliveryDate, - DestinationTimeZone = track.DestinationTimeZone, - TrackPlacePostcode = track.TrackPlacePostcode, - TrackPlaceNation = track.TrackPlaceNation, - TrackPlaceNationId = track.TrackPlaceNationId, - TrackPlaceProvince = track.TrackPlaceProvince, - TrackPlaceProvinceId = track.TrackPlaceProvinceId, - TrackPlaceCity = track.TrackPlaceCity, - TrackPlaceCityId = track.TrackPlaceCityId, - Files = track.Files?.Select(x => new FileItemDto - { - FileName = x.FileName, - OssKey = x.OssKey - }).ToList(), - Remarks = track.Remarks, - Source = track.Source, - IsExternal = track.IsExternal, - TrackTimeZone = track.TrackTimeZone, - TrackTime = track.TrackTime, - TrackPlace = track.TrackPlace, - TrackStatusCode = track.TrackStatusCode, - EnStatus = track.TrackEnContent, - Content = track.TrackContent, - TrackCode = track.TrackCode, - OperationMan = @event.OperationMan, - OrderNo = track.OrderNo, - OrderIdentifiers = new[] - { - new OrderIdentifier - { - OrderId = track.OrderId, - OrderNo = track.OrderNo, - OrderType = track.OrderType - } - }, - AppKey = track.AppKey, - Extra = track.Extra - }; - - var content = contentDict[track.TrackContentId]; - _broadcastingService.TrackChangePublicTopic(eto, content.AppKey); - } - } - - private async Task TrackDeleteAfterUpdateOrderAsync( - TrackDeleteEvent @event, - List oldTracks, - List deleteTracks, - HashSet orderIds - ) - { - // 排除掉已经删除的轨迹后的新的轨迹 - var deleteTrackIds = deleteTracks.Select(m => m.Id).ToList(); - var newTracks = oldTracks.Where(x => !deleteTrackIds.Contains(x.Id)).ToList(); - - var orders = await _orderRepository.GetListAsync(x => orderIds.Contains(x.Id), - x => new OrderEntity - { - Id = x.Id, - LatestTrackStatus = x.LatestTrackStatus, - LatestTrackContent = x.LatestTrackContent, - LatestTrackEnContent = x.LatestTrackEnContent, - LatestTrackPlace = x.LatestTrackPlace, - LatestTrackTime = x.LatestTrackTime, - LatestTrackTimeZone = x.LatestTrackTimeZone, - EstimatedDeliveryDate = x.EstimatedDeliveryDate, - DestinationTimeZone = x.DestinationTimeZone - }); - if (orders.Count == 0) return; - - foreach (var order in orders) - { - //找到最新的轨迹数据 - var track = newTracks?.OrderByDescending(x => x.ZeroTrackTime) - .FirstOrDefault(x => x.OrderId == order.Id); - - // 删除全部轨迹导致的重置行为 - if (track == null) - { - if (order.LatestTrackTime == null) continue; - - // tips：订单是支持从已完成态到未完成态的 - order.ResetTrackInfo(); - } - else - { - if (order.LatestTrackContent == track.TrackContent) continue; - // 删除引发的回退更新 - order.LatestTrackStatus = track.TrackStatus; - order.LatestTrackContent = track.TrackContent; - order.LatestTrackEnContent = track.TrackEnContent; - order.LatestTrackPlace = track.TrackPlace; - order.LatestTrackTime = track.TrackTime; - order.LatestTrackTimeZone = track.TrackTimeZone; - order.EstimatedDeliveryDate = track.EstimatedDeliveryDate; - order.DestinationTimeZone = track.DestinationTimeZone; - } - } - - //更新订单上冗余轨迹明细的字段 - await _orderRepository.UpdateRangeAsync(orders, - x => new - { - x.LatestTrackStatus, - x.LatestTrackContent, - x.LatestTrackEnContent, - x.LatestTrackPlace, - x.LatestTrackTime, - x.LatestTrackTimeZone, - x.EstimatedDeliveryDate, - x.DestinationTimeZone - }); - - //批量写入订单操作日志 - _ = _orderOperationRecordEsRepository.AddManyAsync(orders.Select(order => new OrderOperationRecordIndex - { - OrderId = order.Id, - OrderNo = order.OrderNo, - OptionType = OrderOperationType.删除轨迹.ToString(), - Content = deleteTracks.Where(m => m.OrderId == order.Id).ToList().ToJson(), - CreateUserId = @event.OperationManId, - CreateUserName = @event.OperationMan ?? "", - CreateTime = DateTime.Now - }).ToList()); - } + OrderId = order.Id, + OrderNo = order.OrderNo, + OptionType = OrderOperationType.删除轨迹.ToString(), + Content = deleteTracks.Where(m => m.OrderId == order.Id).ToList().ToJson(), + CreateUserId = @event.OperationManId, + CreateUserName = @event.OperationMan ?? "", + CreateTime = DateTime.Now + }).ToList()); + } } diff --git a/src/TrackCenter.Application/CAP/EventHandlers/TrackPushEventHandler.cs b/src/TrackCenter.Application/CAP/EventHandlers/TrackPushEventHandler.cs index d5b0d923..28eb4252 100644 --- a/src/TrackCenter.Application/CAP/EventHandlers/TrackPushEventHandler.cs +++ b/src/TrackCenter.Application/CAP/EventHandlers/TrackPushEventHandler.cs @@ -1,6 +1,7 @@ using System.Collections.Immutable; using System.Text.Json; using DotNetCore.CAP; +using Fake.UnitOfWork; using InterfaceForward.Domain.Shared.Helpers; using Microsoft.Extensions.Options; using Newtonsoft.Json; @@ -49,6 +50,7 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe _interfaceForwardService = interfaceForwardService; } + [UnitOfWork] [CapSubscribe(GlobalConst.TrackPushSubscribe, Group = GlobalConst.TrackPushSubscribe)] [CapSubscribe(GlobalConst.TrackManualPushSubscribe, Group = GlobalConst.TrackPushSubscribe)] public async Task ConsumerAsync(TrackPushEvent @event) diff --git a/src/TrackCenter.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs b/src/TrackCenter.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs index 4b5bce7e..71a815bb 100644 --- a/src/TrackCenter.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs +++ b/src/TrackCenter.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs @@ -1,4 +1,5 @@ using DotNetCore.CAP; +using Fake.UnitOfWork; using InterfaceForward.Domain.Shared.Helpers; using TrackCenter.Application.Events; using TrackCenter.Application.Kafka; @@ -43,6 +44,7 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe _warningSignalPublisher = warningSignalPublisher; } + [UnitOfWork] [CapSubscribe(GlobalConst.TrackUpsertSubscribe, Group = GlobalConst.TrackUpsertSubscribe)] [CapSubscribe(GlobalConst.TrackUpsertForRefreshSubscribe, Group = GlobalConst.TrackUpsertForRefreshSubscribe)] public async Task ConsumerAsync(TrackUpsertEvent @event)