ideship/track
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

optimize&fix：force push event case

Browse Source
This commit is contained in:
xiaolipro 2025-01-15 15:02:03 +08:00
parent 0b6e20da3f
commit 08f0ebeeae
1 changed files with 153 additions and 138 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

237
src/SJZY.Track.Application/CAP/EventHandlers/TrackPushEventHandler.cs
View File

@ -96,107 +96,72 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe
private async Task DoPushAsync(TrackPushEvent @event)
{
var items = new List<TrackPushInput>();
TrackPushContentConfigDto? config;
var isSystemPush = @event.OperationMan == null;
if (isSystemPush && !@event.ForcedPush)
if (@event.OperationMan == null)
{
config = _pushConfigs.FirstOrDefault(x => x.TrackCode == @event.TrackCode);
foreach (var order in _orders)
if (@event.OptionType == OptionType.新增)
{
if (config == null)
{
TrackPushFailedPublicTopic($"没有在推送映射配置中找到关于轨迹代码[{@event.TrackCode}]的配置", @event, order);
return;
await BuiltSystemPushItemsAsync(@event, items);
}
// 系统不推异常单，只能人工操作
if (order.QuestionType == (int) TrackProblemType.Error)
else if (@event.OptionType == OptionType.修改)
{
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);
await BuiltManualPushItemsAsync(@event, items);
}
}
// 手工或强制推送，不会校验已推和顺序问题，因为可能是更新
else
{
config = _pushConfigs.FirstOrDefault(x => x.TrackCode == @event.TrackCode);
foreach (var order in _orders)
{
if (config == null)
{
TrackPushFailedPublicTopic($"没有在推送映射配置中找到关于轨迹代码[{@event.TrackCode}]的配置", @event, order);
return;
// 手工推送，不会校验已推和顺序问题，因为可能是更新
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)
{
var (resultConfig, nextCode) = GetValidPushConfigAsync(_pushConfigs, order.LatestPushTrackCode, @event.TrackCode);
if (resultConfig == null)
config = GetValidPushConfigAsync(_pushConfigs, order.LatestPushTrackCode, @event);
}
}
if (config == null)
{
TrackPushFailedPublicTopic($"推送顺序异常，下一个要推送的应为[{nextCode}]", @event, order);
continue;
}
config = resultConfig;
}
// tips:强制推送直接使用配置
TrackPushFailedPublicTopic("推送顺序异常，找不到有效推送配置", @event, order);
return;
}
//配置的类型如何和订单类型不一致则不允许推送
if (config.OrderType != order.OrderType)
{
TrackPushFailedPublicTopic($"订单类型{order.OrderType}和配置中的订单类型[{config.OrderType}]不一致，不能推送", @event, order);
TrackPushFailedPublicTopic($"订单类型{order.OrderType}和配置中的订单类型[{config.OrderType}]不一致，不能推送", @event,
order);
continue;
}
@ -222,23 +187,66 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe
}
}
if (items.Count == 0) return;
using (Track2InterfaceRelaySubSystemContext.Use(
new Track2InterfaceRelaySubSystemContext(@event.SubSystemCode)))
private async Task BuiltSystemPushItemsAsync(TrackPushEvent @event, List<TrackPushInput> items)
{
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)
var config = _pushConfigs.FirstOrDefault(x => x.TrackCode == @event.TrackCode);
foreach (var order in _orders)
{
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}");
if (config == null)
{
TrackPushFailedPublicTopic($"没有在推送映射配置中找到关于轨迹代码[{@event.TrackCode}]的配置", @event, order);
return;
}
// 系统不推异常单，只能人工操作
if (order.QuestionType == (int) TrackProblemType.Error)
{
TrackPushFailedPublicTopic("系统不推异常单，只能人工操作", @event, order);
continue;
}
// 推送新增，需要根据上一次的推送 推断 本次推送 是否可行，为了约束顺序
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);
}
}
@ -253,7 +261,8 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe
//配置的类型如何和订单类型不一致则不允许推送
if (config.OrderType != order.OrderType)
{
TrackPushFailedPublicTopic($"订单类型{order.OrderType}和配置中的订单类型[{config.OrderType}]不一致，不能推送", @event, order);
TrackPushFailedPublicTopic($"订单类型{order.OrderType}和配置中的订单类型[{config.OrderType}]不一致，不能推送", @event,
order);
continue;
}
@ -292,8 +301,10 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe
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}");
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}");
}
}
}
@ -363,12 +374,15 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe
{
//反序列化出的JsonElement和字典字符串类型做比较
var a = ((JsonElement) x["code"]!).GetString();
var b = cur["code"]!.ToString(); //tips: 需要要将object类型的字符串转为string类型，否则使用的是引用比较，反序列化出的string一定和本地string引用不同
var b = cur["code"]!
.ToString(); //tips: 需要要将object类型的字符串转为string类型，否则使用的是引用比较，反序列化出的string一定和本地string引用不同
return a == b;
}).WhereIf(!compareFields.IsNullOrWhiteSpace(), x =>
})
.WhereIf(!compareFields.IsNullOrWhiteSpace(), x =>
{
if (x["extra"] == null || log.Extra == null) return false;
return TextJsonHelper.EqByJsonElementField(x["extra"]!, TextJsonHelper.Deserialize<JsonElement>(log.Extra), compareFields!);
return TextJsonHelper.EqByJsonElementField(x["extra"]!,
TextJsonHelper.Deserialize<JsonElement>(log.Extra), compareFields!);
})
.FirstOrDefault();
@ -376,8 +390,8 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe
{
body.Data.Remove(old);
}
}
// //添加当前要推送的轨迹
body!.Data.Add(cur);
}
@ -392,24 +406,28 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe
/// 根据轨迹代码推断下一个（升序）要推送的内容，如果找不到或者已经是最后一个了
/// 则查询有没有可交换的推送，没有就返回null
/// </summary>
private static (TrackPushContentConfigDto? config, string nextCode) GetValidPushConfigAsync(List<TrackPushContentConfigDto> cache,
string? latestPushTrackCode, string? trackCode)
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);
: 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);
return matched != null && matched.TrackCode == trackCode ? matched : default;
}
/// <summary>
@ -433,10 +451,7 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe
}
//轨迹推送失败(被拦截)
private void TrackPushFailedPublicTopic(
string failedMessage,
TrackPushEvent @event,
OrderEntity order)
private void TrackPushFailedPublicTopic(string failedMessage, TrackPushEvent @event, OrderEntity order)
{
_broadcastingService.TrackPushFailedPubilcTopic(new TrackPushRecordDto
{