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

temp

Browse Source
This commit is contained in:
李鹏军 2024-04-24 17:49:31 +08:00
parent 9bf4150990
commit 5ab91c3d8c
24 changed files with 436 additions and 319 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
src/SJZY.Track.Application.Contract/Dtos/TrackPush/TrackPushInput.cs
View File

@ -5,6 +5,17 @@ public class TrackPushInput
public string OrderNo { get; set; } = null!;
public List<TrackItem> Data { get; set; } = new();
public string Log { get; set; }
}
public class TrackUpdateInput
{
public string OrderNo { get; set; } = null!;
public List<TrackItem> Data { get; set; } = new();
public string Log { get; set; }
}
public class TrackItem

2
src/SJZY.Track.Application.Contract/Events/OrderIdentifier.cs
View File

@ -5,4 +5,6 @@ public struct OrderIdentifier
public int OrderId { get; set; }
public string OrderNo { get; set; }
public int OrderType { get; set; }
public int CustomId { get; set; }
}

54
src/SJZY.Track.Application.Contract/Events/TrackPushEvent.cs
View File

@ -2,45 +2,58 @@
public class TrackPushEvent
{
/// <summary>
/// 重试次数
/// </summary>
public int RetryNum { get; set; }
/// <summary>
/// 操作类型
/// </summary>
public OptionType OptionType { get; set; }
/// <summary>
/// 子系统id
/// </summary>
public int SubSystemId { get; set; }
/// <summary>
/// 客户id
/// </summary>
public int CustomId { get; set; }
/// <summary>
/// 订单id
/// </summary>
public int OrderId { get; set; }
/// <summary>
/// 订单号
/// </summary>
public string OrderNo { get; set; } = null!;
/// <summary>
/// 轨迹状态
/// </summary>
public string TrackStatus { get; set; } = null!;
public int[] OrderIds { get; set; } = null!;
/// <summary>
/// 轨迹代码（我们系统的）
/// </summary>
public string TrackCode { get; set; } = null!;
/// <summary>
/// 操作人
/// </summary>
public string? OperationMan { get; set; }
/// <summary>
/// 系统推送信息
/// </summary>
public SystemPushStruct SystemPushInfo { get; set; }
}
public struct SystemPushStruct
{
/// <summary>
/// 轨迹状态
/// </summary>
public string TrackStatus { get; set; }
/// <summary>
/// 轨迹内容
/// </summary>
public string TrackContent { get; set; } = null!;
public string TrackContent { get; set; }
/// <summary>
/// 轨迹发生地
/// </summary>
public string TrackPlace { get; set; } = null!;
public string TrackPlace { get; set; }
/// <summary>
/// 轨迹发生时间
@ -51,9 +64,4 @@ public class TrackPushEvent
/// 轨迹发生时区
/// </summary>
public float TrackTimeZone { get; set; }
/// <summary>
/// 操作人
/// </summary>
public string? OperationMan { get; set; }
}

4
src/SJZY.Track.Application.Contract/Events/TrackUpsertEvent.cs
View File

@ -18,9 +18,9 @@ public class TrackUpsertEvent
public string? OperationMan { get; set; }
/// <summary>
/// true新增，false更新，null表示刷新
/// true新增，false更新
/// </summary>
public bool? IsAdd { get; set; }
public bool IsAdd { get; set; }
/// <summary>
/// 轨迹内容代码

22
src/SJZY.Track.Application.Contract/Services/ITrackPushCacheService.cs
View File

@ -19,26 +19,4 @@ public interface ITrackPushCacheService
/// <param name="customId"></param>
/// <returns></returns>
Task<List<TrackPushContentConfigDto>?> SetPushContentConfigsAsync(int subSystemId, int customId);
/// <summary>
/// 获取服务商拉取配置
/// </summary>
/// <param name="subSystemId">子系统id</param>
/// <param name="customId">服务商id</param>
/// <param name="trackConde">轨迹代码</param>
/// <returns></returns>
Task<TrackPushContentConfigDto?> GetPushContentConfigAsync(int subSystemId, int customId, string trackConde);
/// <summary>
/// 根据轨迹代码推断下一个（升序）要推送的内容，如果找不到或者已经是最后一个了
/// 则查询有没有可交换的推送，没有就返回null
/// </summary>
/// <param name="subSystemId"></param>
/// <param name="customId"></param>
/// <param name="latestPushTrackCode"></param>
/// <param name="trackCode"></param>
/// <returns></returns>
Task<TrackPushContentConfigDto?> IsValidPushAsync(int subSystemId, int customId,
string? latestPushTrackCode,
string? trackCode);
}

2
src/SJZY.Track.Application/EventHandlers/TrackDeleteEventHandler.cs
View File

@ -32,7 +32,7 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe
var dup = @event.Visited.Intersect(@event.OrderIds).ToArray();
if (dup.Any())
{
LogHelper.Error($"删除轨迹订单时，发现循环依赖：{dup.JoinAsString(",")}");
LogHelper.Error($"删除轨迹时，发现循环依赖：{dup.JoinAsString(",")}");
return;
}

85
src/SJZY.Track.Application/EventHandlers/TrackPushCompletedEventHandler.cs Normal file
View File

