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

Merge branch 'refs/heads/S03-轨迹中台优化项' into dev

Browse Source
This commit is contained in:
李鹏军 2024-04-24 17:49:56 +08:00
commit 79c0a8f52c
34 changed files with 576 additions and 381 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
.idea/.idea.SJZY.Track.Api/.idea/.gitignore generated vendored Normal file
View File

@ -0,0 +1,13 @@
# Default ignored files
/shelf/
/workspace.xml
# Rider ignored files
/projectSettingsUpdater.xml
/.idea.SJZY.Track.Api.iml
/modules.xml
/contentModel.xml
# Editor-based HTTP Client requests
/httpRequests/
# Datasource local storage ignored files
/dataSources/
/dataSources.local.xml

10
src/SJZY.Track.Application.Contract/Dtos/TrackFlags/GetTrackFlagListDto.cs
View File

@ -58,11 +58,11 @@
/// <summary>
/// 选中的轨迹标记id
/// </summary>
public List<int>? SelectIds { get; set; }
public List<int>? SelectIds { get; set; }
///// <summary>
///// 关键字
///// </summary>
//public string Keywords { get; set; }
/// <summary>
/// 关键字
/// </summary>
public string? Keywords { get; set; }
}
}

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);
}

22
src/SJZY.Track.Application/EventHandlers/CAPFilter.cs
View File

@ -57,12 +57,12 @@ public class CapFilter : SubscribeFilter
/// <returns></returns>
public override Task OnSubscribeExecutedAsync(ExecutedContext context)
{
context.DeliverMessage.Headers.TryGetValue("sjzy-receive-time", out var receiveTime);
context.DeliverMessage.Headers.TryGetValue("cap-msg-name", out var receiveName);
context.DeliverMessage.Headers.TryGetValue("sjzy-arguments", out var arguments);
context.DeliverMessage.Headers.TryGetValue("cap-msg-id", out var messageId);
BuildLog(receiveTime!, receiveName!, arguments!, false, messageId!);
// context.DeliverMessage.Headers.TryGetValue("sjzy-receive-time", out var receiveTime);
// context.DeliverMessage.Headers.TryGetValue("cap-msg-name", out var receiveName);
// context.DeliverMessage.Headers.TryGetValue("sjzy-arguments", out var arguments);
// context.DeliverMessage.Headers.TryGetValue("cap-msg-id", out var messageId);
//
// BuildLog(receiveTime!, receiveName!, arguments!, false, messageId!);
return Task.CompletedTask;
}
@ -86,11 +86,11 @@ public class CapFilter : SubscribeFilter
LogHelper.Error($"cap-msg-id：{messageId}\n{context.Exception}");
context.DeliverMessage.Headers.TryGetValue("sjzy-receive-time", out var receiveTime);
context.DeliverMessage.Headers.TryGetValue("cap-msg-name", out var receiveName);
context.DeliverMessage.Headers.TryGetValue("sjzy-arguments", out var arguments);
BuildLog(receiveTime!, receiveName!, arguments, true, messageId!);
// context.DeliverMessage.Headers.TryGetValue("sjzy-receive-time", out var receiveTime);
// context.DeliverMessage.Headers.TryGetValue("cap-msg-name", out var receiveName);
// context.DeliverMessage.Headers.TryGetValue("sjzy-arguments", out var arguments);
//
// BuildLog(receiveTime!, receiveName!, arguments, true, messageId!);
return Task.CompletedTask;
}

19
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;
}
@ -49,7 +49,6 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe
nextEvent.Visited.AddRange(@event.OrderIds);
nextEvent.TrackCodes = @event.TrackCodes;
nextEvent.SubSystemId = group.Key;
nextEvent.OrderIds = group.Select(x => x.OrderId).ToArray();
// 不同子系统需要做内容映射
if (@event.SubSystemId != nextEvent.SubSystemId)
@ -60,13 +59,25 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe
if (mapCodes.Count == 0)
{
//throw new BusinessException(message: "订单跨子系统，未配置内容映射，请联系管理员");
return;
continue;
}
nextEvent.TrackCodes = mapCodes;
}
await _capPublisher.PublishAsync(GlobalConst.TrackDeleteSubscribe, nextEvent);
var identifiers = group.Select(x => x.OrderId).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.OrderIds = identifiers.GetRange(idx, cnt).ToArray();
idx += GlobalConst.ChunkSize;
await _capPublisher.PublishAsync(GlobalConst.TrackDeleteSubscribe, subEvent);
}
}
}
}

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
});
}
}

