Merge branch 'S02-轨迹中台第二部分' into S03-轨迹中台
# Conflicts: # src/SJZY.Track.Application.Contract/SJZY.Track.Application.Contract.csproj
This commit is contained in:
commit
24c47652d8
@ -2,10 +2,10 @@
|
||||
|
||||
public class TrackRefreshAttach
|
||||
{
|
||||
public TrackRefreshAttach(int serviceProviderId, string serviceProviderCode, int subSystemId, int orderType)
|
||||
public TrackRefreshAttach(int serviceProviderId, string serviceProviderName, int subSystemId, int orderType)
|
||||
{
|
||||
ServiceProviderId = serviceProviderId;
|
||||
ServiceProviderCode = serviceProviderCode;
|
||||
ServiceProviderName = serviceProviderName;
|
||||
SubSystemId = subSystemId;
|
||||
OrderType = orderType;
|
||||
}
|
||||
@ -16,9 +16,9 @@ public class TrackRefreshAttach
|
||||
public int ServiceProviderId { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// 服务商code
|
||||
/// 服务商name
|
||||
/// </summary>
|
||||
public string ServiceProviderCode { get; set; }
|
||||
public string ServiceProviderName { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// 订单类型(主数据枚举TrackOrderType)
|
||||
|
||||
@ -5,7 +5,7 @@
|
||||
<PropertyGroup>
|
||||
<TargetFramework>net7.0</TargetFramework>
|
||||
<GenerateDocumentationFile>True</GenerateDocumentationFile>
|
||||
<Version>1.0.2</Version>
|
||||
<Version>1.0.3</Version>
|
||||
<GenerateDocumentationFile>True</GenerateDocumentationFile>
|
||||
</PropertyGroup>
|
||||
<ItemGroup>
|
||||
|
||||
@ -19,7 +19,8 @@ public interface ITrackPullCacheService
|
||||
/// <param name="subSystemId">子系统id</param>
|
||||
/// <param name="serviceProviderId">服务商id</param>
|
||||
/// <returns></returns>
|
||||
Task<TrackPullParamWithContentConfigDto> GetPullParamWithContentConfigAsync(int subSystemId, int serviceProviderId);
|
||||
Task<TrackPullParamWithContentConfigDto?>
|
||||
GetPullParamWithContentConfigAsync(int subSystemId, int serviceProviderId);
|
||||
|
||||
/// <summary>
|
||||
/// 设置拉取配置缓存
|
||||
@ -28,6 +29,6 @@ public interface ITrackPullCacheService
|
||||
/// <param name="subSystemId">子系统id</param>
|
||||
/// <param name="serviceProviderId">服务商id</param>
|
||||
/// <returns></returns>
|
||||
Task<TrackPullParamWithContentConfigDto> SetPullConfigAsync(int? id = null, int? subSystemId = null,
|
||||
Task<TrackPullParamWithContentConfigDto?> SetPullConfigAsync(int? id = null, int? subSystemId = null,
|
||||
int? serviceProviderId = null);
|
||||
}
|
||||
@ -8,6 +8,7 @@ using SJZY.InterfaceRelay.Application.Contracts;
|
||||
using SJZY.Track.Application.Contract.Dtos.TrackPull;
|
||||
using SJZY.Track.Application.Contract.Events;
|
||||
using SJZY.Track.Application.Contract.Services;
|
||||
using SJZY.Track.Application.Helpers;
|
||||
using SJZY.Track.Domain.Aggregates.OrderAggregate;
|
||||
using SJZY.Track.Domain.Consts;
|
||||
using SJZY.Track.Domain.Enums;
|
||||
@ -32,13 +33,14 @@ public class TrackRefreshEventHandler : ITransientDependency, ICapSubscribe
|
||||
[CapSubscribe(GlobalConst.TrackRefreshSubscribe, Group = GlobalConst.TrackRefreshSubscribe)]
|
||||
public async Task ConsumerAsync(InterfaceRelayForwardReplayEvent @event)
|
||||
{
|
||||
var attach = JsonConvert.DeserializeObject<TrackRefreshAttach>(@event.Attach);
|
||||
var attach = Base64Helper.DeSerialize<TrackRefreshAttach>(@event.Attach);
|
||||
if (attach is null) return;
|
||||
|
||||
var config = await _trackPullCacheService.GetPullParamWithContentConfigAsync(attach.SubSystemId,
|
||||
attach.ServiceProviderId);
|
||||
if (@event.Data == null || config == null) return;
|
||||
|
||||
if (@event.Data == null) return;
|
||||
var data = JObject.Parse(@event.Data.ToString()!);
|
||||
var data = JObject.Parse(@event.Data);
|
||||
var orderToken = data[config.ParamConfig.OrderPath];
|
||||
if (orderToken is JArray orderTokens)
|
||||
{
|
||||
|
||||
33
src/SJZY.Track.Application/Helpers/Base64Helper.cs
Normal file
33
src/SJZY.Track.Application/Helpers/Base64Helper.cs
Normal file
@ -0,0 +1,33 @@
|
||||
using System.Text;
|
||||
using Newtonsoft.Json;
|
||||
|
||||
namespace SJZY.Track.Application.Helpers;
|
||||
|
||||
public class Base64Helper
|
||||
{
|
||||
/// <summary>
|
||||
/// json序列化base64
|
||||
/// </summary>
|
||||
/// <param name="obj"></param>
|
||||
/// <returns></returns>
|
||||
public static string SerializeObject(object obj)
|
||||
{
|
||||
var json = JsonConvert.SerializeObject(obj);
|
||||
var bytes = Encoding.UTF8.GetBytes(json);
|
||||
|
||||
return Convert.ToBase64String(bytes);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// json反序列化base64
|
||||
/// </summary>
|
||||
/// <param name="json"></param>
|
||||
/// <typeparam name="T"></typeparam>
|
||||
/// <returns></returns>
|
||||
public static T? DeSerialize<T>(string json)
|
||||
{
|
||||
byte[] bytes = Convert.FromBase64String(json);
|
||||
|
||||
return JsonConvert.DeserializeObject<T>(Encoding.UTF8.GetString(bytes));
|
||||
}
|
||||
}
|
||||
@ -12,7 +12,7 @@
|
||||
</ItemGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<PackageReference Include="SJZY.InterfaceRelay.Application.Contracts" Version="7.1.6" />
|
||||
<PackageReference Include="SJZY.InterfaceRelay.Application.Contracts" Version="7.1.8"/>
|
||||
<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" />
|
||||
|
||||
@ -1,10 +1,10 @@
|
||||
using Microsoft.AspNetCore.Authorization;
|
||||
using Newtonsoft.Json;
|
||||
using SJZY.Expand.ABP.Core;
|
||||
using SJZY.InterfaceRelay.Application.Contracts;
|
||||
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.Domain.Consts;
|
||||
using SJZY.Track.Domain.Queries;
|
||||
using Volo.Abp.Http.Client.DynamicProxying;
|
||||
@ -51,18 +51,21 @@ public class TrackPullService : ApplicationService
|
||||
throw new BusinessException(message: "一个能更新的都没有");
|
||||
}
|
||||
|
||||
foreach (var group in orders.GroupBy(x =>
|
||||
new TrackRefreshAttach(x.ServiceProviderId, x.ServiceProviderCode, x.SubSystemId, x.OrderType)))
|
||||
|
||||
foreach (var group in orders.GroupBy(x => new
|
||||
{ x.ServiceProviderId, x.ServiceProviderName, x.SubSystemId, x.OrderType }))
|
||||
{
|
||||
var data = group.Select(x => new
|
||||
{
|
||||
x.OrderNo
|
||||
}).ToList();
|
||||
|
||||
var serviceProviderCode = group.FirstOrDefault()!.ServiceProviderCode;
|
||||
|
||||
var attach = Base64Helper.SerializeObject(group.Key);
|
||||
// 批量获取轨迹
|
||||
await _httpClientProxy.Service.BatchForwardThenPushToQueueAsync(
|
||||
"A012", group.Key.ServiceProviderCode, data,
|
||||
GlobalConst.TrackRefreshSubscribe, JsonConvert.SerializeObject(group.Key));
|
||||
"A012", serviceProviderCode, data, GlobalConst.TrackRefreshSubscribe, attach);
|
||||
}
|
||||
|
||||
return true;
|
||||
@ -87,6 +90,7 @@ public class TrackPullService : ApplicationService
|
||||
|
||||
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 }))
|
||||
@ -96,12 +100,13 @@ public class TrackPullService : ApplicationService
|
||||
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,
|
||||
JsonConvert.SerializeObject(new TrackRefreshAttach(config.ServiceProviderId, serviceProviderCode!,
|
||||
group.Key.SubSystemId, group.Key.OrderType)));
|
||||
GlobalConst.TrackRefreshSubscribe, attach);
|
||||
}
|
||||
|
||||
if (config.PullInterval < 60) config.PullInterval = 60;
|
||||
|
||||
@ -35,7 +35,7 @@ public class TrackPullCacheService : ApplicationService, ITrackPullCacheService
|
||||
}
|
||||
|
||||
[NonAction]
|
||||
public async Task<TrackPullParamWithContentConfigDto> GetPullParamWithContentConfigAsync(int subSystemId,
|
||||
public async Task<TrackPullParamWithContentConfigDto?> GetPullParamWithContentConfigAsync(int subSystemId,
|
||||
int serviceProviderId)
|
||||
{
|
||||
var data = await RedisHelper.Client.GetAsync<TrackPullParamWithContentConfigDto>(TrackPullContentKey(
|
||||
@ -48,23 +48,22 @@ public class TrackPullCacheService : ApplicationService, ITrackPullCacheService
|
||||
}
|
||||
|
||||
[NonAction]
|
||||
public async Task<TrackPullParamWithContentConfigDto> SetPullConfigAsync(int? id = null, int? subSystemId = null,
|
||||
public async Task<TrackPullParamWithContentConfigDto?> SetPullConfigAsync(int? id = null, int? subSystemId = null,
|
||||
int? serviceProviderId = null)
|
||||
{
|
||||
var data = new TrackPullParamWithContentConfigDto();
|
||||
|
||||
Expression<Func<TrackPullConfigEntity, bool>> exp = x => true;
|
||||
if (id.HasValue) exp = x => x.Id == id;
|
||||
else if (subSystemId.HasValue && serviceProviderId.HasValue)
|
||||
exp = x => x.SubSystemId == subSystemId && x.ServiceProviderId == serviceProviderId;
|
||||
|
||||
var trackPullConfigEntity = await _trackPullConfigRepository.GetFirstAsync(exp);
|
||||
if (trackPullConfigEntity == null) return data;
|
||||
if (trackPullConfigEntity == null) return null;
|
||||
|
||||
var trackPullConfigDetail = await _trackPullConfigRepository.GetDetailAsync(trackPullConfigEntity.Id);
|
||||
if (trackPullConfigDetail is null) return data;
|
||||
var trackPullContentList = await _trackPullConfigRepository.GetContentListCacheAsync(trackPullConfigEntity.Id);
|
||||
if (trackPullContentList.Count == 0) return data;
|
||||
if (trackPullContentList.Count == 0) return null;
|
||||
|
||||
var data = new TrackPullParamWithContentConfigDto();
|
||||
|
||||
data.ParamConfig =
|
||||
ObjectMapper.Map<GetTrackPullConfigDetailOutputVO, TrackPullParamConfigDto>(trackPullConfigDetail);
|
||||
|
||||
@ -86,7 +86,7 @@ namespace SJZY.Track.Application
|
||||
configure.Port = configuration.GetValue<int>("RabbitMQ:Port");
|
||||
});
|
||||
|
||||
options.TopicNamePrefix = "v1";
|
||||
// options.TopicNamePrefix = "v1";
|
||||
options.SucceedMessageExpiredAfter = 6 * 60 * 60;
|
||||
options.FailedRetryInterval = 60;
|
||||
options.FailedRetryCount = 3;
|
||||
@ -103,8 +103,8 @@ namespace SJZY.Track.Application
|
||||
LogHelper.Error($@"
|
||||
|-CAP失败回调事件:
|
||||
|-处理类型:{(faileCalback.MessageType == DotNetCore.CAP.Messages.MessageType.Subscribe ? "消费消息达到最大重试次数,需要开发人员介入" : "生产消息推送失败")}
|
||||
|-头部信息:{faileCalback?.Message?.Headers?.ToJson()}
|
||||
|-消息内容:{((JsonElement)faileCalback?.Message?.Value).GetRawText()}");
|
||||
|-头部信息:{faileCalback.Message.Headers.ToJson()}
|
||||
|-消息内容:{(faileCalback.Message.Value is string ? faileCalback.Message.Value : ((JsonElement?)faileCalback.Message.Value)?.GetRawText())}");
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
|
||||
@ -16,6 +16,11 @@ public class OrderWithServiceProviderVO
|
||||
/// </summary>
|
||||
public string ServiceProviderCode { get; set; } = null!;
|
||||
|
||||
/// <summary>
|
||||
/// 服务商name
|
||||
/// </summary>
|
||||
public string ServiceProviderName { get; set; } = null!;
|
||||
|
||||
/// <summary>
|
||||
/// 子系统id
|
||||
/// </summary>
|
||||
|
||||
@ -120,6 +120,7 @@ public class ProvideQueries : ISugarRepository, IProvideQueries,
|
||||
OrderType = a.OrderType,
|
||||
ServiceProviderId = b.ServiceProviderId,
|
||||
ServiceProviderCode = c.Code,
|
||||
ServiceProviderName = c.Name,
|
||||
SubSystemId = a.SubSystemId
|
||||
})
|
||||
.ToListAsync();
|
||||
@ -137,6 +138,7 @@ public class ProvideQueries : ISugarRepository, IProvideQueries,
|
||||
OrderType = a.OrderType,
|
||||
ServiceProviderId = b.ServiceProviderId,
|
||||
ServiceProviderCode = c.Code,
|
||||
ServiceProviderName = c.Name,
|
||||
SubSystemId = a.SubSystemId
|
||||
})
|
||||
.ToListAsync();
|
||||
|
||||
Loading…
Reference in New Issue
Block a user