@ -0,0 +1,85 @@
using DotNetCore.CAP;
using Newtonsoft.Json;
using SJZY.InterfaceRelay.Application.Contracts;
using SJZY.Track.Domain.Aggregates.OrderAggregate;
using SJZY.Track.Domain.Aggregates.TrackPushAggregate;
using SJZY.Track.Domain.Consts;
using SJZY.Track.Domain.Enums;
using Volo.Abp.DependencyInjection;
namespace SJZY.Track.Application.EventHandlers;
public class TrackPushCompletedEventHandler : ITransientDependency, ICapSubscribe
{
private readonly ITrackPushRecordsRepository _trackPushRecordsRepository;
private readonly IOrderRepository _orderRepository;
public TrackPushCompletedEventHandler(ITrackPushRecordsRepository trackPushRecordsRepository,
IOrderRepository orderRepository)
{
_trackPushRecordsRepository = trackPushRecordsRepository;
_orderRepository = orderRepository;
}
[CapSubscribe(GlobalConst.TrackPushCompletedSubscribe, Group = GlobalConst.TrackPushSubscribe)]
public async Task ConsumerAsync(InterfaceRelayForwardReplayEvent @event)
{
var log = JsonConvert.DeserializeObject<TrackPushRecordIndex>(@event.RequestParams);
if (log == null) throw new BusinessException(message: "轨迹推送返回结果解析异常");
log.IsSuccess = @event.IsSuccess;
log.Message = @event.Message;
log.RequestId = @event.RequestId;
// 往es 写推送日志
await _trackPushRecordsRepository.AddAsync(log);
// if (!log.IsSuccess)
// {
// // 重试过程中不会去更新订单主表
// if (@event.RetryNum < 3)
// {
// @event.RetryNum++;
// await _capPublisher.PublishAsync(GlobalConst.TrackPushSubscribe, @event);
// return;
// }
// }
// 同步去更新订单主表
await TrackPushAfterUpdateOrderAsync(log);
}
private async Task TrackPushAfterUpdateOrderAsync(TrackPushRecordIndex log)
{
var order = new OrderEntity();
order.Id = log.OrderId;
/*
* 异常：最近一条推送是失败
*/
if (!log.IsSuccess)
{
order.QuestionType = (int)TrackProblemType.Error;
order.LatestPushError = log.Message;
}
else
{
order.QuestionType = (int)TrackProblemType.None;
order.LatestPushError = null;
if (log.OptionType == OptionType.新增.ToString())
{
// tips：只有新增会影响推送进度？
order.LatestPushTrackCode = log.TrackCode;
order.LatestPushTime = DateTime.Now;
await _orderRepository.UpdateColumnAsync(order, x => new
{
x.LatestPushTrackCode, x.LatestPushTime, x.QuestionType, x.LatestPushError
});
return;
}
}
await _orderRepository.UpdateColumnAsync(order, x => new
{
x.QuestionType, x.LatestPushError
});
}
}

334
src/SJZY.Track.Application/EventHandlers/TrackPushEventHandler.cs
View File