336
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,214 +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();
}
if (!log.IsSuccess)
{
if (@event.RetryNum < 3)
{
@event.RetryNum++;
// 1, 2, 4min
await _capPublisher.PublishDelayAsync(
TimeSpan.FromSeconds((1 << @event.RetryNum - 1) * RandomHelper.GetRandom(30, 60)),
GlobalConst.TrackPushSubscribe, @event);
return;
}
}
// 往es 写推送日志
await _trackPushRecordsRepository.AddAsync(log);
// 同步去更新订单主表
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,
@ -234,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);
}
}

81
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,17 +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.OrderIdentifiers = group.Select(x => new OrderIdentifier
{
OrderId = x.OrderId,
OrderNo = x.OrderNo,
OrderType = x.OrderType
}).ToArray();
nextEvent.SubSystemId = group.Key.SubSystemId;
nextEvent.TrackRefreshData = new TrackRefreshStruct
{
ServiceProviderName = @event.TrackRefreshData.ServiceProviderName,
@ -78,13 +72,31 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe
if (mapCode == default)
{
// 跨子系统没做映射的轨迹不处理
return;
continue;
}
nextEvent.TrackCode = mapCode;
}
await _capPublisher.PublishAsync(GlobalConst.TrackUpsertSubscribe, nextEvent);
var identifiers = group.Select(x => new OrderIdentifier
{
OrderId = x.OrderId,
OrderNo = x.OrderNo,
OrderType = x.OrderType,
CustomId = x.CustomId
}).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);
}
}
}
}
@ -94,7 +106,7 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe
var orderIds = @event.OrderIdentifiers.Select(x => x.OrderId).ToArray();
var tracks = await _trackDetailsRepository.GetListAsync(orderIds, @event.TrackCode);
TrackDetailsIndex firstTrack = null;
TrackDetailsIndex? firstTrack = null;
var insertTracks = new List<TrackDetailsIndex>();
@ -102,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)
{
@ -156,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;
@ -194,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,

45
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;
@ -88,32 +89,36 @@ public class TrackPullService : ApplicationService
continue;
}
var orders = await _provideQueries.FindOrderWithServiceProviderAsync(config.ServiceProviderId);
var serviceProviderCode = orders.FirstOrDefault()?.ServiceProviderCode;
var serviceProviderName = orders.FirstOrDefault()?.ServiceProviderName;
foreach (var group in orders.GroupBy(x =>
new { x.SubSystemId, x.OrderType }))
var page = 1;
var orders = await _provideQueries.FindOrderWithServiceProviderAsync(config.ServiceProviderId, page);
while (orders.Any())
{
var data = group.Select(x => new
{
x.OrderNo
}).ToList();
var serviceProviderCode = orders.FirstOrDefault()?.ServiceProviderCode;
var serviceProviderName = orders.FirstOrDefault()?.ServiceProviderName;
var attach = Base64Helper.SerializeObject(new TrackRefreshAttach(config.ServiceProviderId,
serviceProviderName!,
group.Key.SubSystemId, group.Key.OrderType));
// 批量获取轨迹
await _httpClientProxy.Service.BatchForwardThenPushToQueueAsync(
InterfaceRelayConst.GetTrackInterfaceCode, serviceProviderCode, data,
GlobalConst.TrackRefreshSubscribe, attach);
foreach (var group in orders.GroupBy(x =>
new { x.SubSystemId, x.OrderType }))
{
var data = group.Select(x => new
{
x.OrderNo
}).ToList();
var attach = Base64Helper.SerializeObject(new TrackRefreshAttach(config.ServiceProviderId,
serviceProviderName!,
group.Key.SubSystemId, group.Key.OrderType));
// 批量获取轨迹
await _httpClientProxy.Service.BatchForwardThenPushToQueueAsync(
InterfaceRelayConst.GetTrackInterfaceCode, serviceProviderCode, data,
GlobalConst.TrackRefreshSubscribe, attach);
}
orders = await _provideQueries.FindOrderWithServiceProviderAsync(config.ServiceProviderId, ++page);
}
if (config.PullInterval < 60) config.PullInterval = 60;
await RedisHelper.Client.SetExAsync(redisKey, config.PullInterval * 60, orders.Count);
}
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}个订单";
}

