fix cap uow

This commit is contained in:
xiaolipro 2025-09-05 10:42:15 +08:00
parent 198b24ba64
commit f5b87695d9
4 changed files with 274 additions and 266 deletions

View File

@ -1,4 +1,5 @@
using DotNetCore.CAP; using DotNetCore.CAP;
using Fake.UnitOfWork;
using TrackCenter.Application.Events; using TrackCenter.Application.Events;
using TrackCenter.Application.Services.Warning.Handlers; using TrackCenter.Application.Services.Warning.Handlers;
using TrackCenter.Domain.Aggregates.OrderAggregate; using TrackCenter.Domain.Aggregates.OrderAggregate;
@ -16,6 +17,7 @@ public class TrackChangeAfterUpdateOrderEventHandler : ITransientDependency, ICa
_serviceProvider = serviceProvider; _serviceProvider = serviceProvider;
} }
[UnitOfWork]
[CapSubscribe(GlobalConst.TrackChangeAfterUpdateOrderSubscribe)] [CapSubscribe(GlobalConst.TrackChangeAfterUpdateOrderSubscribe)]
public async Task ConsumerAsync(TrackChangeAfterUpdateOrderEvent @event) public async Task ConsumerAsync(TrackChangeAfterUpdateOrderEvent @event)
{ {

View File

@ -1,4 +1,5 @@
using DotNetCore.CAP; using DotNetCore.CAP;
using Fake.UnitOfWork;
using InterfaceForward.Domain.Shared.Helpers; using InterfaceForward.Domain.Shared.Helpers;
using TrackCenter.Application.Dtos.ProvideServices; using TrackCenter.Application.Dtos.ProvideServices;
using TrackCenter.Application.Events; using TrackCenter.Application.Events;
@ -41,301 +42,302 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe
_warningSignalPublisher = warningSignalPublisher; _warningSignalPublisher = warningSignalPublisher;
} }
[UnitOfWork]
[CapSubscribe(GlobalConst.TrackDeleteSubscribe, Group = GlobalConst.TrackDeleteSubscribe)] [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; foreach (var group in linkedOrders.Where(x => x.OwnerOrderId == orderId).GroupBy(x => x.AppKey))
// 处理删除轨迹
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)) 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(); var codeMaps = await _provideQueries.GetTrackCodeMapsAsync(
nextEvent.Visited.AddRange(@event.OrderIds); @event.AppKey,
nextEvent.TrackIdentifiers = @event.TrackIdentifiers; nextEvent.AppKey,
nextEvent.AppKey = group.Key; @event.TrackIdentifiers.Select(x => x.TrackCode).ToArray()
);
// 不同应用需要做内容映射 if (codeMaps.Count == 0)
if (@event.AppKey != nextEvent.AppKey)
{ {
var codeMaps = await _provideQueries.GetTrackCodeMapsAsync( LogHelper.Warn($"订单跨应用,未配置内容映射,请联系管理员:{@event}", true);
@event.AppKey, continue;
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;
}
} }
foreach (var subOrderIds in group.Select(x => x.OrderId).Chunk(GlobalConst.ChunkSize)) //映射轨迹code
foreach (var trackIdentity in nextEvent.TrackIdentifiers)
{ {
var subEvent = nextEvent.DeepCopy(); var match = codeMaps.FirstOrDefault(x => x.MapTrackContentCode == trackIdentity.TrackCode);
subEvent.OrderIds = subOrderIds; if (match == null) continue;
await _capPublisher.PublishAsync(GlobalConst.TrackUpsertSubscribe, subEvent); 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) private static bool CanDelete(TrackDeleteEvent @event)
{ {
if (@event.OrderIds.Length == 0) 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;
}
return true; 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<int> orderIds;
List<TrackDetailsIndex> oldTracks;
List<TrackDetailsIndex> deleteTracks;
// 优先根据EsId查询轨迹
if (@event.EsIds.Length > 0)
{ {
HashSet<int> orderIds; deleteTracks = await _trackDetailsEsRepository.GetDataByIds(@event.EsIds);
List<TrackDetailsIndex> oldTracks;
List<TrackDetailsIndex> deleteTracks; // 外部api获取的轨迹不允许删除
// 优先根据EsId查询轨迹 if (deleteTracks.Any(m => m.Source == (int) TrackSourceEnum.ExternalServiceProvide)) return;
if (@event.EsIds.Length > 0)
// 从被删除轨迹寻找订单id,稍后会更新订单主表
orderIds = deleteTracks.Select(m => m.OrderId).ToHashSet();
// 清理已经删除过的EsId,并设置用于下刷删除的其他信息
@event.EsIds = Array.Empty<string>();
@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<string>(); deleteTracks = oldTracks
@event.OrderIds = orderIds.ToArray(); .Where(x => trackIds.Contains($"{x.TrackCode}{x.GetUtc0TrackTime():G}"))
@event.TrackIdentifiers = deleteTracks.Select(x => new TrackIdentifierDto .WhereIf(!string.IsNullOrWhiteSpace(@event.CompareFields), y =>
{ {
TrackCode = x.TrackCode, // 扩展信息不存在,跳过
TrackTime = x.TrackTime if (string.IsNullOrWhiteSpace(y.Extra)) return false;
}).ToList();
// 查出订单下的所有轨迹,以供后面回退使用 // 从扩展信息提取比较值和传入值比较匹配
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<TrackDetailsIndex> 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<TrackDetailsIndex> oldTracks,
List<TrackDetailsIndex> deleteTracks,
HashSet<int> 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 else
{ {
// 根据订单查询出所有的轨迹,查所有是因为主单上需要进行回退更新 if (order.LatestTrackContent == track.TrackContent) continue;
oldTracks = await _trackDetailsEsRepository.GetListAsync(@event.OrderIds); // 删除引发的回退更新
order.LatestTrackStatus = track.TrackStatus;
// 构建轨迹标识符 order.LatestTrackContent = track.TrackContent;
var trackIds = @event.TrackIdentifiers.Select(y => $"{y.TrackCode}{y.GetUtc0TrackTime():G}").ToHashSet(); order.LatestTrackEnContent = track.TrackEnContent;
order.LatestTrackPlace = track.TrackPlace;
// 选出要删除的轨迹 order.LatestTrackTime = track.TrackTime;
deleteTracks = oldTracks order.LatestTrackTimeZone = track.TrackTimeZone;
.Where(x => trackIds.Contains($"{x.TrackCode}{x.GetUtc0TrackTime():G}")) order.EstimatedDeliveryDate = track.EstimatedDeliveryDate;
.WhereIf(!string.IsNullOrWhiteSpace(@event.CompareFields), y => order.DestinationTimeZone = track.DestinationTimeZone;
{
// 扩展信息不存在,跳过
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 (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<TrackDetailsIndex> 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(); OrderId = order.Id,
var trackContents = await _provideQueries.GetTrackContentAsync(trackContentIds); OrderNo = order.OrderNo,
var contentDict = trackContents.ToDictionary(x => x.Id); OptionType = OrderOperationType.删除轨迹.ToString(),
Content = deleteTracks.Where(m => m.OrderId == order.Id).ToList().ToJson(),
var needPublicTracks = delTracks CreateUserId = @event.OperationManId,
// 轨迹内容已经不存在了,不需要广播 CreateUserName = @event.OperationMan ?? "",
.Where(x => contentDict.ContainsKey(x.TrackContentId)) CreateTime = DateTime.Now
// 当删除操作没有标记广播时,根据轨迹状态/轨迹标记 }).ToList());
.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<TrackDetailsIndex> oldTracks,
List<TrackDetailsIndex> deleteTracks,
HashSet<int> 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());
}
} }

View File

@ -1,6 +1,7 @@
using System.Collections.Immutable; using System.Collections.Immutable;
using System.Text.Json; using System.Text.Json;
using DotNetCore.CAP; using DotNetCore.CAP;
using Fake.UnitOfWork;
using InterfaceForward.Domain.Shared.Helpers; using InterfaceForward.Domain.Shared.Helpers;
using Microsoft.Extensions.Options; using Microsoft.Extensions.Options;
using Newtonsoft.Json; using Newtonsoft.Json;
@ -49,6 +50,7 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe
_interfaceForwardService = interfaceForwardService; _interfaceForwardService = interfaceForwardService;
} }
[UnitOfWork]
[CapSubscribe(GlobalConst.TrackPushSubscribe, Group = GlobalConst.TrackPushSubscribe)] [CapSubscribe(GlobalConst.TrackPushSubscribe, Group = GlobalConst.TrackPushSubscribe)]
[CapSubscribe(GlobalConst.TrackManualPushSubscribe, Group = GlobalConst.TrackPushSubscribe)] [CapSubscribe(GlobalConst.TrackManualPushSubscribe, Group = GlobalConst.TrackPushSubscribe)]
public async Task ConsumerAsync(TrackPushEvent @event) public async Task ConsumerAsync(TrackPushEvent @event)

View File

@ -1,4 +1,5 @@
using DotNetCore.CAP; using DotNetCore.CAP;
using Fake.UnitOfWork;
using InterfaceForward.Domain.Shared.Helpers; using InterfaceForward.Domain.Shared.Helpers;
using TrackCenter.Application.Events; using TrackCenter.Application.Events;
using TrackCenter.Application.Kafka; using TrackCenter.Application.Kafka;
@ -43,6 +44,7 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe
_warningSignalPublisher = warningSignalPublisher; _warningSignalPublisher = warningSignalPublisher;
} }
[UnitOfWork]
[CapSubscribe(GlobalConst.TrackUpsertSubscribe, Group = GlobalConst.TrackUpsertSubscribe)] [CapSubscribe(GlobalConst.TrackUpsertSubscribe, Group = GlobalConst.TrackUpsertSubscribe)]
[CapSubscribe(GlobalConst.TrackUpsertForRefreshSubscribe, Group = GlobalConst.TrackUpsertForRefreshSubscribe)] [CapSubscribe(GlobalConst.TrackUpsertForRefreshSubscribe, Group = GlobalConst.TrackUpsertForRefreshSubscribe)]
public async Task ConsumerAsync(TrackUpsertEvent @event) public async Task ConsumerAsync(TrackUpsertEvent @event)