@ -6,6 +6,7 @@ using SJZY.Track.Application.Contract.Dtos.TrackPush;
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.Aggregates.TrackPushAggregate;
using SJZY.Track.Domain.Consts;
using SJZY.Track.Domain.Enums;
@ -17,212 +18,176 @@ namespace SJZY.Track.Application.EventHandlers;
public class TrackPushEventHandler : ITransientDependency, ICapSubscribe
{
private readonly IHttpClientProxy<IInterfaceForwardServiceV2> _httpClientProxy;
private readonly ITrackPushRecordsRepository _trackPushRecordsRepository;
private readonly IOrderRepository _orderRepository;
private readonly ITrackPushCacheService _trackPushCacheService;
private readonly ICapPublisher _capPublisher;
private readonly ITrackDetailsRepository _trackDetailsRepository;
private readonly ITrackPushRecordsRepository _trackPushRecordsRepository;
private List<TrackPushContentConfigDto> _pushConfigs;
private List<OrderEntity> _orders;
public TrackPushEventHandler(IHttpClientProxy<IInterfaceForwardServiceV2> httpClientProxy,
ITrackPushRecordsRepository trackPushRecordsRepository,
IOrderRepository orderRepository, ITrackPushCacheService trackPushCacheService, ICapPublisher capPublisher)
IOrderRepository orderRepository, ITrackPushCacheService trackPushCacheService,
ITrackDetailsRepository trackDetailsRepository, ITrackPushRecordsRepository trackPushRecordsRepository)
{
_httpClientProxy = httpClientProxy;
_trackPushRecordsRepository = trackPushRecordsRepository;
_orderRepository = orderRepository;
_trackPushCacheService = trackPushCacheService;
_capPublisher = capPublisher;
_trackDetailsRepository = trackDetailsRepository;
_trackPushRecordsRepository = trackPushRecordsRepository;
}
[CapSubscribe(GlobalConst.TrackPushSubscribe, Group = GlobalConst.TrackPushSubscribe)]
public async Task ConsumerAsync(TrackPushEvent @event)
{
var order = await _orderRepository.GetFirstAsync(x =>
x.Id == @event.OrderId, x => new
var pushConfigs = await _trackPushCacheService.GetPushContentConfigsAsync(@event.SubSystemId, @event.CustomId);
if (pushConfigs == null || pushConfigs.Count == 0) return;
_pushConfigs = pushConfigs;
var errorType = (int)TrackProblemType.Error;
var orders = await _orderRepository.GetListAsync(x =>
@event.OrderIds.Contains(x.Id) && x.QuestionType != errorType, x => new OrderEntity
{
x.SubSystemId, x.LatestPushTrackCode, x.QuestionType, x.CustomId, x.OrderType
Id = x.Id, OrderNo = x.OrderNo, LatestPushTrackCode = x.LatestPushTrackCode,
OrderType = x.OrderType
});
if (order == null) return;
var log = new TrackPushRecordIndex
{
IndexName = $"{GlobalConst.TrackPushRecordsIndex}-{DateTime.Now:yyyy.MM.dd}",
OptionType = @event.OptionType.ToString(),
OrderId = @event.OrderId,
TrackStatus = @event.TrackStatus,
TrackCode = @event.TrackCode,
TrackContent = @event.TrackContent,
TrackPlace = @event.TrackPlace,
TrackTime = @event.TrackTime,
TrackTimeZone = @event.TrackTimeZone,
CreateUser = @event.OperationMan ?? GlobalConst.SystemUser,
CreateTime = DateTime.Now
};
TrackPushContentConfigDto? config = null;
if (orders.Count == 0) return;
_orders = orders;
switch (@event.OptionType)
{
case OptionType.无:
return;
case OptionType.新增:
if (@event.OperationMan == null)
{
if (order.QuestionType != (int)TrackProblemType.Error)
{
// 推送新增，需要根据上一次的推送 推断 本次推送 是否可行
config = await _trackPushCacheService.IsValidPushAsync(order.SubSystemId, order.CustomId,
order.LatestPushTrackCode, @event.TrackCode);
if (config == null) return;
if (!config.IsAutoPush) return;
if (config.OrderType != order.OrderType) return;
}
}
else
{
// 手工推送单条不推断，直接无脑推
config = await _trackPushCacheService.GetPushContentConfigAsync(order.SubSystemId, order.CustomId,
@event.TrackCode);
if (config is { IsAllPush: true })
{
// 推断 本次推送 是否可行
config = await _trackPushCacheService.IsValidPushAsync(order.SubSystemId, order.CustomId,
order.LatestPushTrackCode, @event.TrackCode);
}
}
if (config == null) return;
await DoAddAsync(@event, config, log);
await DoPushAsync(@event);
break;
case OptionType.删除:
break;
case OptionType.修改:
config = await _trackPushCacheService.GetPushContentConfigAsync(order.SubSystemId, order.CustomId,
@event.TrackCode);
if (config == null || !config.IsCanUpdate) return;
await DoUpdateAsync(@event, config, log);
await DoUpdateAsync(@event);
break;
default:
throw new ArgumentOutOfRangeException();
}
// 往es 写推送日志
await _trackPushRecordsRepository.AddAsync(log);
if (!log.IsSuccess)
{
// 重试过程中不会去更新订单主表
if (@event.RetryNum < 3)
{
@event.RetryNum++;
await _capPublisher.PublishAsync(GlobalConst.TrackPushSubscribe, @event);
return;
}
}
// 同步去更新订单主表
await TrackPushAfterUpdateOrderAsync(new TrackPushAfterUpdateOrderDto
{
OptionType = @event.OptionType,
OrderId = @event.OrderId,
TrackCode = @event.TrackCode,
IsSuccess = log.IsSuccess,
Message = log.Message
});
}
private async Task TrackPushAfterUpdateOrderAsync(TrackPushAfterUpdateOrderDto dto)
private async Task DoPushAsync(TrackPushEvent @event)
{
var order = new OrderEntity();
order.Id = dto.OrderId;
/*
* 异常：最近一条推送是失败
*/
if (!dto.IsSuccess)
var logs = new List<TrackPushRecordIndex>();
TrackPushContentConfigDto? config;
if (@event.OperationMan == null)
{
order.QuestionType = (int)TrackProblemType.Error;
order.LatestPushError = dto.Message;
}
else
{
order.QuestionType = (int)TrackProblemType.None;
order.LatestPushError = null;
if (dto.OptionType == OptionType.新增)
foreach (var order in _orders)
{
// tips：只有新增会影响推送进度？
order.LatestPushTrackCode = dto.TrackCode;
order.LatestPushTime = DateTime.Now;
await _orderRepository.UpdateColumnAsync(order, x => new
// 推送新增，需要根据上一次的推送 推断 本次推送 是否可行
config = GetValidPushConfigAsync(_pushConfigs, order.LatestPushTrackCode, @event.TrackCode);
if (config == null) continue;
if (!config.IsAutoPush) continue;
if (config.OrderType != order.OrderType) continue;
var log = new TrackPushRecordIndex
{
x.LatestPushTrackCode, x.LatestPushTime, x.QuestionType, x.LatestPushError
});
return;
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,
};
await BuiltPushItem(config, log);
logs.Add(log);
}
}
await _orderRepository.UpdateColumnAsync(order, x => new
else //手工推送
{
x.QuestionType, x.LatestPushError
});
}
config = _pushConfigs.FirstOrDefault(x => x.TrackCode == @event.TrackCode);
if (config == null) return;
private async Task DoAddAsync(TrackPushEvent @event,
TrackPushContentConfigDto config, TrackPushRecordIndex log)
{
var track = new TrackItem
{
// 推真实的发生地和发生时间（时间是计算后的）
Location = @event.TrackPlace,
Date = @event.TrackTime.AddHours(
config.TrackTimeZone ?? @event.TrackTimeZone - @event.TrackTimeZone),
TimeZone = config.TrackTimeZone ?? @event.TrackTimeZone,
Description = config.TrackEnContentValue,
DescriptionCn = config.TrackContentValue,
Code = config.TrackContentCodeValue,
Status = config.TrackStatusValue,
Extra = config.ExpandFieldsValue
};
var tracks = await _trackDetailsRepository.GetListAsync(@event.OrderIds, @event.TrackCode);
tracks = tracks.DistinctBy(x => x.OrderId).ToList(); // 防止重复推送
var body = new TrackPushInput
{
OrderNo = @event.OrderNo,
};
body.Data.Add(track);
if (tracks.Count == 0) return; // 一个能推的轨迹都不存在
if (config.IsAllPush)
{
var history = await _trackPushRecordsRepository.GetLatestTrackPushRecordAsync(@event.OrderId);
if (history != default)
foreach (var track in tracks)
{
body = JsonConvert.DeserializeObject<TrackPushInput>(history.PushContent);
body!.Data.Add(track);
var order = _orders.First(x => x.Id == track.OrderId);
if (config is { IsAllPush: true })
{
// 推断 本次推送 是否可行
config = GetValidPushConfigAsync(_pushConfigs, order.LatestPushTrackCode, @event.TrackCode);
}
if (config == null) continue;
var log = new TrackPushRecordIndex
{
OptionType = @event.OptionType.ToString(),
OrderId = track.OrderId,
OrderNo = order.OrderNo,
TrackStatus = track.TrackStatus,
TrackCode = @event.TrackCode,
TrackContent = track.TrackContent,
TrackPlace = track.TrackPlace,
TrackTime = track.TrackTime,
TrackTimeZone = track.TrackTimeZone,
CreateUser = @event.OperationMan,
};
await BuiltPushItem(config, log);
logs.Add(log);
}
}
var res = await _httpClientProxy.Service.ForwardWithSubscribeAsync(InterfaceRelayConst.PushTrackInterfaceCode,
config.ServiceProviderCode, body);
log.PushContent = body.ToJson();
log.IsSuccess = res.Code == InterfaceRelayConst.SuccessCode;
log.RequestId = res.RequestId;
log.Message = res.Msg;
await _httpClientProxy.Service.BatchForwardThenPushToQueueAsync(InterfaceRelayConst.PushTrackInterfaceCode,
_pushConfigs.First().ServiceProviderCode, logs, GlobalConst.TrackPushCompletedSubscribe);
}
private async Task DoUpdateAsync(TrackPushEvent @event,
TrackPushContentConfigDto config, TrackPushRecordIndex log)
private async Task DoUpdateAsync(TrackPushEvent @event)
{
var config = _pushConfigs.FirstOrDefault(x => x.TrackCode == @event.TrackCode);
if (config == null || !config.IsCanUpdate) return;
var logs = new List<TrackPushRecordIndex>();
foreach (var order in _orders)
{
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,
};
BuiltUpdateItem(config, log);
logs.Add(log);
}
await _httpClientProxy.Service.BatchForwardThenPushToQueueAsync(InterfaceRelayConst.TrackUpdateInterfaceCode,
_pushConfigs.First().ServiceProviderCode, logs, GlobalConst.TrackPushCompletedSubscribe);
}
private void BuiltUpdateItem(TrackPushContentConfigDto config, TrackPushRecordIndex log)
{
var data = new
{
orderNo = @event.OrderNo,
orderNo = log.OrderNo,
data = new[]
{
new
{
// 推真实的发生地和发生时间
location = @event.TrackPlace,
date = @event.TrackTime.AddHours(
config.TrackTimeZone ?? @event.TrackTimeZone - @event.TrackTimeZone),
timeZone = config.TrackTimeZone ?? @event.TrackTimeZone,
location = log.TrackPlace,
date = log.TrackTime.AddHours(
config.TrackTimeZone ?? log.TrackTimeZone - log.TrackTimeZone),
timeZone = config.TrackTimeZone ?? log.TrackTimeZone,
description = config.TrackEnContentValue,
description_cn = config.TrackContentValue,
code = config.TrackContentCodeValue,
@ -232,13 +197,66 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe
}
};
var res = await _httpClientProxy.Service.ForwardWithSubscribeAsync(InterfaceRelayConst.TrackUpdateInterfaceCode,
config.ServiceProviderCode, data);
log.PushContent = data.ToJson();
log.IsSuccess = res.Code == InterfaceRelayConst.SuccessCode;
log.RequestId = res.RequestId;
log.Message = res.Msg;
}
private async Task BuiltPushItem(TrackPushContentConfigDto config, TrackPushRecordIndex log)
{
var track = new TrackItem
{
// 推真实的发生地和发生时间（时间是计算后的）
Location = log.TrackPlace,
Date = log.TrackTime.AddHours(
config.TrackTimeZone ?? log.TrackTimeZone - log.TrackTimeZone),
TimeZone = config.TrackTimeZone ?? log.TrackTimeZone,
Description = config.TrackEnContentValue,
DescriptionCn = config.TrackContentValue,
Code = config.TrackContentCodeValue,
Status = config.TrackStatusValue,
Extra = config.ExpandFieldsValue
};
var body = new TrackPushInput
{
OrderNo = log.OrderNo,
};
body.Data.Add(track);
if (config.IsAllPush)
{
var history = await _trackPushRecordsRepository.GetLatestTrackPushRecordAsync(log.OrderId);
if (history != default)
{
body = JsonConvert.DeserializeObject<TrackPushInput>(history.PushContent);
body!.Data.Add(track);
}
}
log.PushContent = body.ToJson();
}
/// <summary>
/// 根据轨迹代码推断下一个（升序）要推送的内容，如果找不到或者已经是最后一个了
/// 则查询有没有可交换的推送，没有就返回null
/// </summary>
/// <param name="cache"></param>
/// <param name="latestPushTrackCode"></param>
/// <param name="trackCode"></param>
/// <returns></returns>
TrackPushContentConfigDto? 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 : default;
}
}

