track/src/SJZY.Track.Application/CAP/EventHandlers/TrackPushEventHandler.cs

549 lines
23 KiB
C#
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

using System.Collections.Immutable;
using System.Text.Json;
using DotNetCore.CAP;
using Microsoft.Extensions.Options;
using Newtonsoft.Json;
using SJZY.Expand.ABP.Core.PublicMethords;
using SJZY.InterfaceRelay.Application.Contracts;
using SJZY.Track.Application.Contract.Dtos.TrackPush;
using SJZY.Track.Application.Contract.Events;
using SJZY.Track.Application.Contract.Services;
using SJZY.Track.Application.Helpers;
using SJZY.Track.Application.Options;
using SJZY.Track.Application.Track2InterfaceRelay;
using SJZY.Track.Domain.Aggregates.OrderAggregate;
using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate;
using SJZY.Track.Domain.Aggregates.TrackPushAggregate;
using SJZY.Track.Domain.Enums;
using Volo.Abp.DependencyInjection;
using Volo.Abp.Http.Client.DynamicProxying;
namespace SJZY.Track.Application.CAP.EventHandlers;
public class TrackPushEventHandler : ITransientDependency, ICapSubscribe
{
private readonly IHttpClientProxy<IInterfaceForwardServiceV2> _httpClientProxy;
private readonly IOptionsMonitor<InterfaceRelayParamsConfigOptions> _options;
private readonly IOrderRepository _orderRepository;
private readonly ITrackPushCacheService _trackPushCacheService;
private readonly ITrackPushRecordEsRepository _trackPushRecordEsRepository;
private readonly ITrackDetailsEsRepository _trackDetailsEsRepository;
private readonly IBroadcastingService _broadcastingService;
private List<OrderEntity> _orders = null!;
private List<TrackPushContentConfigDto> _pushConfigs = null!;
public TrackPushEventHandler(IHttpClientProxy<IInterfaceForwardServiceV2> httpClientProxy,
IOrderRepository orderRepository,
ITrackPushCacheService trackPushCacheService,
ITrackPushRecordEsRepository trackPushRecordEsRepository,
IOptionsMonitor<InterfaceRelayParamsConfigOptions> options,
IBroadcastingService broadcastingService,
ITrackDetailsEsRepository trackDetailsEsRepository)
{
_httpClientProxy = httpClientProxy;
_orderRepository = orderRepository;
_trackPushCacheService = trackPushCacheService;
_trackPushRecordEsRepository = trackPushRecordEsRepository;
_options = options;
_broadcastingService = broadcastingService;
_trackDetailsEsRepository = trackDetailsEsRepository;
}
[CapSubscribe(GlobalConst.TrackPushSubscribe, Group = GlobalConst.TrackPushSubscribe)]
[CapSubscribe(GlobalConst.TrackManualPushSubscribe, Group = GlobalConst.TrackPushSubscribe)]
public async Task ConsumerAsync(TrackPushEvent @event)
{
var pushConfigs = await _trackPushCacheService.GetPushContentConfigsAsync(@event.SubSystemId, @event.CustomId);
if (pushConfigs == null || pushConfigs.Count == 0) return;
if (pushConfigs.First().ServiceProviderCode.IsNullOrWhiteSpace())
{
throw new BusinessException(GlobalConst.CapNoRetryErrorCode, $"该客户{@event.CustomId}未关联服务商,无法推送");
}
_pushConfigs = pushConfigs;
var orders = await _orderRepository.GetListAsync(x =>
@event.OrderIds.Contains(x.Id),
x => new OrderEntity
{
Id = x.Id,
OrderNo = x.OrderNo,
LatestPushTrackCode = x.LatestPushTrackCode,
OrderType = x.OrderType
});
if (orders.Count == 0) return;
_orders = orders;
if (@event.ForcedPush)
{
//当为强制推送,会走手动推送逻辑
await DoPushAsync(@event);
return;
}
switch (@event.OptionType)
{
case OptionType.无:
return;
case OptionType.新增:
await DoPushAsync(@event);
break;
case OptionType.删除:
break;
case OptionType.修改:
await DoUpdateAsync(@event);
break;
default:
throw new ArgumentOutOfRangeException();
}
}
private async Task DoPushAsync(TrackPushEvent @event)
{
var items = new List<TrackPushInput>();
if (@event.OperationMan == null)
{
if (@event.OptionType == OptionType.新增)
{
await BuiltSystemPushItemsAsync(@event, items);
}
else if (@event.OptionType == OptionType.修改)
{
await BuiltManualPushItemsAsync(@event, items);
}
}
else
{
// 手工推送,不会校验已推和顺序问题,因为可能是更新
await BuiltManualPushItemsAsync(@event, items);
}
if (items.Count == 0) return;
using (Track2InterfaceRelaySubSystemContext.Use(
new Track2InterfaceRelaySubSystemContext(@event.SubSystemCode)))
{
var asciiString = Base64Helper.SerializeObject(OptionType.新增.ToString());
var serviceProviderCode = _pushConfigs.First().ServiceProviderCode;
var r = await _httpClientProxy.Service.BatchForwardThenPushToQueueAsync(
_options.CurrentValue.PushTrackInterfaceCode,
serviceProviderCode, items.Cast<object>().ToList(), GlobalConst.TrackPushCompletedSubscribe,
attach: asciiString);
if (r.Code != GlobalConst.SuccessCode)
{
TrackPushFailedPublicTopic($"轨迹中台调用接口通失败,RequestId: {r.RequestId}; Msg: {r.Msg}", @event, r.RequestId,
items.Cast<dynamic>().ToList());
throw new BusinessException(
message: $"requestId: {r.RequestId}; serviceProviderCode:{serviceProviderCode} Msg: {r.Msg}");
}
}
}
private async Task BuiltManualPushItemsAsync(TrackPushEvent @event, List<TrackPushInput> items)
{
// tips: 手工强制推送直接使用配置
var config = _pushConfigs.FirstOrDefault(x => x.TrackCode == @event.TrackCode);
foreach (var order in _orders)
{
if (config is { IsAllPush: true })
{
// 非强制推送需要 推断 本次推送 是否可行
if (!@event.ForcedPush)
{
config = GetValidPushConfigAsync(_pushConfigs, order.LatestPushTrackCode, @event);
}
}
if (config == null)
{
TrackPushFailedPublicTopic("推送顺序异常,找不到有效推送配置", @event, order);
return;
}
//配置的类型如何和订单类型不一致则不允许推送
if (config.OrderType != order.OrderType)
{
TrackPushFailedPublicTopic($"订单类型{order.OrderType}和配置中的订单类型[{config.OrderType}]不一致,不能推送", @event,
order);
continue;
}
var log = new TrackPushRecordIndex
{
OptionType = @event.OptionType.ToString(),
OrderId = order.Id,
OrderNo = order.OrderNo,
TrackStatus = @event.SystemPushInfo.TrackStatus,
TrackCode = @event.TrackCode,
TrackContent = @event.SystemPushInfo.TrackContent,
TrackPlace = @event.SystemPushInfo.TrackPlace,
TrackTime = @event.SystemPushInfo.TrackTime,
TrackTimeZone = @event.SystemPushInfo.TrackTimeZone,
CreateUser = @event.OperationMan,
SubSystemCode = @event.SubSystemCode,
IsToMq = @event.IsToMq,
Extra = @event.SystemPushInfo.Extra
};
var item = await BuiltPushItem(config, log, @event.SystemPushInfo.CompareFields);
item.Log = JsonConvert.SerializeObject(log);
items.Add(item);
}
}
private async Task BuiltSystemPushItemsAsync(TrackPushEvent @event, List<TrackPushInput> items)
{
var config = _pushConfigs.FirstOrDefault(x => x.TrackCode == @event.TrackCode);
foreach (var order in _orders)
{
if (config == null)
{
TrackPushFailedPublicTopic($"没有在推送映射配置中找到关于轨迹代码[{@event.TrackCode}]的配置", @event, order);
return;
}
// 系统不推异常单,只能人工操作
if (order.QuestionType == (int) TrackProblemType.Error)
{
TrackPushFailedPublicTopic("系统不推异常单,只能人工操作", @event, order);
continue;
}
// 推送新增,需要根据上一次的推送 推断 本次推送 是否可行,为了约束顺序
// 如果当前推送code与订单上的最后的code不一致,则考虑推送映射顺序
if (!string.Equals(order.LatestPushTrackCode, @event.TrackCode))
{
config = GetValidPushConfigAsync(_pushConfigs, order.LatestPushTrackCode, @event);
if (config == null)
{
TrackPushFailedPublicTopic("推送顺序异常,找不到有效推送配置", @event, order);
continue;
}
}
// 是否自动推送
if (!config.IsAutoPush)
{
TrackPushFailedPublicTopic("没有设置为自动推送,如果要推送请手动推送", @event, order);
continue;
}
// 配置的类型如何和订单类型不一致则不允许推送
if (config.OrderType != order.OrderType)
{
TrackPushFailedPublicTopic($"订单类型{order.OrderType}和配置中的订单类型[{config.OrderType}]不一致,不能推送", @event,
order);
continue;
}
var log = new TrackPushRecordIndex
{
OptionType = @event.OptionType.ToString(),
OrderId = order.Id,
OrderNo = order.OrderNo,
TrackStatus = @event.SystemPushInfo.TrackStatus,
TrackCode = @event.TrackCode,
TrackContent = @event.SystemPushInfo.TrackContent,
TrackPlace = @event.SystemPushInfo.TrackPlace,
TrackTime = @event.SystemPushInfo.TrackTime,
TrackTimeZone = @event.SystemPushInfo.TrackTimeZone,
CreateUser = GlobalConst.SystemUser,
SubSystemCode = @event.SubSystemCode,
IsToMq = @event.IsToMq,
Extra = @event.SystemPushInfo.Extra
};
var item = await BuiltPushItem(config, log, @event.SystemPushInfo.CompareFields);
item.Log = JsonConvert.SerializeObject(log);
items.Add(item);
}
}
private async Task DoUpdateAsync(TrackPushEvent @event)
{
var config = _pushConfigs.FirstOrDefault(x => x.TrackCode == @event.TrackCode);
if (config == null || !config.IsCanUpdate) return;
var items = new List<TrackUpdateInput>();
foreach (var order in _orders)
{
//配置的类型如何和订单类型不一致则不允许推送
if (config.OrderType != order.OrderType)
{
TrackPushFailedPublicTopic($"订单类型{order.OrderType}和配置中的订单类型[{config.OrderType}]不一致,不能推送", @event,
order);
continue;
}
var log = new TrackPushRecordIndex
{
OptionType = @event.OptionType.ToString(),
OrderId = order.Id,
OrderNo = order.OrderNo,
TrackStatus = @event.SystemPushInfo.TrackStatus,
TrackCode = @event.TrackCode,
TrackContent = @event.SystemPushInfo.TrackContent,
TrackPlace = @event.SystemPushInfo.TrackPlace,
TrackTime = @event.SystemPushInfo.TrackTime,
TrackTimeZone = @event.SystemPushInfo.TrackTimeZone,
CreateUser = GlobalConst.SystemUser,
SubSystemCode = @event.SubSystemCode,
IsToMq = @event.IsToMq,
Extra = @event.SystemPushInfo.Extra
};
var item = BuiltUpdateItem(config, log);
item.Log = JsonConvert.SerializeObject(log);
items.Add(item);
}
if (items.Count == 0) return;
using (Track2InterfaceRelaySubSystemContext.Use(
new Track2InterfaceRelaySubSystemContext(@event.SubSystemCode)))
{
var asciiString = Base64Helper.SerializeObject(OptionType.修改.ToString());
var serviceProviderCode = _pushConfigs.First().ServiceProviderCode;
var r = await _httpClientProxy.Service.BatchForwardThenPushToQueueAsync(
_options.CurrentValue.TrackUpdateInterfaceCode,
serviceProviderCode, items.Cast<object>().ToList(), GlobalConst.TrackPushCompletedSubscribe,
attach: asciiString);
if (r.Code != GlobalConst.SuccessCode)
{
TrackPushFailedPublicTopic($"轨迹中台调用接口通失败,RequestId: {r.RequestId}; Msg: {r.Msg}", @event, r.RequestId,
items.Cast<dynamic>().ToList());
throw new BusinessException(
message: $"requestId: {r.RequestId}; serviceProviderCode:{serviceProviderCode} Msg: {r.Msg}");
}
}
}
private TrackUpdateInput BuiltUpdateItem(TrackPushContentConfigDto config, TrackPushRecordIndex log)
{
var trackDatetime = log.TrackTime.AddHours(config.TrackTimeZone ?? log.TrackTimeZone - log.TrackTimeZone);
var trackDateStr = DateTimeFormat(trackDatetime, config.TrackTimeZone ?? log.TrackTimeZone, config);
var body = new TrackUpdateInput
{
OrderNo = log.OrderNo
};
var dict = new Dictionary<string, object?>()
{
//推真实的发生地和发生时间(时间是计算后的)
{ "location", log.TrackPlace },
{ "date", trackDateStr },
{ "timeZone", config.TrackTimeZone ?? log.TrackTimeZone },
{ "description", config.TrackEnContentValue },
{ "descriptionCn", config.TrackContentValue },
{ "code", config.TrackContentCodeValue },
{ "status", config.TrackStatusValue },
{ "extra", TextJsonHelper.Deserialize<object>(log.Extra) }
};
body.Data.Add(dict);
log.PushContent = body.ToJson();
return body;
}
private async Task<TrackPushInput> BuiltPushItem(TrackPushContentConfigDto config, TrackPushRecordIndex log,
string? compareFields)
{
var trackDatetime = log.TrackTime.AddHours(config.TrackTimeZone ?? log.TrackTimeZone - log.TrackTimeZone);
var trackDateStr = DateTimeFormat(trackDatetime, config.TrackTimeZone ?? log.TrackTimeZone, config);
var body = new TrackPushInput
{
OrderNo = log.OrderNo,
};
var cur = new Dictionary<string, object?>
{
// 推真实的发生地和发生时间(时间是计算后的)
{ "location", log.TrackPlace },
{ "date", trackDateStr },
{ "timeZone", config.TrackTimeZone ?? log.TrackTimeZone },
{ "description", config.TrackEnContentValue },
{ "descriptionCn", config.TrackContentValue },
{ "code", config.TrackContentCodeValue },
{ "status", config.TrackStatusValue },
{ "extra", TextJsonHelper.Deserialize<object>(log.Extra) }
};
body.Data.Add(cur);
if (config.IsAllPush)
{
var history = await _trackPushRecordEsRepository.GetLatestTrackPushRecordAsync(log.OrderId, true);
if (history != default)
{
body = TextJsonHelper.Deserialize<TrackPushInput>(history.PushContent);
//在推送历史中,选出当前轨迹明细一致的轨迹
var tracks = await _trackDetailsEsRepository.GetListAsync(log.OrderId);
//确定轨迹标识符
var trackIdentifies = tracks.Select(x =>
{
//寻找映射配置(转换后的code)
var code = _pushConfigs.First(y => y.TrackCode == x.TrackCode).TrackContentCodeValue;
var timeStr = DateTimeFormat(x.TrackTime, config.TrackTimeZone ?? log.TrackTimeZone, config);
return string.IsNullOrWhiteSpace(compareFields)
? $"{code}{timeStr}"
: $"{code}{TextJsonHelper.ExtractValue(x.Extra, compareFields)}";
}
).ToImmutableHashSet();
//在历史中选取在订单当前轨迹明细中的记录
body!.Data = body.Data.Where(x =>
{
var code = ((JsonElement) x["code"]!).GetString();
var timeStr = ((JsonElement) x["date"]!).GetString();
if (string.IsNullOrWhiteSpace(compareFields))
return trackIdentifies.Contains($"{code}{timeStr}");
if (((JsonElement) x["extra"]!).TryGetProperty(compareFields, out var comparerValueJsonElement))
return trackIdentifies.Contains($"{code}{TextJsonHelper.GetString(comparerValueJsonElement)}");
return trackIdentifies.Contains($"{code}{null}");
}).ToList();
if (config.IsAllPushWithLatest)
{
//同code同比较字段保证唯一
var old = body.Data
.Where(x =>
{
//反序列化出的JsonElement和字典字符串类型做比较
var a = ((JsonElement) x["code"]!).GetString();
var b = cur["code"]!
.ToString(); //tips: 需要要将object类型的字符串转为string类型,否则使用的是引用比较,反序列化出的string一定和本地string引用不同
return a == b;
})
.WhereIf(!compareFields.IsNullOrWhiteSpace(), x =>
{
if (x["extra"] == null || log.Extra == null) return false;
return TextJsonHelper.EqByJsonElementField(x["extra"]!,
TextJsonHelper.Deserialize<JsonElement>(log.Extra), compareFields!);
})
.FirstOrDefault();
if (old != null)
{
body.Data.Remove(old);
}
}
// 添加当前要推送的轨迹
body.Data.Add(cur);
}
}
log.PushContent = TextJsonHelper.Serialize(body)!;
return body;
}
/// <summary>
/// 根据轨迹代码推断下一个(升序)要推送的内容,如果找不到或者已经是最后一个了
/// 则查询有没有可交换的推送,没有就返回null
/// </summary>
private static TrackPushContentConfigDto? GetValidPushConfigAsync(List<TrackPushContentConfigDto> cache,
string? latestPushTrackCode, TrackPushEvent @event)
{
var trackCode = @event.TrackCode;
var idx = cache.FindIndex(x => x.TrackCode == latestPushTrackCode);
if (@event.ForcedPush && trackCode == latestPushTrackCode)
{
//强制推送,允许重复推送当前轨迹
return cache[idx];
}
if (idx + 1 >= cache.Count) return default; // 推完了
// find next
var matched = cache[idx + 1].TrackCode == trackCode
? cache[idx + 1]
// find swap
: cache.FirstOrDefault(x => x.TrackCode == cache[idx + 1].SwapTrackCode);
//没推完,可能不符合推送顺序
return matched != null && matched.TrackCode == trackCode ? matched : default;
}
/// <summary>
/// 转换推送时间格式
/// </summary>
private static string DateTimeFormat(DateTime dateTime, double timezone, TrackPushContentConfigDto config)
{
if (config.DateTimeFormatType == (int) DateTimeFormatEnum.Timestamp)
return DateTimeHelper.ConvertToUnixTimestamp(dateTime).ToString();
if (config.DateTimeFormatType == (int) DateTimeFormatEnum.DateTimeAndTimezone
&& !string.IsNullOrEmpty(config.DateTimeFormat)
&& !string.IsNullOrEmpty(config.TimezoneFormat))
{
var zoneStr = DateTimeHelper.TimezoneFormat(timezone, config.TimezoneFormat);
var datetimeStr = DateTimeHelper.FormatDateTime(dateTime, config.DateTimeFormat);
return $"{datetimeStr} {zoneStr}";
}
return dateTime.ToString("yyyy-MM-ddTHH:mm:ss.fffK");
}
//轨迹推送失败(被拦截)
private void TrackPushFailedPublicTopic(string failedMessage, TrackPushEvent @event, OrderEntity order)
{
_broadcastingService.TrackPushFailedPubilcTopic(new TrackPushRecordDto
{
//还没请求接口通
RequestId = null,
IsSuccess = false,
Message = failedMessage,
OperationMan = @event.OperationMan,
PushTime = DateTime.Now,
OptionType = @event.OptionType.ToString(),
IsManualPush = @event.OperationMan != null,
//还没到构建推送内容那一步就被拦截
PushContent = string.Empty,
OrderNo = order.OrderNo,
TrackStatus = @event.SystemPushInfo.TrackStatus,
TrackContent = @event.SystemPushInfo.TrackStatus,
TrackTimeZone = @event.SystemPushInfo.TrackTimeZone,
TrackTime = @event.SystemPushInfo.TrackTime,
TrackPlace = @event.SystemPushInfo.TrackPlace,
TrackCode = @event.TrackCode,
Extra = @event.SystemPushInfo.Extra,
});
}
//轨迹推送失败(请求接口通)
private void TrackPushFailedPublicTopic(
string failedMessage,
TrackPushEvent @event,
Guid requestId,
List<dynamic> pushOrUpdateInputs)
{
foreach (var pushInput in pushOrUpdateInputs)
{
_broadcastingService.TrackPushFailedPubilcTopic(new TrackPushRecordDto
{
RequestId = requestId,
IsSuccess = false,
Message = failedMessage,
OperationMan = @event.OperationMan,
PushTime = DateTime.Now,
OptionType = @event.OptionType.ToString(),
IsManualPush = @event.OperationMan != null,
PushContent = TextJsonHelper.Serialize(pushInput),
OrderNo = pushInput.OrderNo,
TrackStatus = @event.SystemPushInfo.TrackStatus,
TrackContent = @event.SystemPushInfo.TrackStatus,
TrackTimeZone = @event.SystemPushInfo.TrackTimeZone,
TrackTime = @event.SystemPushInfo.TrackTime,
TrackPlace = @event.SystemPushInfo.TrackPlace,
TrackCode = @event.TrackCode,
Extra = @event.SystemPushInfo.Extra,
});
}
}
}