250 lines
11 KiB
C#
250 lines
11 KiB
C#
using DotNetCore.CAP;
|
||
using SJZY.Expand.ABP.Core.Logger;
|
||
using SJZY.Expand.ABP.Core.PublicMethords;
|
||
using SJZY.Track.Application.Contract.Events;
|
||
using SJZY.Track.Application.Contract.Services;
|
||
using SJZY.Track.Domain.Aggregates.OrderAggregate;
|
||
using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate;
|
||
using SJZY.Track.Domain.Consts;
|
||
using SJZY.Track.Domain.Dtos.Order;
|
||
using SJZY.Track.Domain.Queries;
|
||
using Volo.Abp.DependencyInjection;
|
||
using OrderIdentifier = SJZY.Track.Application.Contract.Events.OrderIdentifier;
|
||
|
||
namespace SJZY.Track.Application.CAP.EventHandlers;
|
||
|
||
public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe
|
||
{
|
||
private readonly IOrderRepository _orderRepository;
|
||
private readonly IProvideQueries _provideQueries;
|
||
private readonly ITrackDetailsRepository _trackDetailsRepository;
|
||
private readonly ICapPublisher _capPublisher;
|
||
private readonly ITrackPullCacheService _trackPullCacheService;
|
||
|
||
public TrackUpsertEventHandler(IOrderRepository orderRepository, ICapPublisher capPublisher,
|
||
IProvideQueries provideQueries, ITrackDetailsRepository trackDetailsRepository, ITrackPullCacheService trackPullCacheService)
|
||
{
|
||
_orderRepository = orderRepository;
|
||
_capPublisher = capPublisher;
|
||
_provideQueries = provideQueries;
|
||
_trackDetailsRepository = trackDetailsRepository;
|
||
_trackPullCacheService = trackPullCacheService;
|
||
}
|
||
|
||
[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($"{(@event.IsAdd ? "添加" : "修改")}轨迹时，发现循环依赖：{dup.JoinAsString(",")}");
|
||
return;
|
||
}
|
||
|
||
var canNextOrderIds = new List<int>();
|
||
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 => new { x.SubSystemId, x.CustomId }))
|
||
{
|
||
var nextEvent = @event.DeepCopy();
|
||
nextEvent.Visited.AddRange(orderIds);
|
||
nextEvent.SubSystemId = group.Key.SubSystemId;
|
||
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)
|
||
{
|
||
// 跨子系统没做映射的轨迹不处理
|
||
continue;
|
||
}
|
||
|
||
nextEvent.TrackCode = mapCode;
|
||
}
|
||
|
||
var identifiers = group.Select(x => new OrderIdentifier
|
||
{
|
||
OrderId = x.OrderId,
|
||
OrderNo = x.OrderNo,
|
||
OrderType = x.OrderType,
|
||
CustomId = x.CustomId
|
||
}).ToList();
|
||
var idx = 0;
|
||
while (idx < identifiers.Count)
|
||
{
|
||
var subEvent = nextEvent.DeepCopy();
|
||
var cnt = idx + GlobalConst.ChunkSize > identifiers.Count
|
||
? identifiers.Count - idx
|
||
: GlobalConst.ChunkSize;
|
||
subEvent.OrderIdentifiers = identifiers.GetRange(idx, cnt).ToArray();
|
||
idx += GlobalConst.ChunkSize;
|
||
|
||
await _capPublisher.PublishAsync(GlobalConst.TrackUpsertSubscribe, subEvent);
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
private async Task HandleTrackAsync(TrackUpsertEvent @event, List<int> canNextOrderIds)
|
||
{
|
||
var content = await _provideQueries.GetTrackContentAsync(@event.SubSystemId, @event.TrackCode);
|
||
|
||
if (content == null) return;
|
||
|
||
var orderIds = @event.OrderIdentifiers.Select(x => x.OrderId).ToArray();
|
||
var tracks = await _trackDetailsRepository.GetListAsync(orderIds, @event.TrackCode, true);
|
||
|
||
var newOrderIds = orderIds.Except(tracks.Select(x => x.OrderId)).ToArray();
|
||
var trackContents = await _provideQueries.MatchTrackContentAsync(newOrderIds, @event.TrackCode);
|
||
|
||
var insertTracks = new List<TrackDetailsIndex>();
|
||
var triggerPullOrder = new List<int>();
|
||
|
||
var providerDic=await GetProviderDic(orderIds);
|
||
foreach (var order in @event.OrderIdentifiers)
|
||
{
|
||
var track = tracks.FirstOrDefault(x => x.OrderId == order.OrderId);
|
||
|
||
if (track == null && !@event.IsAdd) continue; // 不存在不让改
|
||
if (track != null && @event.IsAdd) 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
|
||
};
|
||
|
||
// 根据轨迹code在订单模板内容中寻找匹配的
|
||
var trackContent = trackContents.FirstOrDefault(x => x.OrderId == order.OrderId);
|
||
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);
|
||
}
|
||
|
||
canNextOrderIds.Add(order.OrderId);
|
||
|
||
if(!providerDic.TryGetValue(order.OrderId,out var provider))
|
||
continue;
|
||
|
||
var config =
|
||
await _trackPullCacheService.GetPullParamWithContentConfigAsync(provider.SubSystemId,
|
||
provider.ServiceProviderId);
|
||
if (config!=null && config.ParamConfig!=null && config.ParamConfig.TrackContentCode == @event.TrackCode)
|
||
{
|
||
triggerPullOrder.Add(order.OrderId);
|
||
}
|
||
}
|
||
|
||
if (canNextOrderIds.Count == 0) return;
|
||
|
||
// 批量add to es
|
||
await _trackDetailsRepository.AddManyAsync(insertTracks);
|
||
|
||
// 触发订单主表更新事件（异步）
|
||
await _capPublisher.PublishAsync(GlobalConst.TrackChangeAfterUpdateOrderSubscribe,
|
||
new TrackChangeAfterUpdateOrderEvent
|
||
{
|
||
OrderIds = canNextOrderIds.ToArray(),
|
||
IsCompleted = @event.TrackRefreshData.IsCompleted,
|
||
TrackContent = content.TrackContent,
|
||
TrackEnContent = content.TrackEnContent,
|
||
TrackStatusId = content.TrackStatusId,
|
||
TrackPlace = @event.TrackPlace,
|
||
TrackTime = @event.TrackTime,
|
||
TrackTimeZone = @event.TrackTimeZone,
|
||
TriggerPullOrderIds = triggerPullOrder.ToArray()
|
||
});
|
||
|
||
// 触发轨迹推送事件（异步）
|
||
foreach (var group in @event.OrderIdentifiers
|
||
.Where(x => x.OrderId.IsIn(canNextOrderIds))
|
||
.GroupBy(x => x.CustomId))
|
||
{
|
||
var trackPushEvent = new TrackPushEvent
|
||
{
|
||
OptionType = @event.IsAdd ? OptionType.新增 : OptionType.修改,
|
||
SubSystemId = @event.SubSystemId,
|
||
CustomId = group.Key,
|
||
OrderIds = group.Select(x => x.OrderId).ToArray(),
|
||
TrackCode = @event.TrackCode,
|
||
SystemPushInfo = new SystemPushStruct
|
||
{
|
||
TrackStatus = content.TrackStatus,
|
||
TrackContent = content.TrackContent,
|
||
TrackPlace = @event.TrackPlace,
|
||
TrackTime = @event.TrackTime,
|
||
TrackTimeZone = @event.TrackTimeZone
|
||
}
|
||
};
|
||
await _capPublisher.PublishAsync(GlobalConst.TrackPushSubscribe, trackPushEvent);
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// 获取服务商字典
|
||
/// </summary>
|
||
/// <param name="orderIds"></param>
|
||
/// <returns></returns>
|
||
private async Task<Dictionary<int, OrderTransportDto>> GetProviderDic(int[] orderIds)
|
||
{
|
||
var providerList = await _provideQueries.FindOrderWithServiceProviderAsync(orderIds);
|
||
return providerList.GroupBy(val => val.OrderId)
|
||
.ToDictionary(val => val.Key, val => val.First());
|
||
}
|
||
|
||
} |