18
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,
@ -186,7 +188,7 @@ public partial class TrackService : ApplicationService, ITrackService
var @event = new TrackDeleteEvent
{
OrderIds = group.Select(x => x.OrderId).ToArray(),
TrackCodes = codes.Select(x => x.TrackCode).ToList(),
TrackCodes = input.TrackCodes,
SubSystemId = orders.First(x => x.Id == group.Key).SubSystemId,
OperationMan = CurrentUserRequest.CurrentUser.Name
};

2
src/SJZY.Track.Application/Services/TrackPull/TrackPullService.cs
View File

@ -86,6 +86,8 @@ public class TrackPullService : ApplicationService
[HttpPost("Edit")]
public async Task<bool> EditAsync([FromBody] EditTrackPullConfigInput input)
{
if (input.PullInterval < 60) input.PullInterval = 60;
if (input.Options != OptionTypeNoNull.删除)
{
Expression<Func<TrackPullConfigEntity, bool>> isAnyExp = x =>

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>

1
src/SJZY.Track.Domain/Consts/GlobalConst.cs
View File

@ -4,4 +4,5 @@ public partial class GlobalConst
{
public const string SystemUser = "系统";
public const string EmptyString = "--";
public const int ChunkSize = 100;
}

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; }
}

8
src/SJZY.Track.Domain/Dtos/TrackFlags/GetPageTrackFlagVO.cs
View File

@ -18,10 +18,10 @@ namespace SJZY.Track.Domain.Dtos.TrackFlags
/// </summary>
public List<int>? SelectIds { get; set; }
///// <summary>
///// 关键字
///// </summary>
//public string Keywords { get; set; }
/// <summary>
/// 关键字
/// </summary>
public string? Keywords { get; set; }
}
public class GetTrackFlagListVO

5
src/SJZY.Track.Domain/Queries/IProvideQueries.cs
View File

@ -60,6 +60,9 @@ public interface IProvideQueries
/// 获取服务商下所有单（未完成的）
/// </summary>
/// <param name="serviceProvider"></param>
/// <param name="pageIndex"></param>
/// <param name="pageSize"></param>
/// <returns></returns>
Task<List<OrderWithServiceProviderVO>> FindOrderWithServiceProviderAsync(int serviceProvider);
Task<List<OrderWithServiceProviderVO>> FindOrderWithServiceProviderAsync(int serviceProvider, int pageIndex,
int pageSize = 1000);
}

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

@ -8,7 +8,7 @@
</PropertyGroup>
<ItemGroup>
<PackageReference Include="SJZY.Expand.ABP.Core" Version="7.7.1" />
<PackageReference Include="SJZY.Expand.ABP.Core" Version="7.7.5"/>
<PackageReference Include="SJZY.Expand.ElasticSearch" Version="7.0.17" />
</ItemGroup>

5
src/SJZY.Track.Repository/Queries/ProvideQueries.cs
View File

@ -113,7 +113,8 @@ public class ProvideQueries : ISugarRepository, IProvideQueries,
.ToListAsync();
}
public async Task<List<OrderWithServiceProviderVO>> FindOrderWithServiceProviderAsync(int serviceProvider)
public async Task<List<OrderWithServiceProviderVO>> FindOrderWithServiceProviderAsync(int serviceProvider,
int pageIndex, int pageSize = 1000)
{
return await Context.Queryable<OrderEntity>()
.RightJoin<OrderTemplateEntity>((a, b) => a.Id == b.OrderId)
@ -128,6 +129,8 @@ public class ProvideQueries : ISugarRepository, IProvideQueries,
ServiceProviderName = c.Name,
SubSystemId = a.SubSystemId
})
.Skip((pageIndex - 1) * pageSize)
.Take(pageSize)
.ToListAsync();
}
}

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();
}

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

