550 lines
24 KiB
C#
550 lines
24 KiB
C#
using System.Diagnostics;
|
||
using DotNetCore.CAP;
|
||
using Microsoft.Extensions.Logging;
|
||
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.Application.Services.Warning.Handlers;
|
||
using SJZY.Track.Domain.Aggregates.OrderAggregate;
|
||
using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate;
|
||
using SJZY.Track.Domain.Aggregates.WarningAggregate.Signals;
|
||
using SJZY.Track.Domain.Dtos.TrackContents;
|
||
using SJZY.Track.Domain.Enums;
|
||
using SJZY.Track.Domain.Extensions;
|
||
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 ICapPublisher _capPublisher;
|
||
private readonly IOrderRepository _orderRepository;
|
||
private readonly IProvideQueries _provideQueries;
|
||
private readonly ITrackDetailsEsRepository _trackDetailsEsRepository;
|
||
private readonly IBroadcastingService _broadcastingService;
|
||
private readonly ILogger<TrackUpsertEventHandler> _logger;
|
||
private readonly IOrderOperationRecordEsRepository _orderOperationRecordEsRepository;
|
||
private readonly IServiceProvider _serviceProvider;
|
||
|
||
public TrackUpsertEventHandler(
|
||
IOrderRepository orderRepository,
|
||
ICapPublisher capPublisher,
|
||
IProvideQueries provideQueries,
|
||
ITrackDetailsEsRepository trackDetailsEsRepository,
|
||
IOrderOperationRecordEsRepository orderOperationRecordEsRepository,
|
||
IBroadcastingService broadcastingService,
|
||
ILogger<TrackUpsertEventHandler> logger,
|
||
IServiceProvider serviceProvider)
|
||
{
|
||
_orderRepository = orderRepository;
|
||
_capPublisher = capPublisher;
|
||
_provideQueries = provideQueries;
|
||
_trackDetailsEsRepository = trackDetailsEsRepository;
|
||
_orderOperationRecordEsRepository = orderOperationRecordEsRepository;
|
||
_broadcastingService = broadcastingService;
|
||
_logger = logger;
|
||
_serviceProvider = serviceProvider;
|
||
}
|
||
|
||
[CapSubscribe(GlobalConst.TrackUpsertSubscribe, Group = GlobalConst.TrackUpsertSubscribe)]
|
||
[CapSubscribe(GlobalConst.TrackUpsertForRefreshSubscribe, Group = GlobalConst.TrackUpsertForRefreshSubscribe)]
|
||
public async Task ConsumerAsync(TrackUpsertEvent @event)
|
||
{
|
||
if (!CanDelete(@event)) return;
|
||
|
||
var canNextOrderIds = new List<int>();
|
||
|
||
var content = await HandleTrackAsync(@event, canNextOrderIds);
|
||
|
||
// 处理广播事件
|
||
if (content != null && (content.IsToMq || @event.IsToMq))
|
||
_ = EnsureWriteAfterPublicTopicAsync(@event, content.IsExternal, content.SubSystemCode, content.TrackStatusCode);
|
||
|
||
if (canNextOrderIds.Count == 0) return;
|
||
// 根据订单关联关系异步向下刷（异步）
|
||
await DownByOrderLink(@event, canNextOrderIds);
|
||
}
|
||
|
||
private async Task DownByOrderLink(TrackUpsertEvent @event, List<int> 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(canNextOrderIds);
|
||
nextEvent.SubSystemId = group.Key.SubSystemId;
|
||
nextEvent.CustomId = group.Key.CustomId;
|
||
nextEvent.TrackRefreshData = new TrackRefreshStruct
|
||
{
|
||
ServiceProviderCode = @event.TrackRefreshData.ServiceProviderCode,
|
||
OriginTrackContent = @event.TrackRefreshData.OriginTrackContent,
|
||
OriginTrackDataHash = @event.TrackRefreshData.OriginTrackDataHash,
|
||
};
|
||
|
||
// 不同子系统需要做内容映射
|
||
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,
|
||
}).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 bool CanDelete(TrackUpsertEvent @event)
|
||
{
|
||
// 系统创建的轨迹不允许修改
|
||
if (@event.Visited.IsNullOrEmpty() && @event is { OperationMan: GlobalConst.SystemUser, IsAdd: false })
|
||
{
|
||
LogHelper.Warn($"系统创建的轨迹不允许被修改：{@event}", true);
|
||
return false;
|
||
}
|
||
|
||
var orderIds = @event.OrderIdentifiers.Select(x => x.OrderId).ToArray();
|
||
|
||
var dup = @event.Visited.Intersect(orderIds).ToArray();
|
||
if (dup.Any())
|
||
{
|
||
LogHelper.Warn($"{(@event.IsAdd ? "添加" : "修改")}轨迹时，发现循环依赖：{dup.JoinAsString(",")}", true);
|
||
return false;
|
||
}
|
||
|
||
return true;
|
||
}
|
||
|
||
private async Task<TrackContentVO?> HandleTrackAsync(TrackUpsertEvent @event, List<int> canNextOrderIds)
|
||
{
|
||
var trackContent = await _provideQueries.GetTrackContentAsync(@event.SubSystemId, @event.TrackCode);
|
||
|
||
if (trackContent == null)
|
||
{
|
||
// 找不到轨迹内容，直接结束
|
||
LogHelper.Warn($"找不到轨迹内容：{@event}", true);
|
||
return null;
|
||
}
|
||
|
||
//#7001:未来潜在需求，当轨迹内容匹配预设，记录一个标记，保证下刷过程都是使用预设内容
|
||
//虽然在引发事件之前做了判断，引发之后并不一定立即处理，在此过程中轨迹内容仍然可能被修改，所以要二次进行判断。
|
||
//通过来源指示是否为手动操作
|
||
var manualOperation = @event.Source == (int) TrackSourceEnum.Artificial;
|
||
//轨迹内容标识不允许手动创建的
|
||
if (manualOperation && @event.IsAdd && trackContent.IsAllowManuallyCreate == false) return trackContent;
|
||
//轨迹内容标识不允许手动更新的
|
||
if (manualOperation && !@event.IsAdd && trackContent.IsAllowManuallyUpdate == false) return trackContent;
|
||
|
||
// k:[orderId] v:[k:EsId,v:Version]
|
||
var orderIdEsIdVersionDict = new Dictionary<int, KeyValuePair<string, long>>();
|
||
|
||
var insertTracks = await UpsertTrackAsync(@event, canNextOrderIds, trackContent, orderIdEsIdVersionDict);
|
||
|
||
if (canNextOrderIds.Count == 0) return trackContent;
|
||
|
||
if (insertTracks.Any())
|
||
{
|
||
// 批量add to es tips:注意副作用填充版本号
|
||
await _trackDetailsEsRepository.AddManyAsync(insertTracks);
|
||
|
||
// es新增后填充版本号字典
|
||
insertTracks.ForEach(t => orderIdEsIdVersionDict.Add(t.OrderId, KeyValuePair.Create(t.Id, t.Version)));
|
||
|
||
//批量写入订单操作日志
|
||
_ = _orderOperationRecordEsRepository.AddManyAsync(insertTracks.Select(t => BuildOrderOperationRecord(@event, t, new OrderIdentifier
|
||
{
|
||
OrderNo = t.OrderNo,
|
||
OrderId = t.OrderId
|
||
})).ToList());
|
||
|
||
// 添加轨迹触发订单拉取状态更新
|
||
await _orderRepository.UpdatePullStatusToSilenceAsync(insertTracks.Select(x => x.OrderId).ToList(),
|
||
@event.TrackCode);
|
||
|
||
// todo:引发预警计算 改MQ批量+分流
|
||
WarningCalc_AddedTrack(insertTracks);
|
||
|
||
// 添加轨迹触发预警计算（异步）
|
||
await _capPublisher.PublishAsync(GlobalConst.TrackWarningCalcSubscribe, new TrackWarningCalcEvent
|
||
{
|
||
SubSystemId = @event.SubSystemId,
|
||
CustomId = @event.CustomId,
|
||
OrderIds = insertTracks.Select(x => x.OrderId).ToArray(),
|
||
TrackStatusId = trackContent.TrackStatusId,
|
||
TrackTime = @event.TrackTime,
|
||
});
|
||
}
|
||
|
||
// 触发订单主表更新事件（异步）
|
||
await _capPublisher.PublishAsync(GlobalConst.TrackChangeAfterUpdateOrderSubscribe,
|
||
new TrackChangeAfterUpdateOrderEvent
|
||
{
|
||
OrderIds = canNextOrderIds.ToArray(),
|
||
IsCompleted = trackContent.IsCompleted,
|
||
TrackStatus = trackContent.TrackStatus,
|
||
TrackContent = trackContent.TrackContent,
|
||
TrackEnContent = trackContent.TrackEnContent,
|
||
TrackPlace = @event.TrackPlace,
|
||
TrackTime = @event.NewTrackTime ?? @event.TrackTime,
|
||
TrackTimeZone = @event.TrackTimeZone,
|
||
DestinationTimeZone = @event.DestinationTimeZone,
|
||
EstimatedDeliveryDate = @event.EstimatedDeliveryDate
|
||
});
|
||
|
||
// 触发轨迹推送事件（异步）
|
||
// tips:同一个轨迹内容，多个订单多条轨迹
|
||
var trackPushEvent = new TrackPushEvent
|
||
{
|
||
OptionType = @event.IsAdd ? OptionType.新增 : OptionType.修改,
|
||
SubSystemId = @event.SubSystemId,
|
||
SubSystemCode = trackContent.SubSystemCode,
|
||
CustomId = @event.CustomId,
|
||
OrderIdEsIdVersionDict = orderIdEsIdVersionDict,
|
||
TrackCode = @event.TrackCode,
|
||
SystemPushInfo = new SystemPushStruct
|
||
{
|
||
TrackStatusId = trackContent.TrackStatusId,
|
||
TrackStatus = trackContent.TrackStatus,
|
||
TrackContentId = trackContent.TrackContentId,
|
||
TrackContent = trackContent.TrackContent,
|
||
TrackPlace = @event.TrackPlace,
|
||
TrackTime = @event.NewTrackTime ?? @event.TrackTime,
|
||
TrackTimeZone = @event.TrackTimeZone,
|
||
CompareFields = @event.CompareFields,
|
||
Extra = @event.Extra
|
||
},
|
||
ForcedPush = @event.ForcedPush,
|
||
IsToMq = (trackContent.IsToMq || @event.IsToMq)
|
||
};
|
||
await _capPublisher.PublishAsync(GlobalConst.TrackPushSubscribe, trackPushEvent);
|
||
|
||
return trackContent;
|
||
}
|
||
|
||
private async Task<List<TrackDetailsIndex>> UpsertTrackAsync(
|
||
TrackUpsertEvent @event,
|
||
List<int> canNextOrderIds,
|
||
TrackContentVO trackContent,
|
||
Dictionary<int, KeyValuePair<string, long>> orderIdEsIdVersionDict
|
||
)
|
||
{
|
||
var orderIds = @event.OrderIdentifiers.Select(x => x.OrderId).ToArray();
|
||
|
||
// important：轨迹code相同，时间、发生地不同的，可能存在多条！！
|
||
var tracks = await _trackDetailsEsRepository.GetListAsync(orderIds, @event.TrackCode, true);
|
||
|
||
var insertTracks = new List<TrackDetailsIndex>();
|
||
var updateTracks = new List<TrackDetailsIndex>();
|
||
var operationRecords = new List<OrderOperationRecordIndex>();
|
||
foreach (var order in @event.OrderIdentifiers)
|
||
{
|
||
// fast path: by update with es id case
|
||
if (!@event.IsAdd && !string.IsNullOrWhiteSpace(@event.EsId))
|
||
{
|
||
var trackById = tracks.FirstOrDefault(x => x.Id == @event.EsId);
|
||
if (trackById != null)
|
||
{
|
||
updateTracks.Add(BuiltUpdateTrack(@event, trackById, order, trackContent));
|
||
}
|
||
|
||
continue;
|
||
}
|
||
|
||
var existTracks = tracks.Where(x => x.OrderId == order.OrderId).ToList();
|
||
|
||
if (@event.CompareFields.IsNullOrEmpty())
|
||
{
|
||
existTracks = existTracks.Where(x => x.TrackTime == @event.TrackTime).ToList();
|
||
}
|
||
else
|
||
{
|
||
Debug.Assert(@event.CompareFields != null, "@event.CompareFields != null");
|
||
var extraDict = TextJsonHelper.DeserializeToDictionary(@event.Extra);
|
||
if (extraDict == null || !extraDict.TryGetValue(@event.CompareFields, out var compareValue))
|
||
{
|
||
throw new BusinessException(message: $"请检查拓展字段，为什么找不到比较字段：{@event.CompareFields}");
|
||
}
|
||
|
||
existTracks = existTracks.Where(x =>
|
||
{
|
||
var oldExtraDict = TextJsonHelper.DeserializeToDictionary(x.Extra);
|
||
// tips：注意兼容性，历史轨迹的字段可能为空
|
||
if (oldExtraDict == null || !oldExtraDict.TryGetValue(@event.CompareFields, out var oldValue))
|
||
{
|
||
return false;
|
||
}
|
||
|
||
return TextJsonHelper.EqJsonElement(oldValue, compareValue);
|
||
}).ToList();
|
||
}
|
||
|
||
var track = existTracks.FirstOrDefault();
|
||
|
||
if (!@event.IsAdd)
|
||
{
|
||
if (existTracks.Count > 1)
|
||
{
|
||
_logger.LogInformation("订单存在多条轨迹不允许修改：{@event}", @event);
|
||
continue;
|
||
}
|
||
|
||
if (track == null)
|
||
{
|
||
_logger.LogInformation("订单不存在轨迹不允许修改：{@event}", @event);
|
||
continue; // 不存在不让改
|
||
}
|
||
|
||
updateTracks.Add(BuiltUpdateTrack(@event, track, order, trackContent));
|
||
operationRecords.Add(BuildOrderOperationRecord(@event, track, order));
|
||
}
|
||
else
|
||
{
|
||
// 不允许重复轨迹，跳过
|
||
if (!trackContent.IsRepeatable && existTracks.Count > 0)
|
||
{
|
||
_logger.LogInformation("订单已经存在不允许重复的轨迹，不允许重复添加：{@event}", @event);
|
||
continue;
|
||
}
|
||
|
||
insertTracks.Add(BuiltNewTrack(@event, order, trackContent));
|
||
}
|
||
|
||
canNextOrderIds.Add(order.OrderId);
|
||
}
|
||
|
||
if (updateTracks.Any())
|
||
{
|
||
// 批量差异化更新轨迹
|
||
await _trackDetailsEsRepository.BulkUpdateAsync(updateTracks);
|
||
// 获取更新后的版本号
|
||
updateTracks.ForEach(x => orderIdEsIdVersionDict.Add(x.OrderId, KeyValuePair.Create(x.Id, x.Version)));
|
||
// 批量写入操作日志
|
||
_ = _orderOperationRecordEsRepository.AddManyAsync(operationRecords);
|
||
|
||
// todo:引发预警计算 改MQ批量+分流
|
||
WarningCalc_UpdatedTrack(updateTracks);
|
||
}
|
||
|
||
return insertTracks;
|
||
}
|
||
|
||
|
||
private TrackDetailsIndex BuiltUpdateTrack(TrackUpsertEvent @event, TrackDetailsIndex track, OrderIdentifier order,
|
||
TrackContentVO trackContent)
|
||
{
|
||
//外部轨迹，轨迹发生地，轨迹发生时间，轨迹发生时区，不允许更改
|
||
if (track.Source != (int) TrackSourceEnum.ExternalServiceProvide)
|
||
{
|
||
track.TrackPlace = @event.TrackPlace;
|
||
track.TrackTime = @event.NewTrackTime ?? @event.TrackTime;
|
||
track.TrackTimeZone = @event.TrackTimeZone;
|
||
}
|
||
|
||
track.ZeroTrackTime = Math.Abs(track.TrackTimeZone - GlobalConst.NoneTrackTimeZone) < 1
|
||
? track.TrackTime
|
||
: track.TrackTime.Subtract(TimeSpan.FromHours(track.TrackTimeZone));
|
||
track.UpdateUser = @event.OperationMan ?? GlobalConst.SystemUser;
|
||
track.UpdateTime = DateTime.Now;
|
||
|
||
//允许二次修改，在修改了轨迹内容后进行修改轨迹操作，如果输入的轨迹内容则以输入为准
|
||
//跨子系统?
|
||
track.TrackContent = string.IsNullOrWhiteSpace(@event.Content) ? trackContent.TrackContent : @event.Content;
|
||
track.TrackEnContent = string.IsNullOrWhiteSpace(@event.EnStatus) ? trackContent.TrackEnContent : @event.EnStatus;
|
||
//S06新增
|
||
track.IsVisibility = @event.IsVisibility;
|
||
//当传入的IsExternal为空，则使用轨迹内容中的IsExternal
|
||
track.IsExternal = @event.IsExternal ?? trackContent.IsExternal;
|
||
track.Remarks = @event.Remarks ?? GlobalConst.EmptyString;
|
||
track.Files = @event.Files?.Select(x => new FileItem
|
||
{
|
||
OssKey = x.OssKey,
|
||
FileName = x.FileName
|
||
}).ToList();
|
||
track.TrackPlaceCity = @event.TrackPlaceCity ?? GlobalConst.EmptyString;
|
||
track.TrackPlaceCityId = @event.TrackPlaceCityId ?? GlobalConst.NotExistId;
|
||
track.TrackPlaceProvince = @event.TrackPlaceProvince ?? GlobalConst.EmptyString;
|
||
track.TrackPlaceProvinceId = @event.TrackPlaceProvinceId ?? GlobalConst.NotExistId;
|
||
track.TrackPlaceNation = @event.TrackPlaceNation ?? GlobalConst.EmptyString;
|
||
track.TrackPlaceNationId = @event.TrackPlaceNationId ?? GlobalConst.NotExistId;
|
||
track.TrackPlacePostcode = @event.TrackPlacePostcode ?? GlobalConst.EmptyString;
|
||
track.DestinationTimeZone = @event.DestinationTimeZone;
|
||
track.EstimatedDeliveryDate = @event.EstimatedDeliveryDate;
|
||
track.Extra = @event.Extra;
|
||
|
||
return track;
|
||
}
|
||
|
||
private TrackDetailsIndex BuiltNewTrack(TrackUpsertEvent @event, OrderIdentifier order, TrackContentVO trackContent)
|
||
{
|
||
var track = new TrackDetailsIndex
|
||
{
|
||
SubSystemId = @event.SubSystemId,
|
||
OrderId = order.OrderId,
|
||
OrderNo = order.OrderNo,
|
||
OrderType = order.OrderType,
|
||
TrackCode = @event.TrackCode,
|
||
TrackPlace = @event.TrackPlace,
|
||
TrackTime = @event.NewTrackTime ?? @event.TrackTime,
|
||
TrackTimeZone = @event.TrackTimeZone,
|
||
ServiceProviderCode = @event.TrackRefreshData.ServiceProviderCode ?? GlobalConst.EmptyString,
|
||
OriginTrackContent = @event.TrackRefreshData.OriginTrackContent ?? GlobalConst.EmptyString,
|
||
OriginTrackDataHash = @event.TrackRefreshData.OriginTrackDataHash ?? GlobalConst.EmptyString,
|
||
CreateUser = @event.OperationMan ?? GlobalConst.SystemUser,
|
||
CreateTime = DateTime.Now,
|
||
//来源轨迹内容,第一次新增的时候使用轨迹内容表提供的
|
||
TrackStatusId = trackContent.TrackStatusId,
|
||
TrackStatus = trackContent.TrackStatus,
|
||
TrackStatusCode = trackContent.TrackStatusCode,
|
||
//跨子系统?
|
||
TrackContentId = trackContent.TrackContentId,
|
||
TrackContent = string.IsNullOrWhiteSpace(@event.Content) ? trackContent.TrackContent : @event.Content,
|
||
TrackEnContent = string.IsNullOrWhiteSpace(@event.EnStatus) ? trackContent.TrackEnContent : @event.EnStatus,
|
||
//S06新增轨迹明细字段
|
||
IsVisibility = @event.IsVisibility,
|
||
//当传入的IsExternal为空，则使用轨迹内容中的IsExternal
|
||
IsExternal = @event.IsExternal ?? trackContent.IsExternal,
|
||
Source = @event.Source,
|
||
Remarks = @event.Remarks ?? GlobalConst.EmptyString,
|
||
Files = @event.Files?.Select(x => new FileItem
|
||
{
|
||
OssKey = x.OssKey,
|
||
FileName = x.FileName
|
||
}).ToList(),
|
||
TrackPlaceCity = @event.TrackPlaceCity ?? GlobalConst.EmptyString,
|
||
TrackPlaceCityId = @event.TrackPlaceCityId ?? GlobalConst.NotExistId,
|
||
TrackPlaceProvince = @event.TrackPlaceProvince ?? GlobalConst.EmptyString,
|
||
TrackPlaceProvinceId = @event.TrackPlaceProvinceId ?? GlobalConst.NotExistId,
|
||
TrackPlaceNation = @event.TrackPlaceNation ?? GlobalConst.EmptyString,
|
||
TrackPlaceNationId = @event.TrackPlaceNationId ?? GlobalConst.NotExistId,
|
||
TrackPlacePostcode = @event.TrackPlacePostcode ?? GlobalConst.EmptyString,
|
||
DestinationTimeZone = @event.DestinationTimeZone,
|
||
EstimatedDeliveryDate = @event.EstimatedDeliveryDate,
|
||
Extra = @event.Extra
|
||
};
|
||
|
||
track.ZeroTrackTime = Math.Abs(track.TrackTimeZone - GlobalConst.NoneTrackTimeZone) < 1
|
||
? track.TrackTime
|
||
: track.TrackTime.Subtract(TimeSpan.FromHours(track.TrackTimeZone));
|
||
|
||
return track;
|
||
}
|
||
|
||
private OrderOperationRecordIndex BuildOrderOperationRecord(TrackUpsertEvent @event, TrackDetailsIndex track, OrderIdentifier order)
|
||
{
|
||
return new OrderOperationRecordIndex
|
||
{
|
||
OrderId = order.OrderId,
|
||
OrderNo = order.OrderNo,
|
||
OptionType = @event.IsAdd ? OrderOperationType.创建轨迹.ToString() : OrderOperationType.修改轨迹.ToString(),
|
||
Content = track.ToJson(),
|
||
CreateUserId = @event.OperationManId,
|
||
CreateUserName = @event.OperationMan ?? "",
|
||
CreateTime = DateTime.Now
|
||
};
|
||
}
|
||
|
||
private async Task EnsureWriteAfterPublicTopicAsync(TrackUpsertEvent @event, bool contentIsExternal, string subSystemCode, string statusCode)
|
||
{
|
||
//理想情况在写入之后1s会刷盘 https://www.elastic.co/guide/en/elasticsearch/reference/master/indices-refresh.html
|
||
//增加一个版本号来确定是否真的变更了
|
||
//没有刷盘成功事件回调
|
||
//方案1：!!【轮询等待静默刷盘】查出版本号确实发生了变更在发送广播
|
||
//方案2：考虑强制刷盘（对es压力大）后发送
|
||
//方案3：采用一个较大延时，不查询直接发
|
||
await Task.Delay(1000);
|
||
_broadcastingService.TrackChangePublicTopic(@event, contentIsExternal, subSystemCode, statusCode);
|
||
}
|
||
|
||
private void WarningCalc_AddedTrack(List<TrackDetailsIndex> tracks)
|
||
{
|
||
var signals = tracks.Select(x => new TrackAddedSignal
|
||
{
|
||
EsId = x.Id,
|
||
OrderId = x.OrderId,
|
||
TrackStatusId = x.TrackStatusId
|
||
}).ToList();
|
||
|
||
_ = Task.Run(async () =>
|
||
{
|
||
var scope = _serviceProvider.CreateAsyncScope();
|
||
try
|
||
{
|
||
var handler = scope.ServiceProvider.GetRequiredService<ICompleteWarningHandler>();
|
||
var handler2 = scope.ServiceProvider.GetRequiredService<ITransportWarningHandler>();
|
||
|
||
await Task.Delay(1500); //确保落盘
|
||
await handler.TrackAddedHandleAsync(signals);
|
||
await handler2.TrackAddedOrUpdatedHandleAsync(signals.Cast<IWarningSignal>().ToList());
|
||
}
|
||
finally
|
||
{
|
||
await scope.DisposeAsync();
|
||
}
|
||
});
|
||
}
|
||
|
||
|
||
private void WarningCalc_UpdatedTrack(List<TrackDetailsIndex> tracks)
|
||
{
|
||
var signals = tracks.Select(x => new TrackUpdatedSignal
|
||
{
|
||
OrderId = x.OrderId,
|
||
BeforeChangeTrack = x
|
||
}).ToList();
|
||
_ = Task.Run(async () =>
|
||
{
|
||
var scope = _serviceProvider.CreateAsyncScope();
|
||
try
|
||
{
|
||
var handler = scope.ServiceProvider.GetRequiredService<ITransportWarningHandler>();
|
||
var handler2 = scope.ServiceProvider.GetRequiredService<IReturnWarningHandler>();
|
||
|
||
await Task.Delay(1500); //确保落盘
|
||
await handler.TrackAddedOrUpdatedHandleAsync(signals.Cast<IWarningSignal>().ToList());
|
||
await handler2.TrackUpdatedOrDeletedHandleAsync(signals.Cast<IWarningSignal>().ToList());
|
||
}
|
||
finally
|
||
{
|
||
await scope.DisposeAsync();
|
||
}
|
||
});
|
||
}
|
||
}
|