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

559 lines
24 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.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.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 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)
{
_httpClientProxy = httpClientProxy;
_orderRepository = orderRepository;
_trackPushCacheService = trackPushCacheService;
_trackPushRecordEsRepository = trackPushRecordEsRepository;
_options = options;
_broadcastingService = broadcastingService;
}
[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>();
TrackPushContentConfigDto? config;
var isSystemPush = @event.OperationMan == null;
if (isSystemPush && !@event.ForcedPush)
{
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;
}
// 推送新增,需要根据上一次的推送 推断 本次推送 是否可行,为了约束顺序
var (resultConfig, nextCode) = GetValidPushConfigAsync(_pushConfigs, order.LatestPushTrackCode, @event.TrackCode);
if (resultConfig == null)
{
TrackPushFailedPublicTopic($"推送顺序异常,下一个要推送的应为[{nextCode}]", @event, order);
continue;
}
config = resultConfig;
// 是否自动推送
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);
}
}
// 手工或强制推送,不会校验已推和顺序问题,因为可能是更新
else
{
config = _pushConfigs.FirstOrDefault(x => x.TrackCode == @event.TrackCode);
foreach (var order in _orders)
{
if (config == null)
{
TrackPushFailedPublicTopic($"没有在推送映射配置中找到关于轨迹代码[{@event.TrackCode}]的配置", @event, order);
return;
}
if (config is { IsAllPush: true })
{
// 非强制推送需要 推断 本次推送 是否可行
if (!@event.ForcedPush)
{
var (resultConfig, nextCode) = GetValidPushConfigAsync(_pushConfigs, order.LatestPushTrackCode, @event.TrackCode);
if (resultConfig == null)
{
TrackPushFailedPublicTopic($"推送顺序异常,下一个要推送的应为[{nextCode}]", @event, order);
continue;
}
config = resultConfig;
}
// tips:强制推送直接使用配置
}
//配置的类型如何和订单类型不一致则不允许推送
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);
}
}
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 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);
//找到历史中推送成功的轨迹
var trackPushRecords = await _trackPushRecordEsRepository.GetHistorySuccessTrackPushRecordAsync(log.OrderId);
//构建标识符,code+比较字段的值(如果提供的比较字段,否则使用轨迹时间)
var pushIdentitySuccess = trackPushRecords.Select(x =>
{
//寻找映射配置
var code = _pushConfigs.First(y => y.TrackCode == x.TrackCode).TrackContentCodeValue;
if (string.IsNullOrWhiteSpace(compareFields))
return code;
//尝试从扩展中提取目标值
var extraValue = TextJsonHelper.ExtractValue(x.Extra, compareFields);
return code + extraValue;
}).ToHashSet();
if (history != default)
{
body = TextJsonHelper.Deserialize<TrackPushInput>(history.PushContent);
//todo-test:在历史选出推送成功的历史轨迹
body!.Data = body.Data.Where(x =>
{
var code = ((JsonElement) x["code"]!).GetString();
if (string.IsNullOrWhiteSpace(compareFields))
return pushIdentitySuccess.Contains(code);
var isMatch = ((JsonElement) x["extra"]!).TryGetProperty(compareFields, out var jsonElement);
var extraValue = isMatch ? TextJsonHelper.GetString(jsonElement) : null;
return pushIdentitySuccess.Contains(code + extraValue);
}).ToList();
//添加当前要推送的轨迹
body.Data.Add(cur);
if (config.IsAllPushWithLatest)
{
//历史轨迹中同一个code+比较字段联合作为key最新的一个
var dict = new Dictionary<string, (string dateStr, object dict)>();
//todo-test:根据code和比较字段值分组,在每组里选出一个创建时间最新的
foreach (var x in body.Data)
{
var code = ((JsonElement) x["code"]!).GetString()!;
var dateStr = ((JsonElement) x["date"]!).GetString()!;
var key = string.Empty;
if (string.IsNullOrWhiteSpace(compareFields))
key = code;
else
{
var isMatch = ((JsonElement) x["extra"]!).TryGetProperty(compareFields, out var jsonElement);
var extraValue = isMatch ? TextJsonHelper.GetString(jsonElement) : null;
key = code + extraValue;
}
//只保留最新的一条数据
if (dict.TryGetValue(key, out var tuple))
{
//如果当前值创建时间大于历史值,则替换历史值
if (DateTime.Parse(dateStr) > DateTime.Parse(tuple.dateStr))
dict[key] = (dateStr, x);
}
else
dict[key] = (dateStr, x);
}
// //同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 = dict.Values.Select(x => x.dict).Cast<Dictionary<string, object?>>().ToList();
}
// //添加当前要推送的轨迹
// body.Data.Add(cur);
}
//todo-test:按推送映射顺序排序
}
log.PushContent = TextJsonHelper.Serialize(body)!;
return body;
}
/// <summary>
/// 根据轨迹代码推断下一个(升序)要推送的内容,如果找不到或者已经是最后一个了
/// 则查询有没有可交换的推送,没有就返回null
/// </summary>
private static (TrackPushContentConfigDto? config, string nextCode) GetValidPushConfigAsync(List<TrackPushContentConfigDto> cache,
string? latestPushTrackCode, string? trackCode)
{
var idx = cache.FindIndex(x => x.TrackCode == latestPushTrackCode);
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, matched.TrackCode)
: (default, matched?.TrackCode ?? cache[idx + 1].TrackCode);
}
/// <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,
});
}
}
}