12
src/SJZY.Track.Application/EventHandlers/TrackRefreshEventHandler.cs
View File

@ -36,6 +36,7 @@ public class TrackRefreshEventHandler : ITransientDependency, ICapSubscribe
[CapSubscribe(GlobalConst.TrackRefreshSubscribe, Group = GlobalConst.TrackRefreshSubscribe)]
public async Task ConsumerAsync(InterfaceRelayForwardReplayEvent @event)
{
if (!@event.IsSuccess) return;
var attach = Base64Helper.DeSerialize<TrackRefreshAttach>(@event.Attach);
if (attach is null) return;
@ -222,15 +223,16 @@ public class TrackRefreshEventHandler : ITransientDependency, ICapSubscribe
// 新增/更新轨迹
var trackUpsertEvent = new TrackUpsertEvent
{
OrderIdentifiers = new[]
{
OrderIdentifiers =
[
new OrderIdentifier
{
OrderId = context.OrderId,
OrderNo = context.OrderNo,
OrderType = context.OrderType
OrderType = context.OrderType,
CustomId = context.CustomId
}
},
],
TrackRefreshData = new TrackRefreshStruct
{
ServiceProviderName = context.ServiceProviderName,
@ -247,6 +249,8 @@ public class TrackRefreshEventHandler : ITransientDependency, ICapSubscribe
TrackTimeZone = trackTimeZone ?? matchedContentConfig.TrackTimeZoneValue ?? 8
};
trackUpsertEvent.IsAdd =
!await _trackDetailsRepository.AnyByCodeAsync(context.OrderId, matchedContentConfig.TrackContentCode);
await _capPublisher.PublishAsync(GlobalConst.TrackUpsertForRefreshSubscribe, trackUpsertEvent);
}
}

