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

Merge branch 'refs/heads/test_对接接口通' into test

Browse Source
This commit is contained in:
xiaolipro 2024-08-12 15:17:43 +08:00
commit 7284868a15
15 changed files with 225 additions and 87 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
src/SJZY.Track.Application.Contract/Dtos/TrackPull/TrackPullPayload.cs Normal file
View File

@ -0,0 +1,9 @@
namespace SJZY.Track.Application.Contract.Dtos.TrackPull;
public class TrackPullPayload
{
/// <summary>
/// 订单号
/// </summary>
public string OrderNo { get; set; } = null!;
}

3
src/SJZY.Track.Application.Contract/SJZY.Track.Application.Contract.csproj
View File

@ -15,8 +15,5 @@
<ItemGroup>
<ProjectReference Include="..\SJZY.Track.Shared\SJZY.Track.Shared.csproj" />
</ItemGroup>
<ItemGroup>
<Folder Include="Dtos\Common\" />
</ItemGroup>
</Project>

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

@ -74,7 +74,7 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe
private async Task DoPushAsync(TrackPushEvent @event)
{
var items = new List<TrackPushInput>();
var items = new List<object>();
TrackPushContentConfigDto? config;
if (@event.OperationMan == null)
@ -165,7 +165,7 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe
var config = _pushConfigs.FirstOrDefault(x => x.TrackCode == @event.TrackCode);
if (config == null || !config.IsCanUpdate) return;
var items = new List<TrackUpdateInput>();
var items = new List<object>();
foreach (var order in _orders)
{
var log = new TrackPushRecordIndex

119
src/SJZY.Track.Application/CAP/EventHandlers/TrackRefreshEventHandler.cs → src/SJZY.Track.Application/Rabbit/TrackRefreshEventHandler.cs
View File

@ -1,8 +1,14 @@
using System.Diagnostics;
using System.Linq.Expressions;
using System.Text;
using BSI.Stack.Rabbit;
using BSI.Stack.Rabbit.Handler;
using DotNetCore.CAP;
using Microsoft.Extensions.Logging;
using Newtonsoft.Json;
using Newtonsoft.Json.Linq;
using RabbitMQ.Client.Events;
using SJZY.Expand.ABP.Core.PublicMethords;
using SJZY.InterfaceRelay.Application.Contracts;
using SJZY.Track.Application.Contract.Dtos.TrackPull;
using SJZY.Track.Application.Contract.Events;
@ -15,37 +21,72 @@ using SJZY.Track.Domain.Enums;
using SJZY.Track.Domain.Queries;
using Volo.Abp.DependencyInjection;
namespace SJZY.Track.Application.CAP.EventHandlers;
namespace SJZY.Track.Application.Rabbit;
public class TrackRefreshEventHandler : ITransientDependency, ICapSubscribe
public class TrackRefreshEventHandler : ITransientDependency, IRabbitHandler
{
private readonly ICapPublisher _capPublisher;
private readonly IOrderRepository _orderRepository;
private readonly ITimezoneQueries _timezoneQueries;
private readonly ILogger<TrackRefreshEventHandler> _logger;
private readonly ITrackDetailsRepository _trackDetailsRepository;
private readonly ITrackPullCacheService _trackPullCacheService;
public TrackRefreshEventHandler(IOrderRepository orderRepository,
ICapPublisher capPublisher, ITrackPullCacheService trackPullCacheService,
ITrackDetailsRepository trackDetailsRepository, ITimezoneQueries timezoneQueries)
ITrackDetailsRepository trackDetailsRepository, ITimezoneQueries timezoneQueries,
ILogger<TrackRefreshEventHandler> logger)
{
_orderRepository = orderRepository;
_capPublisher = capPublisher;
_trackPullCacheService = trackPullCacheService;
_trackDetailsRepository = trackDetailsRepository;
_timezoneQueries = timezoneQueries;
_logger = logger;
}
public bool Enable(ConsumeOptions options)
{
return options.Queue == GlobalConst.TrackRefreshSubscribe;
}
public async Task Handle(IServiceProvider sp, BasicDeliverEventArgs args, ConsumeOptions options)
{
var msg = Encoding.UTF8.GetString(args.Body.ToArray());
var @event = System.Text.Json.JsonSerializer.Deserialize<InterfaceRelayForwardReplayEvent>(msg);
if (@event == null)
{
_logger.LogWarning(
$"rabbit on queue({options.Queue}) received msg can't convert to {nameof(InterfaceRelayForwardReplayEvent)}");
return;
}
await ConsumerAsync(@event);
}
[CapSubscribe(GlobalConst.TrackRefreshSubscribe, Group = GlobalConst.TrackRefreshSubscribe)]
public async Task ConsumerAsync(InterfaceRelayForwardReplayEvent @event)
{
if (!@event.IsSuccess) return;
if (!@event.IsSuccess)
{
_logger.LogWarning("请求id: {@RequestId}，轨迹刷新失败", @event.RequestId);
return;
}
var attach = Base64Helper.DeSerialize<TrackRefreshAttach>(@event.Attach);
if (attach is null) return;
if (attach is null || @event.Data == null)
{
_logger.LogWarning("请求id: {@RequestId}，轨迹消费失败，attach或data解析为空", @event.RequestId);
return;
}
var config = await _trackPullCacheService.GetPullParamWithContentConfigAsync(attach.SubSystemId,
attach.ServiceProviderId);
if (@event.Data == null || config == null) return;
if (config == null)
{
_logger.LogWarning("请求id: {@RequestId}，轨迹消费失败，找不到拉取配置：{@Attach}", attach.ToJson(), @event.RequestId);
return;
}
var data = JObject.Parse(@event.Data);
@ -183,27 +224,21 @@ public class TrackRefreshEventHandler : ITransientDependency, ICapSubscribe
// 轨迹时区
var (trackTimeZone, hasTimezoneBottomLine) =
await GetTimezone(paramConfig, matchedContentConfig, context.OrderId, trackToken);
await GetTimezone(paramConfig, matchedContentConfig, context, trackToken);
// 轨迹发生地
var (trackPlace, hasPlaceBottomLine) =
await GetTrackPlace(paramConfig, matchedContentConfig, context.OrderId, trackToken);
await GetTrackPlace(paramConfig, matchedContentConfig, context, trackToken);
if (matchedContentConfig is null)
{
_logger.LogDebug("订单号：{OrderNo}，未匹配到轨迹内容，轨迹报文：{TrackContent}",
context.OrderNo, trackToken.ToString());
await InsertNotMappedTrackAsync(trackToken, context, trackPlace, trackTime, trackTimeZone, paramConfig);
return;
}
int deficiencyTagValue = 0;
// 触发了兜底，但是没有兜底住，对订单进行标记
if (hasPlaceBottomLine || hasTimezoneBottomLine)
{
//1表示时区，2表示发生地，3表示两种
var placeTagValue = hasPlaceBottomLine ? 2 : 0;
var timezoneTagValue = hasTimezoneBottomLine ? 1 : 0;
deficiencyTagValue = placeTagValue + timezoneTagValue;
}
var deficiencyTagValue = CalcDeficiencyTagValue(hasPlaceBottomLine, hasTimezoneBottomLine, trackToken, context);
// 新增/更新轨迹
var trackUpsertEvent = new TrackUpsertEvent
@ -239,6 +274,35 @@ public class TrackRefreshEventHandler : ITransientDependency, ICapSubscribe
await _capPublisher.PublishAsync(GlobalConst.TrackUpsertForRefreshSubscribe, trackUpsertEvent);
}
/// <summary>
/// 触发了兜底，但是没有兜底住，对订单进行标记
/// 1表示时区，2表示发生地，3表示两种
/// </summary>
/// <param name="hasPlaceBottomLine"></param>
/// <param name="hasTimezoneBottomLine"></param>
/// <param name="trackToken"></param>
/// <param name="context"></param>
/// <returns></returns>
private int CalcDeficiencyTagValue(bool hasPlaceBottomLine, bool hasTimezoneBottomLine, JToken trackToken,
TrackRefreshContext context)
{
var deficiencyTagValue = 0;
if (hasTimezoneBottomLine)
{
_logger.LogWarning("订单号：{@OrderNo}, 轨迹发生时区无法解析，触发了兜底，但是没有兜底住：{@Data}", context.OrderNo, trackToken.ToString());
deficiencyTagValue += 1;
}
if (hasPlaceBottomLine)
{
_logger.LogWarning("订单号：{@OrderNo}, 轨迹发生地无法解析，触发了兜底，但是没有兜底住：{@Data}", context.OrderNo, trackToken.ToString());
deficiencyTagValue += 2;
}
return deficiencyTagValue;
}
private async Task InsertNotMappedTrackAsync(JToken trackToken, TrackRefreshContext context, string? trackPlace,
DateTime? trackTime, float? trackTimeZone, TrackPullParamConfigDto paramConfig)
{
@ -263,7 +327,9 @@ public class TrackRefreshEventHandler : ITransientDependency, ICapSubscribe
trackToken[paramConfig.TrackContentName!]?.Value<string>() ?? GlobalConst.EmptyString;
}
track.ZeroTrackTime = track.TrackTime.Subtract(TimeSpan.FromHours(track.TrackTimeZone));
track.ZeroTrackTime = Math.Abs(track.TrackTimeZone - GlobalConst.NoneTrackTimeZone) < 1
? track.TrackTime
: track.TrackTime.Subtract(TimeSpan.FromHours(track.TrackTimeZone));
await _trackDetailsRepository.AddAsync(track);
}
@ -273,11 +339,11 @@ public class TrackRefreshEventHandler : ITransientDependency, ICapSubscribe
/// </summary>
/// <param name="paramConfig">配置信息</param>
/// <param name="matchedContentConfig">配置信息</param>
/// <param name="orderId">订单id</param>
/// <param name="context">订单id</param>
/// <param name="trackToken"></param>
/// <returns>时区,是否执行了兜底轨迹时区</returns>
private async Task<(float? trackTimeZone, bool hasTimezoneBottomLine)> GetTimezone(
TrackPullParamConfigDto paramConfig, TrackPullContentConfigDto? matchedContentConfig, int orderId,
TrackPullParamConfigDto paramConfig, TrackPullContentConfigDto? matchedContentConfig, TrackRefreshContext context,
JToken trackToken)
{
// 轨迹发生时区
@ -289,7 +355,7 @@ public class TrackRefreshEventHandler : ITransientDependency, ICapSubscribe
if (!trackTimeZone.HasValue && matchedContentConfig?.TrackZonePlaceValue is not null)
{
//兜底没配置的就不需要兜底，拉取的发生地没就没
var timezone = await GetTimezone(matchedContentConfig, orderId);
var timezone = await GetTimezone(matchedContentConfig, context.OrderId);
trackTimeZone = timezone;
hasTimezoneBottomLine = !timezone.HasValue;
@ -303,22 +369,21 @@ public class TrackRefreshEventHandler : ITransientDependency, ICapSubscribe
/// </summary>
/// <param name="paramConfig">配置信息</param>
/// <param name="matchedContentConfig">配置信息</param>
/// <param name="orderId">订单id</param>
/// <param name="context">订单id</param>
/// <param name="trackToken"></param>
/// <returns>时区,是否执行了兜底轨迹时区</returns>
private async Task<(string? trackPlace, bool hasPlaceBottomLine)> GetTrackPlace(TrackPullParamConfigDto paramConfig,
TrackPullContentConfigDto? matchedContentConfig, int orderId, JToken trackToken)
TrackPullContentConfigDto? matchedContentConfig, TrackRefreshContext context, JToken trackToken)
{
// 轨迹发生地
var trackPlace = paramConfig.TrackPlaceName == null
? null
: trackToken[paramConfig.TrackPlaceName]?.Value<string>();
var hasPlaceBottomLine = false; //执行了兜底但是没有兜底住，需要记录
if (trackPlace.IsNullOrWhiteSpace() && matchedContentConfig?.TrackPlaceValue != null &&
matchedContentConfig.TrackPlaceValue.HasValue)
if (trackPlace.IsNullOrWhiteSpace() && matchedContentConfig?.TrackPlaceValue is not null)
{
//兜底没配置的就不需要兜底，拉取的发生地没就没
trackPlace = await GetPlaceByOrder(matchedContentConfig!.TrackPlaceValue.Value, orderId);
trackPlace = await GetPlaceByOrder(matchedContentConfig!.TrackPlaceValue.Value, context.OrderId);
hasPlaceBottomLine = string.IsNullOrWhiteSpace(trackPlace);
}

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

@ -12,7 +12,8 @@
</ItemGroup>
<ItemGroup>
<PackageReference Include="SJZY.InterfaceRelay.Application.Contracts" Version="7.2.1"/>
<PackageReference Include="BSI.Stack.Rabbit" Version="2.0.9" />
<PackageReference Include="SJZY.InterfaceRelay.Application.Contracts" Version="7.2.2"/>
<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.9-alpha"/>

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

@ -67,16 +67,21 @@ public partial class OrderService
x => x.LinkOrderNo);
// 计算出未关联的订单
var noLinkOrderNos = input.LinkOrderNos.Except(linkOrderNos);
var noLinkOrderNos = input.LinkOrderNos.Except(linkOrderNos).ToList();
// 过滤掉不存在的订单
var canLinkOrders = await _orderRepository.GetListAsync(x =>
// 查询存在的订单
var existedOrders = await _orderRepository.GetListAsync(x =>
x.OrderType == input.LinkOrderType && noLinkOrderNos.Contains(x.OrderNo),
x => new { x.Id, x.OrderNo, x.OrderType, x.SubSystemId, x.CustomId });
if (canLinkOrders.Count == 0) return;
// 如果有不存在的单，需要抛异常
if (existedOrders.Count != noLinkOrderNos.Count)
{
throw new BusinessException(
message: $"关联单：{noLinkOrderNos.Except(existedOrders.Select(x => x.OrderNo)).JoinAsString(", ")}不存在");
}
await _orderRepository.LinkOrderAsync(canLinkOrders.Select(x => new OrderLinkEntity
await _orderRepository.LinkOrderAsync(existedOrders.Select(x => new OrderLinkEntity
{
OrderId = orderId,
OrderNo = input.OrderNo,
@ -89,13 +94,14 @@ public partial class OrderService
var tracks = await _trackDetailsRepository.GetListAsync(orderId);
// 补轨迹
foreach (var group in canLinkOrders.GroupBy(x => x.SubSystemId))
foreach (var group in existedOrders.GroupBy(x => new {x.SubSystemId, x.CustomId}))
{
foreach (var track in tracks)
{
var fillTrackEvent = new TrackUpsertEvent
{
SubSystemId = group.Key,
SubSystemId = group.Key.SubSystemId,
CustomId = group.Key.CustomId,
OrderIdentifiers = group.Select(x => new OrderIdentifier
{
OrderId = x.Id,

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

@ -1,14 +1,13 @@
using Microsoft.AspNetCore.Authorization;
using Microsoft.Extensions.Logging;
using SJZY.Expand.ABP.Core;
using SJZY.Expand.ABP.Core.Logger;
using SJZY.Expand.ABP.Core.PublicMethords;
using SJZY.InterfaceRelay.Application.Contracts;
using SJZY.Track.Application.CAP.EventHandlers;
using SJZY.Track.Application.Contract.Dtos.ProvideServices;
using SJZY.Track.Application.Contract.Dtos.TrackPull;
using SJZY.Track.Application.Contract.Services;
using SJZY.Track.Application.Helpers;
using SJZY.Track.Application.Rabbit;
using SJZY.Track.Domain.Consts;
using SJZY.Track.Domain.Queries;
using Volo.Abp.Http.Client.DynamicProxying;
@ -59,11 +58,10 @@ public class TrackPullService : ApplicationService
foreach (var group in orders.GroupBy(x => new
{ x.ServiceProviderId, x.ServiceProviderName, x.SubSystemId, x.OrderType }))
{
var data = group.Select(x => new
var data = new List<object>(group.Select(x => new TrackPullPayload
{
x.OrderNo
}).ToList();
OrderNo = x.OrderNo
}));
var serviceProviderCode = group.FirstOrDefault()!.ServiceProviderCode;
var attach = Base64Helper.SerializeObject(group.Key);
@ -105,7 +103,8 @@ public class TrackPullService : ApplicationService
#if DEBUG
[AllowAnonymous]
#endif
public async Task<bool> TrackRefreshEventTest([FromServices]TrackRefreshEventHandler trackRefreshEventHandler, InterfaceRelayForwardReplayEvent input)
public async Task<bool> TrackRefreshEventTest([FromServices] TrackRefreshEventHandler trackRefreshEventHandler,
InterfaceRelayForwardReplayEvent input)
{
await trackRefreshEventHandler.ConsumerAsync(input);
@ -121,7 +120,7 @@ public class TrackPullService : ApplicationService
}
var page = 1;
var orders = await _provideQueries.FindOrderWithServiceProviderAsync(config.ServiceProviderId, page, 2000);
var orders = await _provideQueries.FindOrderWithServiceProviderAsync(config.ServiceProviderId, page);
while (orders.Any())
{
var serviceProviderCode = orders.FirstOrDefault()?.ServiceProviderCode;
@ -129,9 +128,15 @@ public class TrackPullService : ApplicationService
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));
var res = await _httpClientProxy.Service.BatchForwardThenPushToQueueAsync(InterfaceRelayConst.GetTrackInterfaceCode, serviceProviderCode, data, GlobalConst.TrackRefreshSubscribe, attach);
var data = new List<object>(group.Select(x => new TrackPullPayload
{
OrderNo = x.OrderNo
}));
var attach = Base64Helper.SerializeObject(new TrackRefreshAttach(config.ServiceProviderId,
serviceProviderName!, group.Key.SubSystemId, group.Key.OrderType));
var res = await _httpClientProxy.Service.BatchForwardThenPushToQueueAsync(
InterfaceRelayConst.GetTrackInterfaceCode, serviceProviderCode, data,
GlobalConst.TrackRefreshSubscribe, attach);
if (res.Code != InterfaceRelayConst.SuccessCode)
{

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

@ -27,7 +27,24 @@ public partial class TrackService
var res = new List<GetTrackOutput>();
input.OrderNos = input.OrderNos.Where(x => !x.IsNullOrWhiteSpace()).Distinct().ToList(); // 单号去重、去空
var details = await _trackDetailsRepository.GetListAsync(input.OrderType, input.OrderNos);
object[]? searchAfter = null;
List<TrackDetailsIndex> details = new();
int limit = 1000, cnt, time = 10;
do
{
var (currentResults, nextSearchAfter) =
await _trackDetailsRepository.GetListAsync(input.OrderType, input.OrderNos, searchAfter, limit);
cnt = currentResults.Count;
details.AddRange(currentResults);
// 更新 searchAfter 以获取下一批数据
searchAfter = nextSearchAfter;
time--;
} while (cnt > 0 && time > 0);
if (details.Count == limit * time) throw new BusinessException(message: "查询轨迹量过大，请缩小查询范围");
foreach (var group in details.GroupBy(x => x.OrderId))
{
@ -42,8 +59,12 @@ public partial class TrackService
Content = x.TrackContent,
OriginContent = x.OriginTrackContent,
Location = x.TrackPlace,
Time = Math.Abs(x.TrackTimeZone - GlobalConst.NoneTrackTimeZone) < 1 ? x.TrackTime: x.ZeroTrackTime.AddHours(input.TimeZone ?? 8),
TimeZone = Math.Abs(x.TrackTimeZone - GlobalConst.NoneTrackTimeZone) < 1? null: input.TimeZone ?? 8
Time = Math.Abs(x.TrackTimeZone - GlobalConst.NoneTrackTimeZone) < 1
? x.TrackTime
: x.ZeroTrackTime.AddHours(input.TimeZone ?? 8),
TimeZone = Math.Abs(x.TrackTimeZone - GlobalConst.NoneTrackTimeZone) < 1
? null
: input.TimeZone ?? 8
}).ToList();
res.Add(item);

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

@ -66,6 +66,7 @@ public partial class TrackService : ApplicationService, ITrackService
// 新增/更新轨迹
var @event = ObjectMapper.Map<UpsertTrackInput, TrackUpsertEvent>(input);
@event.SubSystemId = order.SubSystemId;
@event.CustomId = order.CustomId;
@event.OrderIdentifiers = new[]
{
new OrderIdentifier

28
src/SJZY.Track.Application/SjzyTrackApplicationModule.cs
View File

@ -1,6 +1,8 @@
using DotNetCore.CAP;
using BSI.Stack.Rabbit;
using DotNetCore.CAP;
using DotNetCore.CAP.Internal;
using Microsoft.Extensions.Configuration;
using RabbitMQ.Client;
using Savorboard.CAP.InMemoryMessageQueue;
using SJZY.Expand.ABP.Core.Localization;
using SJZY.Expand.ABP.Core.Logger;
@ -9,6 +11,7 @@ using SJZY.InterfaceRelay.Application.Contracts;
using SJZY.Track.Application.CAP;
using SJZY.Track.Application.Contract;
using SJZY.Track.Application.Localization;
using SJZY.Track.Domain.Consts;
using SJZY.UserCenter.Contracts;
using SJZY.UserCenter.SharedModel;
using SJZY.UserCenter.SharedModel.Pub;
@ -121,6 +124,29 @@ namespace SJZY.Track.Application
}
};
}).AddSubscribeFilter<CapFilter>();
context.Services.AddRabbitBoot(options =>
{
options.ConnectionString =
$"amqp://{configuration["RabbitMQ:UserName"]}:{configuration["RabbitMQ:Password"]}@{configuration["RabbitMQ:HostName"]}:{configuration["RabbitMQ:Port"]}";
options.PoolSize = configuration.GetValue<int>("RabbitMQ:PoolSize");
options.Consumes = new List<ConsumeOptions>
{
new()
{
Queue = GlobalConst.TrackRefreshSubscribe,
FetchCount = configuration.GetValue<ushort>("RabbitMQ:.TrackRefreshFetchCount"),
FailedRequeue = false,
Declaration = o =>
{
o.ExchangeDeclare(GlobalConst.InterfaceRelayExchange, ExchangeType.Direct, true);
o.QueueDeclare(GlobalConst.TrackRefreshSubscribe, true);
o.QueueBind(GlobalConst.TrackRefreshSubscribe, GlobalConst.InterfaceRelayExchange, GlobalConst.TrackRefreshSubscribe);
}
}
};
}, typeof(SjzyTrackApplicationModule).Assembly);
}
public override void PreConfigureServices(ServiceConfigurationContext context)

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

@ -8,6 +8,8 @@ public interface ITrackDetailsRepository
Task<List<TrackDetailsIndex>> GetLatestListAsync(int[] eventOrderIds, List<string> trackCodes);
Task<(List<TrackDetailsIndex> results, object[]? nextSearchAfter)> GetListAsync(int orderType, List<string> orderNos, object[]? searchAfter = null, int limit = 1000);
Task<bool> AnyAsync(int orderId, DateTime trackTime);
Task<bool> AnyAsync(int orderId, string originTrackDataHash);
@ -20,8 +22,6 @@ public interface ITrackDetailsRepository
Task AddAsync(TrackDetailsIndex track);
Task<List<TrackDetailsIndex>> GetListAsync(int orderType, List<string> orderNos);
Task<(List<TrackDetailsIndex> list, long total)> GetListAsync(TrackDetailsQueryInput input);
Task<List<TrackDetailsIndex>> GetListAsync(int orderId);

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

@ -12,11 +12,6 @@ public partial class GlobalConst
/// </summary>
public const string TrackUpsertForRefreshSubscribe = "SJZY.Track.TrackUpsertForRefresh";
/// <summary>
/// 无法映射的轨迹新增
/// </summary>
public const string TrackInsertForNotMappedEventSubscribe = "SJZY.Track.TrackInsertForNotMapped";
/// <summary>
/// 轨迹删除
/// </summary>
@ -37,6 +32,11 @@ public partial class GlobalConst
/// </summary>
public const string TrackManualPushSubscribe = "SJZY.Track.TrackManualPush";
/// <summary>
/// 接口通交换机
/// </summary>
public const string InterfaceRelayExchange = "sjzy.interface.relay.direct";
/// <summary>
/// 轨迹推送完成（接口通回调）
/// </summary>
@ -47,16 +47,6 @@ public partial class GlobalConst
/// </summary>
public const string TrackChangeAfterUpdateOrderSubscribe = "SJZY.Track.TrackChangeAfterUpdateOrder";
/// <summary>
/// 轨迹推送后触发更新订单主表
/// </summary>
public const string TrackPushAfterUpdateOrderSubscribe = "SJZY.Track.TrackPushAfterUpdateOrder";
/// <summary>
/// 关联订单后补轨迹
/// </summary>
public const string OrderLinkAfterFillTrack = "SJZY.Track.OrderLinkAfterFillTrack";
/// <summary>
/// 推送预警任务
/// </summary>

2
src/SJZY.Track.Domain/Consts/InterfaceRelayConst.cs
View File

@ -13,7 +13,7 @@ public class InterfaceRelayConst
/// <summary>
/// 获取轨迹接口代码
/// </summary>
public const string GetTrackInterfaceCode = "A012";
public const string GetTrackInterfaceCode = "TEST";
/// <summary>
/// 推送轨迹接口代码

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

@ -122,7 +122,7 @@ public class ProvideQueries : ISugarRepository, IProvideQueries,
return await Context.Queryable<OrderEntity>()
.RightJoin<OrderTemplateEntity>((a, b) => a.Id == b.OrderId)
.LeftJoin<VServiceProviderEntity>((a, b, c) => b.ServiceProviderId == c.Id)
.Where((a, b) => b.ServiceProviderId == serviceProvider && !a.IsCompleted && a.IsTriggerPull==1)
.Where((a, b) => b.ServiceProviderId == serviceProvider && !a.IsCompleted && a.IsTriggerPull == 1)
.Select((a, b, c) => new OrderWithServiceProviderVO
{
OrderNo = a.OrderNo,
@ -141,7 +141,7 @@ public class ProvideQueries : ISugarRepository, IProvideQueries,
{
return await Context.Queryable<OrderEntity>()
.InnerJoin<OrderTemplateEntity>((a, b) => a.Id == b.OrderId)
.Where((a, b) => orderIds.Contains( a.Id) && !a.IsCompleted)
.Where((a, b) => orderIds.Contains(a.Id) && !a.IsCompleted)
.Select((a, b) => new OrderTransportDto
{
OrderId = a.Id,

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

@ -210,9 +210,9 @@ internal class TrackDetailsRepository : ElasticSearchContextAbstract<TrackDetail
}
}
public async Task<List<TrackDetailsIndex>> GetListAsync(int orderType, List<string> orderNos)
public async Task<(List<TrackDetailsIndex> results, object[]? nextSearchAfter)> GetListAsync(int orderType, List<string> orderNos, object[]? searchAfter = null, int limit = 1000)
{
if (orderNos.Count < 1) return new List<TrackDetailsIndex>();
if (orderNos.Count < 1) return (new List<TrackDetailsIndex>(), null);
var r = await Context.SearchAsync<TrackDetailsIndex>(x => x
.Source(s => s.Includes(fd => fd
@ -226,12 +226,26 @@ internal class TrackDetailsRepository : ElasticSearchContextAbstract<TrackDetail
.Field(f => f.TrackTimeZone)
))
.Query(q =>
q.Terms(t => t
.Field(f => f.OrderNo.Suffix("keyword"))
.Terms(orderNos))
&& q.Term(f => f.OrderType, orderType)
q.Bool(b => b
.Must(mu => mu
.Terms(t => t
.Field(f => f.OrderNo.Suffix("keyword"))
.Terms(orderNos)
),
mu => mu
.Term(f => f.OrderType, orderType)
)
.MustNot(mn => mn
.Term(t => t.TrackCode.Suffix("keyword"), GlobalConst.EmptyString)
)
)
)
.Size(1000)
.Sort(s => s
.Ascending(f => f.TrackTime) // 需要根据某个字段排序
.Ascending(f => f.OrderNo) // 可以添加更多排序字段以确保唯一性
)
.Size(limit)
.SearchAfter(searchAfter)
);
if (!r.IsValid)
@ -239,7 +253,10 @@ internal class TrackDetailsRepository : ElasticSearchContextAbstract<TrackDetail
throw new Exception($"轨迹查询异常，原因：{r.DebugInformation}。");
}
return r.Documents.ToList();
var results = r.Documents.ToList();
var nextSearchAfter = r.Hits.LastOrDefault()?.Sorts.ToArray();
return (results, nextSearchAfter);
}
public async Task<List<TrackDetailsIndex>> GetListAsync(int orderId)