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

important：对接接口通获取轨迹

Browse Source
This commit is contained in:
xiaolipro 2024-08-12 15:16:47 +08:00
parent 9965372fda
commit 2a76994eda
10 changed files with 157 additions and 66 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

117
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,28 +224,22 @@ 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)
{
@ -275,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)
{
// 轨迹发生时区
@ -291,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;
@ -305,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"/>

31
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);
@ -71,7 +69,7 @@ public class TrackPullService : ApplicationService
var res = await _httpClientProxy.Service.BatchForwardThenPushToQueueAsync(
InterfaceRelayConst.GetTrackInterfaceCode, serviceProviderCode, data, GlobalConst.TrackRefreshSubscribe,
attach);
if (res.Code != InterfaceRelayConst.SuccessCode)
{
LogHelper.Error($"刷新轨迹失败：{res.ToJson()}");
@ -94,7 +92,7 @@ public class TrackPullService : ApplicationService
await Task.WhenAll(tasks);
return true;
}
/// <summary>
/// 轨迹刷新测试
/// </summary>
@ -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)
{

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)

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>
@ -36,6 +31,11 @@ public partial class GlobalConst
/// 轨迹人工手动推送
/// </summary>
public const string TrackManualPushSubscribe = "SJZY.Track.TrackManualPush";
/// <summary>
/// 接口通交换机
/// </summary>
public const string InterfaceRelayExchange = "sjzy.interface.relay.direct";
/// <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>
/// 推送轨迹接口代码

6
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,
@ -136,12 +136,12 @@ public class ProvideQueries : ISugarRepository, IProvideQueries,
.Take(pageSize)
.ToListAsync();
}
public async Task<List<OrderTransportDto>> FindOrderWithServiceProviderAsync(int[] orderIds)
{
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,