54
src/SJZY.Track.Application/EventHandlers/TrackUpsertEventHandler.cs
View File

@ -35,7 +35,7 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe
var dup = @event.Visited.Intersect(orderIds).ToArray();
if (dup.Any())
{
LogHelper.Error($"删除轨迹订单时，发现循环依赖：{dup.JoinAsString(",")}");
LogHelper.Error($"{(@event.IsAdd ? "添加" : "修改")}轨迹时，发现循环依赖：{dup.JoinAsString(",")}");
return;
}
@ -48,11 +48,11 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe
{
// 理论上小包在这不会分组
foreach (var group in linkedOrders.Where(x => x.OwnerOrderId == order.OrderId)
.GroupBy(x => x.SubSystemId))
.GroupBy(x => new { x.SubSystemId, x.CustomId }))
{
var nextEvent = @event.DeepCopy();
nextEvent.Visited.AddRange(orderIds);
nextEvent.SubSystemId = group.Key;
nextEvent.SubSystemId = group.Key.SubSystemId;
nextEvent.TrackRefreshData = new TrackRefreshStruct
{
ServiceProviderName = @event.TrackRefreshData.ServiceProviderName,
@ -82,7 +82,8 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe
{
OrderId = x.OrderId,
OrderNo = x.OrderNo,
OrderType = x.OrderType
OrderType = x.OrderType,
CustomId = x.CustomId
}).ToList();
var idx = 0;
while (idx < identifiers.Count)
@ -113,11 +114,8 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe
{
var track = tracks.FirstOrDefault(x => x.OrderId == order.OrderId);
if (@event.IsAdd != null)
{
if (track == null && !@event.IsAdd!.Value) continue; // 不存在不让改
if (track != null && @event.IsAdd!.Value) continue; // 存在不让创建
}
if (track == null && !@event.IsAdd) continue; // 不存在不让改
if (track != null && @event.IsAdd) continue; // 存在不让创建
if (track == default)
{
@ -167,22 +165,6 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe
await _trackDetailsRepository.UpdateAsync(track);
}
// step2、触发轨迹推送事件（异步）
var trackPushEvent = new TrackPushEvent
{
OptionType = track.Id.IsNullOrWhiteSpace() ? OptionType.新增 : OptionType.修改,
OrderId = order.OrderId,
OrderNo = order.OrderNo,
TrackStatus = track.TrackStatus,
TrackCode = track.TrackCode,
TrackContent = track.TrackContent,
TrackPlace = track.TrackPlace,
TrackTime = track.TrackTime,
TrackTimeZone = track.TrackTimeZone,
};
await _capPublisher.PublishAsync(GlobalConst.TrackPushSubscribe, trackPushEvent);
canNextOrderIds.Add(order.OrderId);
firstTrack ??= track;
@ -205,5 +187,27 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe
TrackTime = @event.TrackTime,
TrackTimeZone = @event.TrackTimeZone
});
// 触发轨迹推送事件（异步）
foreach (var group in @event.OrderIdentifiers.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 = firstTrack.TrackStatus,
TrackContent = firstTrack.TrackContent,
TrackPlace = @event.TrackPlace,
TrackTime = @event.TrackTime,
TrackTimeZone = @event.TrackTimeZone
}
};
await _capPublisher.PublishAsync(GlobalConst.TrackPushSubscribe, trackPushEvent);
}
}
}