@ -1,3 +1,4 @@
using Elasticsearch.Net;
using Nest;
using SJZY.Expand.ElasticSearch.Services;
using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate;
@ -28,6 +29,11 @@ internal class TrackDetailsRepository : ElasticSearchContextAbstract<TrackDetail
)
);
if (!r.IsValid)
{
throw new Exception($"轨迹查询异常，原因：{r.DebugInformation}。");
}
return r.ToIdList();
}
@ -35,7 +41,7 @@ internal class TrackDetailsRepository : ElasticSearchContextAbstract<TrackDetail
{
var excludeTrackCodes = trackCodes.ToList();
excludeTrackCodes.Add(GlobalConst.EmptyString);
var data = await Context.SearchAsync<TrackDetailsIndex>(x => x
var r = await Context.SearchAsync<TrackDetailsIndex>(x => x
.Source(s => s.Includes(fd => fd
.Field(f => f.TrackEnContent)
.Field(f => f.TrackContent)
@ -74,7 +80,12 @@ internal class TrackDetailsRepository : ElasticSearchContextAbstract<TrackDetail
)
);
return data.Documents.ToList();
if (!r.IsValid)
{
throw new Exception($"轨迹查询异常，原因：{r.DebugInformation}。");
}
return r.Documents.ToList();
}
public async Task<bool> AnyAsync(int orderId, DateTime trackTime)
@ -87,6 +98,11 @@ internal class TrackDetailsRepository : ElasticSearchContextAbstract<TrackDetail
)
);
if (!r.IsValid)
{
throw new Exception($"轨迹查询异常，原因：{r.DebugInformation}。");
}
return r.Hits.Count > 0;
}
@ -100,6 +116,29 @@ internal class TrackDetailsRepository : ElasticSearchContextAbstract<TrackDetail
)
);
if (!r.IsValid)
{
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;
}
@ -107,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)
@ -150,7 +188,7 @@ internal class TrackDetailsRepository : ElasticSearchContextAbstract<TrackDetail
{
if (orderNos.Count < 1) return new List<TrackDetailsIndex>();
var data = await Context.SearchAsync<TrackDetailsIndex>(x => x
var r = await Context.SearchAsync<TrackDetailsIndex>(x => x
.Source(s => s.Includes(fd => fd
.Field(f => f.OrderNo)
.Field(f => f.TrackStatus)
@ -170,30 +208,41 @@ internal class TrackDetailsRepository : ElasticSearchContextAbstract<TrackDetail
.Size(1000)
);
return data.Documents.ToList();
if (!r.IsValid)
{
throw new Exception($"轨迹查询异常，原因：{r.DebugInformation}。");
}
return r.Documents.ToList();
}
public async Task<List<TrackDetailsIndex>> GetListAsync(int orderId)
{
var data = await Context.SearchAsync<TrackDetailsIndex>(x => x
var r = await Context.SearchAsync<TrackDetailsIndex>(x => x
.Query(q =>
q.Term(t => t.OrderId, orderId)
)
.Size(1000)
.Sort(s => s.Field(f => f.ZeroTrackTime, SortOrder.Descending)));
return data.Documents.ToList();
if (!r.IsValid)
{
throw new Exception($"轨迹查询异常，原因：{r.DebugInformation}。");
}
return r.Documents.ToList();
}
public async Task DeleteTrackAsync(int[] orderIds, List<string> trackCodes)
{
var r = await Context.DeleteByQueryAsync<TrackDetailsIndex>(x => x
.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)))
//.Refresh(Refresh.True) // 可选：立即刷新索引使更改立即可见
.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)

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

@ -46,8 +46,9 @@ public class TrackFlagRepository : BasicRepository<TrackFlagEntity>, ITrackFlagR
.LeftJoin<VUserEntity>((a, b, c) => a.CreateUserId == c.Id && c.IsDeleted == false)
.LeftJoin<VUserEntity>((a, b, c, d) => a.UpdateUserId == d.Id && d.IsDeleted == false)
.WhereIF(input.SubSystemId > 0, (a, b, c, d) => a.SubSystemId == input.SubSystemId)
.WhereIF(!input.Keywords.IsNullOrWhiteSpace(), (a, b, c, d) => SqlFunc.Contains(a.Code, input.Keywords))
.OrderByIF(input.SelectIds.IsNullOrEmpty(), (a, b, c, d) => a.CreateTime, OrderByType.Desc)
.OrderByIF(!input.SelectIds.IsNullOrEmpty(), (a, b, c, d) => SqlFunc.IIF(input.SelectIds.Contains(a.Id), SqlFunc.DateAdd(SqlFunc.GetDate(), 1), a.CreateTime), OrderByType.Desc)
.OrderByIF(!input.SelectIds.IsNullOrEmpty(), (a, b, c, d) => SqlFunc.IIF(input.SelectIds!.Contains(a.Id), SqlFunc.DateAdd(SqlFunc.GetDate(), 1), a.CreateTime), OrderByType.Desc)
.Select((a, b, c, d) => new GetTrackFlagListVO
{
Id = a.Id,

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