2
src/SJZY.Track.Application/SJZY.Track.Application.csproj
View File

@ -12,7 +12,7 @@
</ItemGroup>
<ItemGroup>
<PackageReference Include="SJZY.InterfaceRelay.Application.Contracts" Version="7.1.9"/>
<PackageReference Include="SJZY.InterfaceRelay.Application.Contracts" Version="7.2.0"/>
<PackageReference Include="Volo.Abp.Http.Client" Version="7.0.2" />
<PackageReference Include="Volo.Abp.FluentValidation" Version="7.0.2" />
<PackageReference Include="SJZY.UserCenter.HttpApi" Version="3.0.3-alpha8" />

5
src/SJZY.Track.Application/Services/ProvideServices/OrderService.Link.cs
View File

@ -72,7 +72,7 @@ public partial class OrderService
// 过滤掉不存在的订单
var canLinkOrders = await _orderRepository.GetListAsync(x =>
x.OrderType == input.LinkOrderType && noLinkOrderNos.Contains(x.OrderNo),
x => new { x.Id, x.OrderNo, x.OrderType, x.SubSystemId });
x => new { x.Id, x.OrderNo, x.OrderType, x.SubSystemId, x.CustomId });
if (canLinkOrders.Count == 0) return;
@ -100,7 +100,8 @@ public partial class OrderService
{
OrderId = x.Id,
OrderNo = x.OrderNo,
OrderType = x.OrderType
OrderType = x.OrderType,
CustomId = x.CustomId
}).ToArray(),
TrackCode = track.TrackCode,
TrackPlace = track.TrackPlace,

3
src/SJZY.Track.Application/Services/ProvideServices/TrackPullService.cs
View File

@ -65,7 +65,8 @@ public class TrackPullService : ApplicationService
var attach = Base64Helper.SerializeObject(group.Key);
// 批量获取轨迹
var _ = await _httpClientProxy.Service.BatchForwardThenPushToQueueAsync(
"A012", serviceProviderCode, data, GlobalConst.TrackRefreshSubscribe, attach);
InterfaceRelayConst.GetTrackInterfaceCode, serviceProviderCode, data, GlobalConst.TrackRefreshSubscribe,
attach);
}
return true;

43
src/SJZY.Track.Application/Services/ProvideServices/TrackPushService.cs
View File

@ -3,7 +3,6 @@ using Microsoft.AspNetCore.Authorization;
using SJZY.Track.Application.Contract.Dtos.ProvideServices;
using SJZY.Track.Application.Contract.Events;
using SJZY.Track.Domain.Aggregates.OrderAggregate;
using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate;
using SJZY.Track.Domain.Aggregates.TrackPushAggregate;
using SJZY.Track.Domain.Consts;
using SJZY.UserCenter.HttpApi.Services;
@ -17,17 +16,14 @@ namespace SJZY.Track.Application.Services.ProvideServices;
[Route("/ProviderService")]
public class TrackPushService : ApplicationService
{
private readonly ITrackDetailsRepository _trackDetailsRepository;
private readonly IOrderRepository _orderRepository;
private readonly ICapPublisher _capPublisher;
private readonly ITrackPushRecordsRepository _trackPushRecordsRepository;
public TrackPushService(ITrackDetailsRepository trackDetailsRepository
, IOrderRepository orderRepository
public TrackPushService(IOrderRepository orderRepository
, ICapPublisher capPublisher
, ITrackPushRecordsRepository trackPushRecordsRepository)
{
_trackDetailsRepository = trackDetailsRepository;
_orderRepository = orderRepository;
_capPublisher = capPublisher;
_trackPushRecordsRepository = trackPushRecordsRepository;
@ -51,38 +47,17 @@ public class TrackPushService : ApplicationService
throw new BusinessException(message: "一个订单都不存在");
}
var tracks = await _trackDetailsRepository.GetListAsync(orderIds, input.TrackCode);
tracks = tracks.DistinctBy(x => x.OrderId).ToList();
if (tracks.Count == 0)
var trackPushEvent = new TrackPushEvent
{
throw new BusinessException(message: "一个能推的轨迹都不存在");
}
OptionType = OptionType.新增,
OrderIds = orderIds,
TrackCode = input.TrackCode,
OperationMan = CurrentUserRequest.CurrentUser.Name,
};
var res = 0;
await _capPublisher.PublishAsync(GlobalConst.TrackPushSubscribe, trackPushEvent);
foreach (var track in tracks)
{
var trackPushEvent = new TrackPushEvent
{
OptionType = OptionType.新增,
OrderId = track.OrderId,
OrderNo = track.OrderNo,
TrackStatus = track.TrackStatus,
TrackCode = track.TrackCode,
TrackContent = track.TrackContent,
TrackPlace = track.TrackPlace,
TrackTime = track.TrackTime,
TrackTimeZone = track.TrackTimeZone,
OperationMan = CurrentUserRequest.CurrentUser.Name,
};
await _capPublisher.PublishAsync(GlobalConst.TrackPushSubscribe, trackPushEvent);
res++;
}
return $"即将推送轨迹代码{input.TrackCode}给{res}个订单";
return $"即将推送轨迹代码{input.TrackCode}给{orderIds.Length}个订单";
}

16
src/SJZY.Track.Application/Services/ProvideServices/TrackService.cs
View File

@ -43,7 +43,7 @@ public partial class TrackService : ApplicationService, ITrackService
var order = await _orderRepository.GetFirstAsync(x =>
x.OrderType == input.OrderType && x.OrderNo == input.OrderNo, x => new
{
x.Id, x.SubSystemId, x.OrderNo, x.OrderType
x.Id, x.SubSystemId, x.OrderNo, x.OrderType, x.CustomId
});
if (order == default)
{
@ -53,15 +53,16 @@ public partial class TrackService : ApplicationService, ITrackService
// 新增/更新轨迹
var @event = ObjectMapper.Map<UpsertTrackInput, TrackUpsertEvent>(input);
@event.SubSystemId = order.SubSystemId;
@event.OrderIdentifiers = new[]
{
@event.OrderIdentifiers =
[
new OrderIdentifier
{
OrderId = order.Id,
OrderNo = order.OrderNo,
OrderType = order.OrderType
OrderType = order.OrderType,
CustomId = order.CustomId
}
};
];
@event.OperationMan = CurrentUserRequest.CurrentUser.Name;
await _capPublisher.PublishAsync(GlobalConst.TrackUpsertSubscribe, @event);
return new UnifyResultDto(true, true);
@ -85,7 +86,7 @@ public partial class TrackService : ApplicationService, ITrackService
var orders = await _orderRepository.GetListAsync(x =>
x.OrderType == input.OrderType && input.OrderNos.Contains(x.OrderNo), x => new
{
x.Id, x.OrderNo, x.SubSystemId
x.Id, x.OrderNo, x.SubSystemId, x.CustomId
});
foreach (var group in orders.GroupBy(x => x.SubSystemId))
@ -98,7 +99,8 @@ public partial class TrackService : ApplicationService, ITrackService
{
OrderId = order.Id,
OrderNo = order.OrderNo,
OrderType = input.OrderType
OrderType = input.OrderType,
CustomId = order.CustomId
}).ToArray(),
IsAdd = input.IsAdd,
TrackCode = input.TrackCode,

28
src/SJZY.Track.Application/Services/TrackPush/TrackPushCacheService.cs
View File

@ -43,34 +43,6 @@ public class TrackPushCacheService : ITrackPushCacheService, ITransientDependenc
return data;
}
public async Task<TrackPushContentConfigDto?> GetPushContentConfigAsync(int subSystemId, int customId,
string trackConde)
{
var configs = await GetPushContentConfigsAsync(subSystemId, customId);
return configs?.FirstOrDefault(x => x.TrackCode == trackConde);
}
public async Task<TrackPushContentConfigDto?> IsValidPushAsync(int subSystemId, int customId,
string? latestPushTrackCode,
string? trackCode)
{
var cache = await GetPushContentConfigsAsync(subSystemId, customId);
if (cache == null || cache.Count < 1) return default;
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 : default;
}
private async Task<List<TrackPushContentConfigDto>?> DoGetPushContentConfigsAsync(int subSystemId, int customId)
{
var trackPushConfig = await _trackPushConfigRepository.GetFirstAsync(

2
src/SJZY.Track.Domain/Aggregates/TrackDetailsAggregate/ITrackDetailsRepository.cs
View File

@ -13,6 +13,8 @@ public interface ITrackDetailsRepository
Task<bool> AnyAsync(int orderId, string originTrackData);
Task<bool> AnyByCodeAsync(int orderId, string trackCode);
Task UpdateAsync(TrackDetailsIndex index);
Task AddManyAsync(List<TrackDetailsIndex> index);

2
src/SJZY.Track.Domain/Aggregates/TrackPushAggregate/ITrackPushRecordsRepository.cs
View File

@ -9,6 +9,8 @@ public interface ITrackPushRecordsRepository
/// <returns></returns>
Task AddAsync(TrackPushRecordIndex index);
Task AddManyAsync(List<TrackPushRecordIndex> index);
/// <summary>
/// 获取轨迹推送记录
/// </summary>

5
src/SJZY.Track.Domain/Aggregates/TrackPushAggregate/TrackPushRecordIndex.cs
View File

@ -17,6 +17,11 @@ public class TrackPushRecordIndex : IdBaseIndex
/// </summary>
public int OrderId { get; set; }
/// <summary>
/// 订单号
/// </summary>
public string OrderNo { get; set; } = null!;
/// <summary>
/// 接口通请求Id
/// </summary>

5
src/SJZY.Track.Domain/Consts/GlobalConst.MQ.cs
View File

@ -32,6 +32,11 @@ public partial class GlobalConst
/// </summary>
public const string TrackPushSubscribe = "SJZY.Track.TrackPush";
/// <summary>
/// 轨迹推送完成（接口通回调）
/// </summary>
public const string TrackPushCompletedSubscribe = "SJZY.Track.TrackPushCompleted";
/// <summary>
/// 轨迹新增/更新/删除后触发更新订单主表
/// </summary>

5
src/SJZY.Track.Domain/Dtos/Order/LinkedOrderVO.cs
View File

@ -26,4 +26,9 @@ public class LinkedOrderVO
/// 子系统id
/// </summary>
public int SubSystemId { get; set; }
/// <summary>
/// 客户id
/// </summary>
public int CustomId { get; set; }
}

3
src/SJZY.Track.Repository/Repositories/OrderRepository.cs
View File

@ -78,7 +78,8 @@ AND `a`.`IsDeleted` = 0 AND `b`.`IsDeleted` = 0;");
OrderId = b.Id,
OrderNo = b.OrderNo,
OrderType = b.OrderType,
SubSystemId = b.SubSystemId
SubSystemId = b.SubSystemId,
CustomId = b.CustomId
})
.ToListAsync();
}

38
src/SJZY.Track.Repository/Repositories/TrackDetailsRepository.cs
View File

@ -31,7 +31,7 @@ internal class TrackDetailsRepository : ElasticSearchContextAbstract<TrackDetail
if (!r.IsValid)
{
throw new Exception($"轨迹查询，原因：{r.DebugInformation}。");
throw new Exception($"轨迹查询异常，原因：{r.DebugInformation}。");
}
return r.ToIdList();
@ -82,7 +82,7 @@ internal class TrackDetailsRepository : ElasticSearchContextAbstract<TrackDetail
if (!r.IsValid)
{
throw new Exception($"轨迹查询，原因：{r.DebugInformation}。");
throw new Exception($"轨迹查询异常，原因：{r.DebugInformation}。");
}
return r.Documents.ToList();
@ -100,7 +100,7 @@ internal class TrackDetailsRepository : ElasticSearchContextAbstract<TrackDetail
if (!r.IsValid)
{
throw new Exception($"轨迹查询，原因：{r.DebugInformation}。");
throw new Exception($"轨迹查询异常，原因：{r.DebugInformation}。");
}
return r.Hits.Count > 0;
@ -118,7 +118,25 @@ internal class TrackDetailsRepository : ElasticSearchContextAbstract<TrackDetail
if (!r.IsValid)
{
throw new Exception($"轨迹查询，原因：{r.DebugInformation}。");
throw new Exception($"轨迹查询异常，原因：{r.DebugInformation}。");
}
return r.Hits.Count > 0;
}
public async Task<bool> AnyByCodeAsync(int orderId, string trackCode)
{
var r = await Context.SearchAsync<TrackDetailsIndex>(x => x
.Source(s => s.Includes(fd => fd.Field(f => f.TrackCode)))
.Query(q =>
q.Term(f => f.OrderId, orderId)
&& q.Term(f => f.TrackCode.Suffix("keyword"), trackCode)
)
);
if (!r.IsValid)
{
throw new Exception($"轨迹查询异常，原因：{r.DebugInformation}。");
}
return r.Hits.Count > 0;
@ -128,9 +146,8 @@ internal class TrackDetailsRepository : ElasticSearchContextAbstract<TrackDetail
public async Task UpdateAsync(TrackDetailsIndex track)
{
var r = await Context.UpdateAsync<TrackDetailsIndex>(track.Id, u => u
.Index(track.IndexName)
.Doc(track)
//.Refresh(Refresh.True) // 可选：立即刷新索引使更改立即可见
.Index(track.IndexName)
.Doc(track)
);
if (!r.IsValid)
@ -193,7 +210,7 @@ internal class TrackDetailsRepository : ElasticSearchContextAbstract<TrackDetail
if (!r.IsValid)
{
throw new Exception($"轨迹查询，原因：{r.DebugInformation}。");
throw new Exception($"轨迹查询异常，原因：{r.DebugInformation}。");
}
return r.Documents.ToList();
@ -210,7 +227,7 @@ internal class TrackDetailsRepository : ElasticSearchContextAbstract<TrackDetail
if (!r.IsValid)
{
throw new Exception($"轨迹查询，原因：{r.DebugInformation}。");
throw new Exception($"轨迹查询异常，原因：{r.DebugInformation}。");
}
return r.Documents.ToList();
@ -219,12 +236,13 @@ internal class TrackDetailsRepository : ElasticSearchContextAbstract<TrackDetail
public async Task DeleteTrackAsync(int[] orderIds, List<string> trackCodes)
{
var r = await Context.DeleteByQueryAsync<TrackDetailsIndex>(x => x
.Conflicts(Conflicts.Proceed)
.Conflicts(Conflicts.Abort)
.Query(
q =>
q.Terms(t => t.Field(f => f.OrderId).Terms(orderIds))
&& q.Term(t => t.IsCanDelete, true)
&& q.Terms(t => t.Field(f => f.TrackCode.Suffix("keyword")).Terms(trackCodes)))
.WaitForCompletion()
);
if (!r.IsValid)

18
src/SJZY.Track.Repository/Repositories/TrackPushRecordsRepository.cs
View File

@ -16,6 +16,8 @@ public class TrackPushRecordsRepository : ElasticSearchContextAbstract<TrackDeta
public async Task AddAsync(TrackPushRecordIndex index)
{
index.IndexName = $"{GlobalConst.TrackIndex}-{DateTime.Now:yyyy.MM.dd}";
index.CreateTime = DateTime.Now;
// 按天写入es
var r = await Context.IndexAsync(index, request => request
.Index(index.IndexName)
@ -28,6 +30,22 @@ public class TrackPushRecordsRepository : ElasticSearchContextAbstract<TrackDeta
}
}
public async Task AddManyAsync(List<TrackPushRecordIndex> records)
{
if (!records.Any()) return;
// 按天写入es
var indexName = $"{GlobalConst.TrackIndex}-{DateTime.Now:yyyy.MM.dd}";
records.ForEach(x => x.IndexName = indexName);
var r = await Context.IndexManyAsync(records, indexName);
if (!r.IsValid)
{
throw new Exception($"{records.Count}条轨迹推送记录写入失败，原因：{r.DebugInformation}。");
}
}
public async Task<List<TrackPushRecordIndex>> GetTrackPushRecordsAsync(int orderId)
{
var data = await Context.SearchAsync<TrackPushRecordIndex>(x => x