Merge pull request 'S03-轨迹中台优化项' (#66) from S03-轨迹中台优化项 into uat

Reviewed-on: http://192.168.80.50:3000/Dotnet/SJZY.Track.Api/pulls/66
This commit is contained in:
lipengjun 2024-06-26 14:48:58 +08:00
commit 01cab202c8
78 changed files with 2379 additions and 1244 deletions

3
.gitignore vendored
View File

@ -6,6 +6,9 @@
*.user
*.sln.docstates
appsettings.json
logs/
# Build results
[Dd]ebug/
[Dd]ebugPublic/

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

View File

@ -1,6 +1,5 @@
<Project>
<PropertyGroup>
<LangVersion>latest</LangVersion>
<Nullable>enable</Nullable>
<ImplicitUsings>enable</ImplicitUsings>
<NoWarn>$(NoWarn);CS1591;CS0436;NU1803</NoWarn>

View File

@ -10,6 +10,7 @@ try
.AddConfigurationContainer()
.AddOptionsFromConfiguration()
.UseSerilogDefault()
.AddOTel() //注入OTel
.Host.AddAppSettingsSecretsJson()
.UseAutofac();
@ -17,6 +18,7 @@ try
var app = builder.Build();
await app.InitializeApplicationAsync();
app.UseOTel(); //启动WebApi的请求管道配置
await app.RunAsync();
}
catch (Exception ex)

View File

@ -3,27 +3,40 @@
"Logging": {
"LogLevel": {
"Default": "Debug",
"Microsoft.AspNetCore": "Warning"
"Microsoft.AspNetCore": "Debug"
}
},
"ConnectionStrings": {
"DefaultConnection": "server=mysql-dev.default.svc.cluster.local;Database=sjzy_track;Uid=sjzy_dev_user;Pwd=818f83e5;AllowLoadLocalInfile=true;SslMode=None;Charset=utf8mb4;",
"DefaultConnection": "server=mysql-dev.default.svc.cluster.local;Database=sjzy_track;Uid=sjzy_dev_user;Pwd=818f83e5;AllowLoadLocalInfile=true;SslMode=None;Charset=utf8mb4;"
},
"CAP": {
"FailedRetryCount": 0,
"FailedRetryInterval": 60,
"PrefetchCount": 1,
"ConsumerThreadCount": 2,
"UseDispatchingPerGroup": true,
"SucceedMessageExpiredAfter": 21600
},
"Redis": {
"Host": "redis-dev.default.svc.cluster.local:6379",
"Password": "gaVEHR3V7qiY",
"MaxPoolSize": 1000,
"Database": 6,
"Prefix": ""
"Prefix": "",
"IdleTimeout": 1000
},
"CorsPolicy": [
""
],
"Swagger": {
"IsEnable": true,
"ServiceUrl": "",
"JsonUrlPrefix": "", //转发目录前缀,部署服务器时需要配置 /kingmall
"ScanPrefix": [ "SJZY" ] //全局扫描程序集或xml文件时只加载以 SJZY 开头的内容
"ServiceUrl": "/trackApi",
"JsonUrlPrefix": "/trackApi",
//转发目录前缀,部署服务器时需要配置 /kingmall
"ScanPrefix": [
"SJZY"
]
//全局扫描程序集或xml文件时只加载以 SJZY 开头的内容
},
"RabbitMQ": {
"UserName": "sjzy",
@ -31,14 +44,18 @@
"HostName": "rabbitmq-dev.default.svc.cluster.local",
"Port": 5672
},
"Notice": {
"Webhook": "https://open.feishu.cn/open-apis/bot/v2/hook/fbe92f24-7f9f-41ea-a93a-63d855a03bc0",
"Title": "😭 轨迹中台开发环境异常"
},
"Serilog": {
"MinimumLevel": {
"Default": "Information",
"Default": "Debug",
"Override": {
"System": "Warning",
"Microsoft": "Warning",
"Microsoft.Hosting.Lifetime": "Information",
"Microsoft.EntityFrameworkCore": "Information"
"System": "Debug",
"Microsoft": "Debug",
"Microsoft.Hosting.Lifetime": "Debug",
"Microsoft.EntityFrameworkCore": "Debug"
}
},
"Enrich": [ "FromLogContext", "WithMachineName", "WithProcessId", "WithThreadId" ],
@ -66,13 +83,18 @@
"Base64PassPassword": "Basic ZWxhc3RpYzphZnlUZ21WYjhqcTY="
},
"RemoteServices": {
"UserCenter": {
"BaseUrl": "http://192.168.80.55:5020/"
},
"InterfaceForward": {
"BaseUrl": "http://192.168.80.55:5041/", //开发环境80.55,测试环境80.57
"AppKey": "从接口通应用获取",
"AppSecret": "从接口通应用获取"
"BaseUrl": "http://interfaceapi-test.default.svc.cluster.local:5041/",
"AppKey": "BSI2302001",
"AppSecret": "ArJYL14XZneXpqvpq80wyi3h2hT918AX"
},
"UserCenter": {
"BaseUrl": "http://userapi-dev.default.svc.cluster.local:5020/"
}
},
"Language": {
"SystemName": "Track",
"Index": 6,
"KeyPrefix": "language"
}
}

View File

@ -5,7 +5,7 @@
"Apollo": {
"AppId": "SJZY.Track.Configuration",
"Env": "DEV",
"Namespace": "Dev",
"Namespace": "Test",
"MetaServer": "http://apollo-service-test-apollo-configservice.apollo.svc.cluster.local:8080",
"ConfigServer": [ "http://apollo-service-test-apollo-configservice.apollo.svc.cluster.local:8080" ]
}

View File

@ -0,0 +1,60 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
namespace SJZY.Track.Application.Contract.Dtos.ProvideServices
{
/// <summary>
/// 查询轨迹推送映射关系入参
/// </summary>
public class QueryTrackPullMappingsInput
{
/// <summary>
/// 子系统id
/// </summary>
public int SubsystemsId { get; set; }
/// <summary>
/// 客户id
/// </summary>
public int CustomerId { get; set; }
}
/// <summary>
/// 查询轨迹推送映射关系返参项
/// </summary>
public class QueryTrackPullMappingsInfo
{
/// <summary>
/// 轨迹代码
/// </summary>
public string? Code { get; set; }
/// <summary>
/// 订单类型
/// </summary>
public int OrderType { get; set; }
/// <summary>
/// 服务商轨迹代码
/// </summary>
public string? ServiceCode { get; set; }
/// <summary>
/// 服务商轨迹状态
/// </summary>
public string? Status { get; set; }
/// <summary>
/// 服务商轨迹内容
/// </summary>
public string? Content { get; set; }
/// <summary>
/// 服务商轨迹内容英文
/// </summary>
public string? ContentEn { get; set; }
}
}

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

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

View File

@ -0,0 +1,10 @@
namespace SJZY.Track.Application.Contract.Events;
public struct OrderIdentifier
{
public int OrderId { get; set; }
public string OrderNo { get; set; }
public int OrderType { get; set; }
public int CustomId { get; set; }
}

View File

@ -1,28 +0,0 @@
namespace SJZY.Track.Application.Contract.Events;
public class OrderLinkAfterFillTrackEvent
{
/// <summary>
/// 关联订单id
/// </summary>
public int OrderId { get; set; }
/// <summary>
/// 子系统id
/// </summary>
public int SubSystemId { get; set; }
/// <summary>
/// 关联订单
/// </summary>
public List<LinkedOrder> LinkedOrders { get; set; } = null!;
}
public struct LinkedOrder
{
public int SubSystemId { get; set; }
public int OrderId { get; set; }
public string OrderNo { get; set; }
public int OrderType { get; set; }
}

View File

@ -5,23 +5,13 @@ public class TrackChangeAfterUpdateOrderEvent
/// <summary>
/// 订单Id
/// </summary>
public int OrderId { get; set; }
/// <summary>
/// 是否删除引起的更新
/// </summary>
public bool IsDelete { get; set; }
public int[] OrderIds { get; set; } = null!;
/// <summary>
/// 这条轨迹是否意味着完成(只有刷新会设为true)
/// </summary>
public bool IsCompleted { get; set; }
/// <summary>
/// 轨迹代码
/// </summary>
public string TrackCode { get; set; } = null!;
/// <summary>
/// 轨迹内容
/// </summary>

View File

@ -2,25 +2,20 @@
public class TrackDeleteEvent
{
/// <summary>
/// 订单id
/// </summary>
public int OrderId { get; set; }
/// <summary>
/// 轨迹代码
/// </summary>
public List<string> TrackCodes { get; set; } = null!;
/// <summary>
/// 子系统id
/// </summary>
public int SubSystemId { get; set; }
/// <summary>
/// 上游子系统id
/// 订单ids
/// </summary>
public int UpSubSystemId { get; set; }
public int[] OrderIds { get; set; } = null!;
/// <summary>
/// 轨迹代码
/// </summary>
public List<string> TrackCodes { get; set; } = null!;
/// <summary>
/// 操作人

View File

@ -7,35 +7,53 @@ public class TrackPushEvent
/// </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>
/// 轨迹发生时间
@ -46,9 +64,4 @@ public class TrackPushEvent
/// 轨迹发生时区
/// </summary>
public float TrackTimeZone { get; set; }
/// <summary>
/// 操作人
/// </summary>
public string? OperationMan { get; set; }
}

View File

@ -0,0 +1,25 @@
namespace SJZY.Track.Application.Contract.Events;
public struct TrackRefreshStruct
{
/// <summary>
/// 服务商
/// </summary>
public string? ServiceProviderName { get; set; }
/// <summary>
/// 原轨迹内容
/// </summary>
public string? OriginTrackContent { get; set; }
/// <summary>
/// 原轨迹报文
/// </summary>
public string? OriginTrackData { get; set; }
/// <summary>
/// 表示订单是否完成(只由轨迹刷新配置决定)
/// tips:能够使得IsCompleted为true的点,全局只有一个:轨迹刷新
/// </summary>
public bool IsCompleted { get; set; }
}

View File

@ -3,50 +3,14 @@
public class TrackUpsertEvent
{
/// <summary>
/// 订单id
/// </summary>
public int OrderId { get; set; }
/// <summary>
/// 订单类型
/// </summary>
public int OrderType { get; set; }
/// <summary>
/// 订单号
/// </summary>
public string OrderNo { get; set; } = null!;
/// <summary>
/// 服务商
/// </summary>
public string? ServiceProviderName { get; set; }
/// <summary>
/// 原轨迹内容
/// </summary>
public string? OriginTrackContent { get; set; }
/// <summary>
/// 原轨迹报文
/// </summary>
public string? OriginTrackData { get; set; }
/// <summary>
/// 表示订单是否完成(只由轨迹刷新配置决定)
/// tips:能够使得IsCompleted为true的点,全局只有一个:轨迹刷新
/// </summary>
public bool IsCompleted { get; set; }
/// <summary>
/// 子系统id
/// 子系统id(操作的订单都是相同子系统下的)
/// </summary>
public int SubSystemId { get; set; }
/// <summary>
/// 上游子系统id(0表示事件起点)
/// 订单id
/// </summary>
public int UpSubSystemId { get; set; }
public OrderIdentifier[] OrderIdentifiers { get; set; } = null!;
/// <summary>
/// 操作人(null则默认系统)
@ -54,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>
/// 轨迹内容代码
@ -82,4 +46,6 @@ public class TrackUpsertEvent
/// 已经处理过的订单
/// </summary>
public List<int> Visited { get; set; } = new();
public TrackRefreshStruct TrackRefreshData { get; set; }
}

View File

@ -0,0 +1,94 @@
namespace SJZY.Track.Application.Contract.Events;
public class TrackUpsertForRefreshEvent
{
/// <summary>
/// 子系统id(操作的订单都是相同子系统下的)
/// </summary>
public int SubSystemId { get; set; }
/// <summary>
/// 订单
/// </summary>
public OrderIdentifier OrderIdentifiers { get; set; }
/// <summary>
/// 服务商
/// </summary>
public string? ServiceProviderName { get; set; }
/// <summary>
/// 原轨迹内容
/// </summary>
public string? OriginTrackContent { get; set; }
/// <summary>
/// 原轨迹报文
/// </summary>
public string? OriginTrackData { get; set; }
/// <summary>
/// 表示订单是否完成(只由轨迹刷新配置决定)
/// tips:能够使得IsCompleted为true的点,全局只有一个:轨迹刷新
/// </summary>
public bool IsCompleted { get; set; }
/// <summary>
/// 轨迹内容代码
/// </summary>
public string TrackCode { get; set; } = null!;
/// <summary>
/// 轨迹发生地
/// </summary>
public string TrackPlace { get; set; } = null!;
/// <summary>
/// 轨迹发生时间
/// </summary>
public DateTime TrackTime { get; set; }
/// <summary>
/// 轨迹发生时区
/// </summary>
public float TrackTimeZone { get; set; }
}
public class TrackHandledDto
{
/// <summary>
/// 原轨迹内容
/// </summary>
public string? OriginTrackContent { get; set; }
/// <summary>
/// 原轨迹报文
/// </summary>
public string? OriginTrackData { get; set; }
/// <summary>
/// 表示订单是否完成(只由轨迹刷新配置决定)
/// tips:能够使得IsCompleted为true的点,全局只有一个:轨迹刷新
/// </summary>
public bool? IsCompleted { get; set; }
/// <summary>
/// 轨迹内容代码
/// </summary>
public string? TrackCode { get; set; }
/// <summary>
/// 轨迹发生地
/// </summary>
public string? TrackPlace { get; set; }
/// <summary>
/// 轨迹发生时间
/// </summary>
public DateTime? TrackTime { get; set; }
/// <summary>
/// 轨迹发生时区
/// </summary>
public float? TrackTimeZone { get; set; }
}

View File

@ -5,7 +5,7 @@
<PropertyGroup>
<TargetFramework>net7.0</TargetFramework>
<GenerateDocumentationFile>True</GenerateDocumentationFile>
<Version>7.0.0</Version>
<Version>7.0.2</Version>
<GenerateDocumentationFile>True</GenerateDocumentationFile>
</PropertyGroup>
<ItemGroup>

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

View File

@ -0,0 +1,20 @@
using SJZY.Track.Application.Contract.Dtos.ProvideServices;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using Volo.Abp;
namespace SJZY.Track.Application.Contract.Services
{
public interface ITrackPushConfigService : IRemoteService
{
/// <summary>
/// 查询轨迹推送配置关系
/// </summary>
/// <param name="input"></param>
/// <returns></returns>
Task<UnifyResultDto> QueryTrackPullMappingsAsync(QueryTrackPullMappingsInput input);
}
}

View File

@ -31,5 +31,5 @@ public interface ITrackService : IRemoteService
/// </summary>
/// <param name="input"></param>
/// <returns></returns>
Task<List<GetTrackOutput>> QueryAsync(GetTrackInput input);
Task<UnifyResultDto> QueryAsync(GetTrackInput input);
}

View File

@ -0,0 +1,22 @@
using AutoMapper;
using SJZY.Track.Application.Contract.Dtos.ProvideServices;
using SJZY.Track.Application.Contract.Dtos.TrackPush;
using SJZY.Track.Domain.Aggregates.TrackPushAggregate;
using SJZY.Track.Domain.Dtos.TrackPush;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
namespace SJZY.Track.Application.AutoMapper
{
public class TrackPushConfigAutoMapperProfile : Profile
{
public TrackPushConfigAutoMapperProfile()
{
CreateMap<QueryTrackPullMappingsInput, QueryTrackPullMappingsInputVO>(MemberList.None);
CreateMap<QueryTrackPullMappingsInfoVO, QueryTrackPullMappingsInfo>(MemberList.None);
}
}
}

View File

@ -0,0 +1,97 @@
using System.Security.Cryptography;
using System.Text;
using DotNetCore.CAP.Filter;
using SJZY.Expand.ABP.Core;
using SJZY.Expand.ABP.Core.Logger;
using SJZY.Expand.ABP.Core.PublicMethords;
using SJZY.Track.Domain.Consts;
namespace SJZY.Track.Application.CAP;
public class CapFilter : SubscribeFilter
{
const string ErrorCode = "Error599";
/// <summary>
/// 执行前
/// </summary>
/// <param name="context"></param>
/// <returns></returns>
public override async Task OnSubscribeExecutingAsync(ExecutingContext context)
{
// 轨迹新增、更新要做幂等
if (!context.DeliverMessage.Headers["cap-msg-name"]
.IsIn(GlobalConst.TrackUpsertSubscribe, GlobalConst.TrackManualPushSubscribe))
{
return;
}
var argumentJsonStr = context.Arguments.ToJson();
if (!string.IsNullOrEmpty(argumentJsonStr))
{
context.DeliverMessage.Headers.TryGetValue("sjzy-arguments-hash", out var hash);
if (hash.IsNullOrEmpty()) // 计算参数hash
{
hash = GetHash(argumentJsonStr);
context.DeliverMessage.Headers["sjzy-arguments-hash"] = hash;
}
if (await RedisHelper.Client.ExistsAsync($"CapExecuteFilter:{hash}"))
{
throw new BusinessException(ErrorCode, $"检测到重复消费,消费参数:{argumentJsonStr}");
}
}
}
public override async Task OnSubscribeExecutedAsync(ExecutedContext context)
{
context.DeliverMessage.Headers.TryGetValue("sjzy-arguments-hash", out var hash);
if (!hash.IsNullOrEmpty())
{
await RedisHelper.Client.SetAsync($"CapExecuteFilter:{hash}", 1, TimeSpan.FromMinutes(10).Seconds);
}
}
/// <summary>
/// 消费异常事件
/// </summary>
/// <param name="context"></param>
/// <returns></returns>
public override Task OnSubscribeExceptionAsync(ExceptionContext context)
{
context.DeliverMessage.Headers.TryGetValue("cap-msg-id", out var messageId);
if (context.Exception is BusinessException ex && ex.Code == ErrorCode)
{
context.ExceptionHandled = true; // 幂等不做重试
LogHelper.Warn(ex.ToString());
}
else
{
// 发飞书,异常继续上抛,进入重试机制
// tips:重试有冷却期(默认4分钟)
LogHelper.Error($"cap-msg-id:{messageId}\n{context.Exception}");
}
return Task.CompletedTask;
}
/// <summary>获取字符串的hash值</summary>
/// <param name="str"></param>
/// <returns></returns>
public static string GetHash(string str)
{
// 将字符串转换为字节
byte[] data = Encoding.UTF8.GetBytes(str);
// 创建SHA256实例
using SHA256 sha256 = SHA256.Create();
// 计算哈希值
byte[] hash = sha256.ComputeHash(data);
// 将哈希值转换为16进制字符串
return BitConverter.ToString(hash);
}
}

View File

@ -0,0 +1,71 @@
using DotNetCore.CAP;
using SJZY.Track.Application.Contract.Events;
using SJZY.Track.Domain.Aggregates.OrderAggregate;
using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate;
using SJZY.Track.Domain.Consts;
using Volo.Abp.DependencyInjection;
namespace SJZY.Track.Application.CAP.EventHandlers;
public class TrackChangeAfterUpdateOrderEventHandler : ITransientDependency, ICapSubscribe
{
private readonly IOrderRepository _orderRepository;
private readonly ITrackDetailsRepository _trackDetailsRepository;
public TrackChangeAfterUpdateOrderEventHandler(IOrderRepository orderRepository,
ITrackDetailsRepository trackDetailsRepository)
{
_orderRepository = orderRepository;
_trackDetailsRepository = trackDetailsRepository;
}
[CapSubscribe(GlobalConst.TrackChangeAfterUpdateOrderSubscribe)]
public async Task ConsumerAsync(TrackChangeAfterUpdateOrderEvent @event)
{
var orders = await _orderRepository.GetListAsync(x => @event.OrderIds.Contains(x.Id),
x => new OrderEntity
{
Id = x.Id,
IsCompleted = x.IsCompleted,
LatestTrackContent = x.LatestTrackContent,
LatestTrackEnContent = x.LatestTrackEnContent,
LatestTrackPlace = x.LatestTrackPlace,
LatestTrackTime = x.LatestTrackTime,
LatestTrackTimeZone = x.LatestTrackTimeZone,
});
if (orders.Count == 0) return;
var updateOrders = new List<OrderEntity>();
foreach (var order in orders)
{
if (order.LatestTrackContent != @event.TrackContent)
{
if (order.LatestTrackTime != null &&
@event.TrackTime.Subtract(TimeSpan.FromHours(@event.TrackTimeZone)) <
order.LatestTrackTime.Value.Subtract(TimeSpan.FromHours(order.LatestTrackTimeZone ?? 0)))
{
continue;
}
}
if (!order.IsCompleted)
{
order.IsCompleted = @event.IsCompleted;
}
order.LatestTrackContent = @event.TrackContent;
order.LatestTrackEnContent = @event.TrackEnContent;
order.LatestTrackPlace = @event.TrackPlace;
order.LatestTrackTime = @event.TrackTime;
order.LatestTrackTimeZone = @event.TrackTimeZone;
updateOrders.Add(order);
}
await _orderRepository.UpdateRangeAsync(updateOrders, x => new
{
x.IsCompleted, x.LatestTrackContent, x.LatestTrackEnContent, x.LatestTrackPlace, x.LatestTrackTime,
x.LatestTrackTimeZone
});
}
}

View File

@ -0,0 +1,135 @@
using DotNetCore.CAP;
using SJZY.Expand.ABP.Core.Logger;
using SJZY.Expand.ABP.Core.PublicMethords;
using SJZY.Track.Application.Contract.Events;
using SJZY.Track.Domain.Aggregates.OrderAggregate;
using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate;
using SJZY.Track.Domain.Consts;
using SJZY.Track.Domain.Queries;
using Volo.Abp.DependencyInjection;
namespace SJZY.Track.Application.CAP.EventHandlers;
public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe
{
private readonly IOrderRepository _orderRepository;
private readonly IProvideQueries _provideQueries;
private readonly ICapPublisher _capPublisher;
private readonly ITrackDetailsRepository _trackDetailsRepository;
public TrackDeleteEventHandler(IProvideQueries provideQueries,
IOrderRepository orderRepository, ICapPublisher capPublisher, ITrackDetailsRepository trackDetailsRepository)
{
_provideQueries = provideQueries;
_orderRepository = orderRepository;
_capPublisher = capPublisher;
_trackDetailsRepository = trackDetailsRepository;
}
[CapSubscribe(GlobalConst.TrackDeleteSubscribe, Group = GlobalConst.TrackDeleteSubscribe)]
public async Task ConsumerAsync(TrackDeleteEvent @event)
{
var dup = @event.Visited.Intersect(@event.OrderIds).ToArray();
if (dup.Any())
{
LogHelper.Error($"删除轨迹时,发现循环依赖:{dup.JoinAsString(",")}");
return;
}
await DeleteTrackAsync(@event);
// 根据订单关联关系异步向下刷(异步)
var linkedOrders = await _orderRepository.GetLinkedOrdersAsync(@event.OrderIds);
foreach (var orderId in @event.OrderIds)
{
foreach (var group in linkedOrders.Where(x => x.OwnerOrderId == orderId)
.GroupBy(x => x.SubSystemId))
{
var nextEvent = @event.DeepCopy();
nextEvent.Visited.AddRange(@event.OrderIds);
nextEvent.TrackCodes = @event.TrackCodes;
nextEvent.SubSystemId = group.Key;
// 不同子系统需要做内容映射
if (@event.SubSystemId != nextEvent.SubSystemId)
{
var mapCodes = await _provideQueries
.MapTrackCodesAsync(@event.SubSystemId, nextEvent.SubSystemId, @event.TrackCodes.ToArray());
if (mapCodes.Count == 0)
{
//throw new BusinessException(message: "订单跨子系统,未配置内容映射,请联系管理员");
continue;
}
nextEvent.TrackCodes = mapCodes;
}
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);
}
}
}
}
private async Task DeleteTrackAsync(TrackDeleteEvent @event)
{
// 删除轨迹(删的是同子系统下的单,所以code是统一的)
var tracks = await _trackDetailsRepository.GetLatestListAsync(@event.OrderIds, @event.TrackCodes);
await _trackDetailsRepository.DeleteTrackAsync(@event.OrderIds, @event.TrackCodes);
var orders = await _orderRepository.GetListAsync(x => @event.OrderIds.Contains(x.Id),
x => new OrderEntity
{
Id = x.Id,
LatestTrackContent = x.LatestTrackContent,
LatestTrackEnContent = x.LatestTrackEnContent,
LatestTrackPlace = x.LatestTrackPlace,
LatestTrackTime = x.LatestTrackTime,
LatestTrackTimeZone = x.LatestTrackTimeZone,
});
if (orders.Count == 0) return;
foreach (var order in orders)
{
var track = tracks.FirstOrDefault(x => x.OrderId == order.Id);
// 删除全部轨迹导致的重置行为
if (track == default)
{
if (order.LatestTrackTime == null) continue;
// tips:订单是支持从已完成态到未完成态的
order.ResetTrackInfo();
}
else
{
if (order.LatestTrackContent == track.TrackContent) continue;
// 删除引发的回退更新
order.LatestTrackContent = track.TrackContent;
order.LatestTrackEnContent = track.TrackEnContent;
order.LatestTrackPlace = track.TrackPlace;
order.LatestTrackTime = track.TrackTime;
order.LatestTrackTimeZone = track.TrackTimeZone;
}
}
await _orderRepository.UpdateRangeAsync(orders, x => new
{
x.LatestTrackContent, x.LatestTrackEnContent, x.LatestTrackPlace, x.LatestTrackTime,
x.LatestTrackTimeZone
});
}
}

View File

@ -4,7 +4,7 @@ using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate;
using SJZY.Track.Domain.Consts;
using Volo.Abp.DependencyInjection;
namespace SJZY.Track.Application.EventHandlers;
namespace SJZY.Track.Application.CAP.EventHandlers;
public class TrackInsertForNotMappedEventHandler : ITransientDependency, ICapSubscribe
{
@ -20,7 +20,7 @@ public class TrackInsertForNotMappedEventHandler : ITransientDependency, ICapSub
{
if (await _trackDetailsRepository.AnyAsync(@event.OrderId, @event.TrackTime))
{
return; // 无法映射的轨迹只做新增
return;
}
// 新增服务商轨迹到es
@ -45,6 +45,6 @@ public class TrackInsertForNotMappedEventHandler : ITransientDependency, ICapSub
track.TrackEnContent = GlobalConst.EmptyString;
track.ZeroTrackTime = track.TrackTime.Subtract(TimeSpan.FromHours(track.TrackTimeZone));
await _trackDetailsRepository.UpsertAsync(track);
await _trackDetailsRepository.AddAsync(track);
}
}

View File

@ -0,0 +1,101 @@
using DotNetCore.CAP;
using Newtonsoft.Json;
using SJZY.InterfaceRelay.Application.Contracts;
using SJZY.Track.Application.Contract.Dtos.TrackPush;
using SJZY.Track.Application.Helpers;
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.CAP.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)
{
string logStr = null;
var optionType = Base64Helper.DeSerialize<string>(@event.Attach);
if (optionType == OptionType.新增.ToString())
{
logStr = JsonConvert.DeserializeObject<TrackPushInput>(@event.RequestParams)!.Log;
}
else if (optionType == OptionType.修改.ToString())
{
logStr = JsonConvert.DeserializeObject<TrackUpdateInput>(@event.RequestParams)!.Log;
}
if (logStr.IsNullOrEmpty()) throw new InvalidOperationException();
var log = JsonConvert.DeserializeObject<TrackPushRecordIndex>(logStr!);
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)
{
// tips:只有新增会影响推送进度?
if (log.OptionType != OptionType.新增.ToString()) return;
var order = new OrderEntity
{
Id = log.OrderId
};
/*
* 异常:最近一条推送是失败
*/
if (!log.IsSuccess)
{
order.QuestionType = (int)TrackProblemType.Error;
order.LatestPushError = log.Message;
await _orderRepository.UpdateColumnAsync(order, x => new
{
x.QuestionType, x.LatestPushError
});
}
else
{
// 还原成正常态
order.QuestionType = (int)TrackProblemType.None;
order.LatestPushError = null;
order.LatestPushTrackCode = log.TrackCode;
order.LatestPushTime = log.CreateTime;
await _orderRepository.UpdateColumnAsync(order, x => new
{
x.LatestPushTrackCode, x.LatestPushTime, x.QuestionType, x.LatestPushError
});
}
}
}

View File

@ -0,0 +1,290 @@
using DotNetCore.CAP;
using Newtonsoft.Json;
using SJZY.Expand.ABP.Core.PublicMethords;
using SJZY.InterfaceRelay.Application.Contracts;
using SJZY.Track.Application.Contract.Dtos.TrackPush;
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.Aggregates.TrackDetailsAggregate;
using SJZY.Track.Domain.Aggregates.TrackPushAggregate;
using SJZY.Track.Domain.Consts;
using SJZY.Track.Domain.Enums;
using Volo.Abp.DependencyInjection;
using Volo.Abp.Http.Client.DynamicProxying;
namespace SJZY.Track.Application.CAP.EventHandlers;
public class TrackPushEventHandler : ITransientDependency, ICapSubscribe
{
private readonly IHttpClientProxy<IInterfaceForwardServiceV2> _httpClientProxy;
private readonly IOrderRepository _orderRepository;
private readonly ITrackPushCacheService _trackPushCacheService;
private readonly ITrackDetailsRepository _trackDetailsRepository;
private readonly ITrackPushRecordsRepository _trackPushRecordsRepository;
private List<TrackPushContentConfigDto> _pushConfigs;
private List<OrderEntity> _orders;
public TrackPushEventHandler(IHttpClientProxy<IInterfaceForwardServiceV2> httpClientProxy,
IOrderRepository orderRepository, ITrackPushCacheService trackPushCacheService,
ITrackDetailsRepository trackDetailsRepository, ITrackPushRecordsRepository trackPushRecordsRepository)
{
_httpClientProxy = httpClientProxy;
_orderRepository = orderRepository;
_trackPushCacheService = trackPushCacheService;
_trackDetailsRepository = trackDetailsRepository;
_trackPushRecordsRepository = trackPushRecordsRepository;
}
[CapSubscribe(GlobalConst.TrackPushSubscribe, Group = GlobalConst.TrackPushSubscribe)]
[CapSubscribe(GlobalConst.TrackManualPushSubscribe, Group = GlobalConst.TrackPushSubscribe)]
public async Task ConsumerAsync(TrackPushEvent @event)
{
var pushConfigs = await _trackPushCacheService.GetPushContentConfigsAsync(@event.SubSystemId, @event.CustomId);
if (pushConfigs == null || pushConfigs.Count == 0) return;
_pushConfigs = pushConfigs;
var orders = await _orderRepository.GetListAsync(x =>
@event.OrderIds.Contains(x.Id), x => new OrderEntity
{
Id = x.Id, OrderNo = x.OrderNo, LatestPushTrackCode = x.LatestPushTrackCode,
OrderType = x.OrderType
});
if (orders.Count == 0) return;
_orders = orders;
switch (@event.OptionType)
{
case OptionType.无:
return;
case OptionType.新增:
await DoPushAsync(@event);
break;
case OptionType.删除:
break;
case OptionType.修改:
await DoUpdateAsync(@event);
break;
default:
throw new ArgumentOutOfRangeException();
}
}
private async Task DoPushAsync(TrackPushEvent @event)
{
var items = new List<TrackPushInput>();
TrackPushContentConfigDto? config;
if (@event.OperationMan == null)
{
foreach (var order in _orders)
{
if (order.QuestionType == (int)TrackProblemType.Error) continue; // 系统不推异常单,只能人工操作
// 推送新增,需要根据上一次的推送 推断 本次推送 是否可行
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
{
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,
};
var item = await BuiltPushItem(config, log);
item.Log = JsonConvert.SerializeObject(log);
items.Add(item);
}
}
else //手工推送
{
config = _pushConfigs.FirstOrDefault(x => x.TrackCode == @event.TrackCode);
if (config == null) return;
var tracks = await _trackDetailsRepository.GetListAsync(@event.OrderIds, @event.TrackCode);
tracks = tracks.DistinctBy(x => x.OrderId).ToList(); // 防止重复推送
if (tracks.Count == 0) return; // 一个能推的轨迹都不存在
// tips:人工推送只要能推就推了,不会校验已推和顺序问题,因为可能是更新
foreach (var track in tracks)
{
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,
};
var item = await BuiltPushItem(config, log);
item.Log = JsonConvert.SerializeObject(log);
items.Add(item);
}
}
if (items.Count == 0) return;
string asciiString = Base64Helper.SerializeObject(OptionType.新增.ToString());
var r = await _httpClientProxy.Service.BatchForwardThenPushToQueueAsync(
InterfaceRelayConst.PushTrackInterfaceCode,
_pushConfigs.First().ServiceProviderCode, items, GlobalConst.TrackPushCompletedSubscribe,
attach: asciiString);
if (r.Code != InterfaceRelayConst.SuccessCode)
{
throw new BusinessException(message: $"requestId: {r.RequestId}; {r.Msg}");
}
}
private async Task DoUpdateAsync(TrackPushEvent @event)
{
var config = _pushConfigs.FirstOrDefault(x => x.TrackCode == @event.TrackCode);
if (config == null || !config.IsCanUpdate) return;
var items = new List<TrackUpdateInput>();
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,
};
var item = BuiltUpdateItem(config, log);
item.Log = JsonConvert.SerializeObject(log);
items.Add(item);
}
if (items.Count == 0) return;
string asciiString = Base64Helper.SerializeObject(OptionType.修改.ToString());
var r = await _httpClientProxy.Service.BatchForwardThenPushToQueueAsync(
InterfaceRelayConst.TrackUpdateInterfaceCode,
_pushConfigs.First().ServiceProviderCode, items, GlobalConst.TrackPushCompletedSubscribe,
attach: asciiString);
if (r.Code != InterfaceRelayConst.SuccessCode)
{
throw new BusinessException(message: $"requestId: {r.RequestId}; {r.Msg}");
}
}
private TrackUpdateInput BuiltUpdateItem(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 TrackUpdateInput
{
OrderNo = log.OrderNo,
};
body.Data.Add(track);
log.PushContent = body.ToJson();
return body;
}
private async Task<TrackPushInput> 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();
return body;
}
/// <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;
}
}

View File

@ -9,29 +9,34 @@ 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.Aggregates.TrackDetailsAggregate;
using SJZY.Track.Domain.Consts;
using SJZY.Track.Domain.Enums;
using Volo.Abp.DependencyInjection;
namespace SJZY.Track.Application.EventHandlers;
namespace SJZY.Track.Application.CAP.EventHandlers;
public class TrackRefreshEventHandler : ITransientDependency, ICapSubscribe
{
private readonly IOrderRepository _orderRepository;
private readonly ICapPublisher _capPublisher;
private readonly ITrackPullCacheService _trackPullCacheService;
private readonly ITrackDetailsRepository _trackDetailsRepository;
public TrackRefreshEventHandler(IOrderRepository orderRepository,
ICapPublisher capPublisher, ITrackPullCacheService trackPullCacheService)
ICapPublisher capPublisher, ITrackPullCacheService trackPullCacheService,
ITrackDetailsRepository trackDetailsRepository)
{
_orderRepository = orderRepository;
_capPublisher = capPublisher;
_trackPullCacheService = trackPullCacheService;
_trackDetailsRepository = trackDetailsRepository;
}
[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;
@ -89,15 +94,19 @@ public class TrackRefreshEventHandler : ITransientDependency, ICapSubscribe
await HandleTrackAsync(item, paramWithContentConfig, context);
}
}
else if (orderToken is JObject)
else if (trackToken is JObject)
{
await HandleTrackAsync(orderToken, paramWithContentConfig, context);
await HandleTrackAsync(trackToken, paramWithContentConfig, context);
}
}
private async Task HandleTrackAsync(JToken trackToken, TrackPullParamWithContentConfigDto paramWithContentConfig,
TrackRefreshContext context)
{
var originTrackData = JsonConvert.SerializeObject(trackToken);
// 如果已经存在一模一样的报文了说明处理过了,直接return
if (await _trackDetailsRepository.AnyAsync(context.OrderId, originTrackData)) return;
var paramConfig = paramWithContentConfig.ParamConfig;
var contentConfigs = paramWithContentConfig.ContentConfigs;
TrackPullContentConfigDto? matchedContentConfig = null;
@ -214,23 +223,34 @@ public class TrackRefreshEventHandler : ITransientDependency, ICapSubscribe
// 新增/更新轨迹
var trackUpsertEvent = new TrackUpsertEvent
{
OrderId = context.OrderId,
OrderType = context.OrderType,
OrderNo = context.OrderNo,
ServiceProviderName = context.ServiceProviderName,
OriginTrackData = JsonConvert.SerializeObject(trackToken),
IsCompleted = matchedContentConfig.IsCompleted,
OrderIdentifiers = new[]
{
new OrderIdentifier
{
OrderId = context.OrderId,
OrderNo = context.OrderNo,
OrderType = context.OrderType,
CustomId = context.CustomId
}
},
TrackRefreshData = new TrackRefreshStruct
{
ServiceProviderName = context.ServiceProviderName,
OriginTrackData = originTrackData,
IsCompleted = matchedContentConfig.IsCompleted,
OriginTrackContent = paramConfig.TrackContentName != null
? trackToken[paramConfig.TrackContentName]?.Value<string>()
: GlobalConst.EmptyString
},
SubSystemId = context.SubSystemId,
TrackCode = matchedContentConfig.TrackContentCode,
TrackPlace = trackPlace ?? GlobalConst.EmptyString,
TrackTime = trackTime ?? DateTime.Now,
TrackTimeZone = trackTimeZone ?? matchedContentConfig.TrackTimeZoneValue ?? 8
};
if (paramConfig.TrackContentName != null)
{
trackUpsertEvent.OriginTrackContent = trackToken[paramConfig.TrackContentName]?.Value<string>();
}
await _capPublisher.PublishAsync(GlobalConst.TrackUpsertSubscribe, trackUpsertEvent);
trackUpsertEvent.IsAdd =
!await _trackDetailsRepository.AnyByCodeAsync(context.OrderId, matchedContentConfig.TrackContentCode);
await _capPublisher.PublishAsync(GlobalConst.TrackUpsertForRefreshSubscribe, trackUpsertEvent);
}
}

View File

@ -0,0 +1,215 @@
using DotNetCore.CAP;
using SJZY.Expand.ABP.Core.Logger;
using SJZY.Expand.ABP.Core.PublicMethords;
using SJZY.Track.Application.Contract.Events;
using SJZY.Track.Domain.Aggregates.OrderAggregate;
using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate;
using SJZY.Track.Domain.Consts;
using SJZY.Track.Domain.Queries;
using Volo.Abp.DependencyInjection;
namespace SJZY.Track.Application.CAP.EventHandlers;
public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe
{
private readonly IOrderRepository _orderRepository;
private readonly IProvideQueries _provideQueries;
private readonly ITrackDetailsRepository _trackDetailsRepository;
private readonly ICapPublisher _capPublisher;
public TrackUpsertEventHandler(IOrderRepository orderRepository, ICapPublisher capPublisher,
IProvideQueries provideQueries, ITrackDetailsRepository trackDetailsRepository)
{
_orderRepository = orderRepository;
_capPublisher = capPublisher;
_provideQueries = provideQueries;
_trackDetailsRepository = trackDetailsRepository;
}
[CapSubscribe(GlobalConst.TrackUpsertSubscribe, Group = GlobalConst.TrackUpsertSubscribe)]
[CapSubscribe(GlobalConst.TrackUpsertForRefreshSubscribe, Group = GlobalConst.TrackUpsertForRefreshSubscribe)]
public async Task ConsumerAsync(TrackUpsertEvent @event)
{
var orderIds = @event.OrderIdentifiers.Select(x => x.OrderId).ToArray();
var dup = @event.Visited.Intersect(orderIds).ToArray();
if (dup.Any())
{
LogHelper.Error($"{(@event.IsAdd ? "添加" : "修改")}轨迹时,发现循环依赖:{dup.JoinAsString(",")}");
return;
}
var canNextOrderIds = new List<int>();
await HandleTrackAsync(@event, canNextOrderIds);
// 根据订单关联关系异步向下刷(异步)
var linkedOrders = await _orderRepository.GetLinkedOrdersAsync(canNextOrderIds.ToArray());
foreach (var order in @event.OrderIdentifiers)
{
// 理论上小包在这不会分组
foreach (var group in linkedOrders.Where(x => x.OwnerOrderId == order.OrderId)
.GroupBy(x => new { x.SubSystemId, x.CustomId }))
{
var nextEvent = @event.DeepCopy();
nextEvent.Visited.AddRange(orderIds);
nextEvent.SubSystemId = group.Key.SubSystemId;
nextEvent.TrackRefreshData = new TrackRefreshStruct
{
ServiceProviderName = @event.TrackRefreshData.ServiceProviderName,
OriginTrackContent = @event.TrackRefreshData.OriginTrackContent,
OriginTrackData = @event.TrackRefreshData.OriginTrackData,
// tips:一定要重置为false,父级订单的完成不会使得关联订单完成,关联单也必须由刷新来完成或手动设置完成!!
IsCompleted = false
};
// 不同子系统需要做内容映射
if (@event.SubSystemId != nextEvent.SubSystemId)
{
var mapCode = (await _provideQueries
.MapTrackCodesAsync(@event.SubSystemId, nextEvent.SubSystemId, @event.TrackCode))
.FirstOrDefault();
if (mapCode == default)
{
// 跨子系统没做映射的轨迹不处理
continue;
}
nextEvent.TrackCode = mapCode;
}
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);
}
}
}
}
private async Task HandleTrackAsync(TrackUpsertEvent @event, List<int> canNextOrderIds)
{
var orderIds = @event.OrderIdentifiers.Select(x => x.OrderId).ToArray();
var tracks = await _trackDetailsRepository.GetListAsync(orderIds, @event.TrackCode, true);
var newOrderIds = orderIds.Except(tracks.Select(x => x.OrderId)).ToArray();
var trackContents = await _provideQueries.MatchTrackContentAsync(newOrderIds, @event.TrackCode);
var insertTracks = new List<TrackDetailsIndex>();
foreach (var order in @event.OrderIdentifiers)
{
var track = tracks.FirstOrDefault(x => x.OrderId == order.OrderId);
if (track == null && !@event.IsAdd) continue; // 不存在不让改
if (track != null && @event.IsAdd) continue; // 存在不让创建
if (track == default)
{
track = new TrackDetailsIndex
{
OrderId = order.OrderId,
OrderNo = order.OrderNo,
OrderType = order.OrderType,
TrackCode = @event.TrackCode,
TrackPlace = @event.TrackPlace,
TrackTime = @event.TrackTime,
TrackTimeZone = @event.TrackTimeZone,
ServiceProviderName = @event.TrackRefreshData.ServiceProviderName ?? GlobalConst.EmptyString,
OriginTrackContent = @event.TrackRefreshData.OriginTrackContent ?? GlobalConst.EmptyString,
OriginTrackData = @event.TrackRefreshData.OriginTrackData ?? GlobalConst.EmptyString,
IsCanDelete = false,
CreateUser = @event.OperationMan ?? GlobalConst.SystemUser,
CreateTime = DateTime.Now
};
// 根据轨迹code在订单模板内容中寻找匹配的
var trackContent = trackContents.FirstOrDefault(x => x.OrderId == order.OrderId);
if (trackContent == default)
{
// 订单所绑定的模板,没有一条内容可以匹配发生的轨迹
continue;
}
track.TrackStatus = trackContent.Status;
track.TrackCode = trackContent.TrackCode;
track.TrackContent = trackContent.Content;
track.TrackEnContent = trackContent.EnContent;
track.IsCanDelete = trackContent.IsCanDelete;
track.ZeroTrackTime = track.TrackTime.Subtract(TimeSpan.FromHours(track.TrackTimeZone));
insertTracks.Add(track);
}
else
{
track.TrackPlace = @event.TrackPlace;
track.TrackTime = @event.TrackTime;
track.TrackTimeZone = @event.TrackTimeZone;
track.UpdateUser = @event.OperationMan ?? GlobalConst.SystemUser;
track.UpdateTime = DateTime.Now;
await _trackDetailsRepository.UpdateAsync(track);
}
canNextOrderIds.Add(order.OrderId);
}
if (canNextOrderIds.Count == 0) return;
var content = await _provideQueries.GetTrackContentAsync(@event.SubSystemId, @event.TrackCode);
// 批量add to es
await _trackDetailsRepository.AddManyAsync(insertTracks);
// 触发订单主表更新事件(异步)
await _capPublisher.PublishAsync(GlobalConst.TrackChangeAfterUpdateOrderSubscribe,
new TrackChangeAfterUpdateOrderEvent
{
OrderIds = canNextOrderIds.ToArray(),
IsCompleted = @event.TrackRefreshData.IsCompleted,
TrackContent = content.TrackContent,
TrackEnContent = content.TrackEnContent,
TrackPlace = @event.TrackPlace,
TrackTime = @event.TrackTime,
TrackTimeZone = @event.TrackTimeZone
});
// 触发轨迹推送事件(异步)
foreach (var group in @event.OrderIdentifiers
.Where(x => x.OrderId.IsIn(canNextOrderIds))
.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 = content.TrackStatus,
TrackContent = content.TrackContent,
TrackPlace = @event.TrackPlace,
TrackTime = @event.TrackTime,
TrackTimeZone = @event.TrackTimeZone
}
};
await _capPublisher.PublishAsync(GlobalConst.TrackPushSubscribe, trackPushEvent);
}
}
}

View File

@ -0,0 +1,89 @@
using Microsoft.Extensions.Logging;
namespace SJZY.Track.Application.CAP;
internal static class LoggerExtensions
{
public static void ConsumerExecutedAfterThreshold(this ILogger logger, string messageId, int retries)
{
logger.LogWarning(
$"The Subscriber of the message({messageId}) still fails after {retries}th executions and we will stop retrying.");
}
public static void SenderAfterThreshold(this ILogger logger, string messageId, int retries)
{
logger.LogWarning(
$"The Publisher of the message({messageId}) still fails after {retries}th sends and we will stop retrying.");
}
public static void ExecutedThresholdCallbackFailed(this ILogger logger, Exception ex)
{
logger.LogWarning(ex, "FailedThresholdCallback action raised an exception:" + ex.Message);
}
public static void ConsumerDuplicates(this ILogger logger, string subscriber, string group)
{
logger.LogWarning(
$"We detected that you have duplicate subscribers ({subscriber}) in same group ({group}), this will cause diversity behavior.");
}
public static void ConsumerExecutionRetrying(this ILogger logger, string messageId, int retries)
{
logger.LogWarning($"The {retries}th retrying consume a message failed. message id: {messageId}");
}
public static void SenderRetrying(this ILogger logger, string messageId, int retries)
{
logger.LogWarning($"The {retries}th retrying send a message failed. message id: {messageId} ");
}
public static void MessageReceived(this ILogger logger, string messageId, string name)
{
logger.LogDebug($"Received message. id:{messageId}, name: {name}");
}
public static void MessagePublishException(this ILogger logger, string? messageId, string reason, Exception? ex)
{
logger.LogError(ex,
$"An exception occurred while publishing a message, reason:{reason}. message id:{messageId}");
}
public static void ConsumerExecuting(this ILogger logger, string className, string methodName, string group)
{
logger.LogInformation($"Executing subscriber method '{className}.{methodName}' on group '{group}'");
}
public static void ConsumerExecuted(this ILogger logger, string className, string methodName, string group,
double milliseconds, string? instance)
{
logger.LogInformation(
$"Executed subscriber method '{className}.{methodName}' on group '{group}' with instance '{instance}' in {milliseconds}ms");
}
public static void ConsumerExecuteFailed(this ILogger logger, string topic, string id, string? instance,
Exception? ex)
{
logger.LogError(ex,
$"An exception occurred while executing the subscription method. Topic:{topic}, Id:{id}, Instance: {instance}");
}
public static void ServerStarting(this ILogger logger)
{
logger.LogInformation("Starting the processing server.");
}
public static void ProcessorsStartedError(this ILogger logger, Exception ex)
{
logger.LogError(ex, "Starting the processors throw an exception.");
}
public static void ServerShuttingDown(this ILogger logger)
{
logger.LogInformation("Shutting down the processing server...");
}
public static void ExpectedOperationCanceledException(this ILogger logger, Exception ex)
{
logger.LogWarning(ex, $"Expected an OperationCanceledException, but found '{ex.Message}'.");
}
}

View File

@ -0,0 +1,268 @@
using System.Diagnostics;
using System.Reflection;
using DotNetCore.CAP;
using DotNetCore.CAP.Diagnostics;
using DotNetCore.CAP.Internal;
using DotNetCore.CAP.Messages;
using DotNetCore.CAP.Persistence;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
namespace SJZY.Track.Application.CAP;
public class MySubscribeExecutor : ISubscribeExecutor
{
// diagnostics listener
// ReSharper disable once InconsistentNaming
private static readonly DiagnosticListener s_diagnosticListener =
new(CapDiagnosticListenerNames.DiagnosticListenerName);
private readonly IDataStorage _dataStorage;
private readonly string? _hostName;
private readonly ILogger _logger;
private readonly CapOptions _options;
private readonly IServiceProvider _provider;
public MySubscribeExecutor(
ILogger<MySubscribeExecutor> logger,
IOptions<CapOptions> options,
IServiceProvider provider)
{
_provider = provider;
_logger = logger;
_options = options.Value;
_dataStorage = _provider.GetRequiredService<IDataStorage>();
Invoker = _provider.GetRequiredService<ISubscribeInvoker>();
_hostName = Helper.GetInstanceHostname();
}
private ISubscribeInvoker Invoker { get; }
public async Task<OperateResult> ExecuteAsync(MediumMessage message, ConsumerExecutorDescriptor? descriptor = null,
CancellationToken cancellationToken = default)
{
if (descriptor == null)
{
var selector = _provider.GetRequiredService<MethodMatcherCache>();
if (!selector.TryGetTopicExecutor(message.Origin.GetName(), message.Origin.GetGroup()!, out descriptor))
{
var error =
$"Message (Name:{message.Origin.GetName()},Group:{message.Origin.GetGroup()}) can not be found subscriber." +
$"{Environment.NewLine} see: https://github.com/dotnetcore/CAP/issues/63";
_logger.LogError(error);
TracingError(DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(), message.Origin, null,
new Exception(error));
var ex = new SubscriberNotFoundException(error);
await SetFailedState(message, ex);
return OperateResult.Failed(ex);
}
}
bool retry;
OperateResult result;
//record instance id
message.Origin.Headers[Headers.ExecutionInstanceId] = _hostName;
do
{
var (shouldRetry, operateResult) = await ExecuteWithoutRetryAsync(message, descriptor, cancellationToken)
.ConfigureAwait(false);
result = operateResult;
if (result.Equals(OperateResult.Success)) return result;
retry = shouldRetry;
} while (retry);
return result;
}
private async Task<(bool, OperateResult)> ExecuteWithoutRetryAsync(MediumMessage message,
ConsumerExecutorDescriptor descriptor, CancellationToken cancellationToken)
{
if (message == null) throw new ArgumentNullException(nameof(message));
cancellationToken.ThrowIfCancellationRequested();
try
{
_logger.ConsumerExecuting(descriptor.ImplTypeInfo.Name, descriptor.MethodInfo.Name,
descriptor.Attribute.Group);
var sp = Stopwatch.StartNew();
await InvokeConsumerMethodAsync(message, descriptor, cancellationToken).ConfigureAwait(false);
sp.Stop();
await SetSuccessfulState(message).ConfigureAwait(false);
CapEventCounterSource.Log.WriteInvokeTimeMetrics(sp.Elapsed.TotalMilliseconds);
_logger.ConsumerExecuted(descriptor.ImplTypeInfo.Name, descriptor.MethodInfo.Name,
descriptor.Attribute.Group, sp.Elapsed.TotalMilliseconds, message.Origin.GetExecutionInstanceId());
return (false, OperateResult.Success);
}
catch (Exception ex)
{
_logger.ConsumerExecuteFailed(message.Origin.GetName(), message.DbId,
message.Origin.GetExecutionInstanceId(), ex);
return (await SetFailedState(message, ex).ConfigureAwait(false), OperateResult.Failed(ex));
}
}
private Task SetSuccessfulState(MediumMessage message)
{
message.ExpiresAt = DateTime.Now.AddSeconds(_options.SucceedMessageExpiredAfter);
return _dataStorage.ChangeReceiveStateAsync(message, StatusName.Succeeded);
}
private async Task<bool> SetFailedState(MediumMessage message, Exception ex)
{
if (ex is SubscriberNotFoundException)
message.Retries = _options.FailedRetryCount; // not retry if SubscriberNotFoundException
var needRetry = UpdateMessageForRetry(message);
message.Origin.AddOrUpdateException(ex);
message.ExpiresAt = message.Added.AddSeconds(_options.FailedMessageExpiredAfter);
await _dataStorage.ChangeReceiveStateAsync(message, StatusName.Failed).ConfigureAwait(false);
return needRetry;
}
private bool UpdateMessageForRetry(MediumMessage message)
{
var retries = ++message.Retries;
if (retries == _options.FailedRetryCount)
{
try
{
_options.FailedThresholdCallback?.Invoke(new FailedInfo
{
ServiceProvider = _provider,
MessageType = MessageType.Subscribe,
Message = message.Origin
});
_logger.ConsumerExecutedAfterThreshold(message.DbId, _options.FailedRetryCount);
}
catch (Exception ex)
{
_logger.ExecutedThresholdCallbackFailed(ex);
}
}
return false;
}
private async Task InvokeConsumerMethodAsync(MediumMessage message, ConsumerExecutorDescriptor descriptor,
CancellationToken cancellationToken)
{
var consumerContext = new ConsumerContext(descriptor, message);
var tracingTimestamp = TracingBefore(message.Origin, descriptor.MethodInfo);
try
{
var ret = await Invoker.InvokeAsync(consumerContext, cancellationToken).ConfigureAwait(false);
TracingAfter(tracingTimestamp, message.Origin, descriptor.MethodInfo);
if (!string.IsNullOrEmpty(ret.CallbackName))
{
var header = new Dictionary<string, string?>
{
[Headers.CorrelationId] = message.Origin.GetId(),
[Headers.CorrelationSequence] = (message.Origin.GetCorrelationSequence() + 1).ToString()
};
if (message.Origin.Headers.TryGetValue(Headers.TraceParent, out var traceparent))
header[Headers.TraceParent] = traceparent;
await _provider.GetRequiredService<ICapPublisher>()
.PublishAsync(ret.CallbackName, ret.Result, header, cancellationToken).ConfigureAwait(false);
}
}
catch (OperationCanceledException)
{
//ignore
}
catch (Exception ex)
{
var e = new SubscriberExecutionFailedException(ex.Message, ex);
TracingError(tracingTimestamp, message.Origin, descriptor.MethodInfo, e);
e.ReThrow();
}
}
#region tracing
private long? TracingBefore(Message message, MethodInfo method)
{
if (s_diagnosticListener.IsEnabled(CapDiagnosticListenerNames.BeforeSubscriberInvoke))
{
var eventData = new CapEventDataSubExecute
{
OperationTimestamp = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(),
Operation = message.GetName(),
Message = message,
MethodInfo = method
};
s_diagnosticListener.Write(CapDiagnosticListenerNames.BeforeSubscriberInvoke, eventData);
return eventData.OperationTimestamp;
}
return null;
}
private void TracingAfter(long? tracingTimestamp, Message message, MethodInfo method)
{
CapEventCounterSource.Log.WriteInvokeMetrics();
if (tracingTimestamp != null &&
s_diagnosticListener.IsEnabled(CapDiagnosticListenerNames.AfterSubscriberInvoke))
{
var now = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
var eventData = new CapEventDataSubExecute
{
OperationTimestamp = now,
Operation = message.GetName(),
Message = message,
MethodInfo = method,
ElapsedTimeMs = now - tracingTimestamp.Value
};
s_diagnosticListener.Write(CapDiagnosticListenerNames.AfterSubscriberInvoke, eventData);
}
}
private void TracingError(long? tracingTimestamp, Message message, MethodInfo? method, Exception ex)
{
if (tracingTimestamp != null &&
s_diagnosticListener.IsEnabled(CapDiagnosticListenerNames.ErrorSubscriberInvoke))
{
var now = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
var eventData = new CapEventDataSubExecute
{
OperationTimestamp = now,
Operation = message.GetName(),
Message = message,
MethodInfo = method,
ElapsedTimeMs = now - tracingTimestamp.Value,
Exception = ex
};
s_diagnosticListener.Write(CapDiagnosticListenerNames.ErrorSubscriberInvoke, eventData);
}
}
#endregion
}

View File

@ -0,0 +1,15 @@
namespace SJZY.Track.Application.CAP;
internal class SubscriberExecutionFailedException : Exception
{
private readonly Exception _originException;
public SubscriberExecutionFailedException(string message, Exception ex) : base(message, ex)
{
_originException = ex;
}
public override string? StackTrace => _originException.StackTrace;
public override string? Source => _originException.Source;
}

View File

@ -1,120 +0,0 @@
using System.Globalization;
using DotNetCore.CAP.Filter;
using SJZY.Expand.ABP.Core;
using SJZY.Expand.ABP.Core.Logger;
using SJZY.Expand.ABP.Core.PublicMethords;
using SJZY.Track.Domain.Consts;
namespace SJZY.Track.Application.EventHandlers;
public class CapFilter : SubscribeFilter
{
const string ErrorCode = "Error599";
/// <summary>
/// 执行前
/// </summary>
/// <param name="context"></param>
/// <returns></returns>
public override Task OnSubscribeExecutingAsync(ExecutingContext context)
{
context.DeliverMessage.Headers["sjzy-receive-time"] = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss fff");
// 轨迹增删改要做幂等
if (!context.DeliverMessage.Headers["cap-msg-name"]
.IsIn(GlobalConst.TrackUpsertSubscribe, GlobalConst.TrackDeleteSubscribe))
return Task.CompletedTask;
// 本地调试不做幂等
if (System.Diagnostics.Debugger.IsAttached)
{
return Task.CompletedTask;
}
var argumentJsonStr = context.Arguments.ToJson();
context.DeliverMessage.Headers["sjzy-arguments"] = argumentJsonStr;
if (!string.IsNullOrEmpty(argumentJsonStr))
{
var hash = argumentJsonStr.GetHash();
using (RedisHelper.Client.Lock("Filter:" + hash, 10))
{
var isSkip = !string.IsNullOrEmpty(RedisHelper.Client.Get($"CapExcuteFilter:{hash}"));
if (isSkip)
throw new BusinessException(ErrorCode, $"检测到重复消费,消费参数:{argumentJsonStr}");
RedisHelper.Client.Set($"CapExcuteFilter:{hash}", 60 * 10,
TimeSpan.FromHours(1).Seconds); // 1h内不允许一模一样的参数重复消费
}
}
return Task.CompletedTask;
}
/// <summary>
/// 执行后
/// </summary>
/// <param name="context"></param>
/// <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!);
return Task.CompletedTask;
}
/// <summary>
/// 消费异常事件
/// </summary>
/// <param name="context"></param>
/// <returns></returns>
public override Task OnSubscribeExceptionAsync(ExceptionContext context)
{
context.ExceptionHandled = true;
context.DeliverMessage.Headers.TryGetValue("cap-msg-id", out var messageId);
if (context.Exception is BusinessException ex && ex.Code == ErrorCode)
LogHelper.Warn(ex.ToString());
else
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!);
return Task.CompletedTask;
}
private void BuildLog(string receiveTime, string receiveName, string? arguments, bool isFromExecptionFliter,
string messageId)
{
var parsedDate = DateTime.ParseExact(receiveTime, "yyyy-MM-dd HH:mm:ss fff", CultureInfo.InvariantCulture);
var useTime = DateTime.Now - parsedDate;
if (useTime.Minutes > 20)
{
var log = $@"
|-消息编号:{messageId}
|-消费者名:{receiveName}
|-跳过执行:{(isFromExecptionFliter ? "是" : "否")}
|-执行时间:{receiveTime}
|-消费耗时:{useTime.Milliseconds}毫秒
|-请求参数:{arguments}";
LogHelper.Error(log);
}
else
{
//LogHelper.Info(log);
}
}
}

View File

@ -1,48 +0,0 @@
using DotNetCore.CAP;
using SJZY.Track.Application.Contract.Events;
using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate;
using SJZY.Track.Domain.Consts;
using Volo.Abp.DependencyInjection;
namespace SJZY.Track.Application.EventHandlers;
public class OrderLinkAfterFillTrackHandler : ITransientDependency, ICapSubscribe
{
private readonly ITrackDetailsRepository _trackDetailsRepository;
private readonly ICapPublisher _capPublisher;
public OrderLinkAfterFillTrackHandler(ITrackDetailsRepository trackDetailsRepository, ICapPublisher capPublisher)
{
_trackDetailsRepository = trackDetailsRepository;
_capPublisher = capPublisher;
}
[CapSubscribe(GlobalConst.OrderLinkAfterFillTrack)]
public async Task ConsumerAsync(OrderLinkAfterFillTrackEvent @event)
{
var tracks = await _trackDetailsRepository.GetListAsync(@event.OrderId);
foreach (var order in @event.LinkedOrders)
{
// 补轨迹
foreach (var track in tracks)
{
var fillTrackEvent = new TrackUpsertEvent
{
OrderId = order.OrderId,
OrderType = order.OrderType,
OrderNo = order.OrderNo,
TrackCode = track.TrackCode,
TrackPlace = track.TrackPlace,
TrackTime = track.TrackTime,
TrackTimeZone = track.TrackTimeZone,
SubSystemId = order.SubSystemId,
UpSubSystemId = @event.SubSystemId,
};
await _capPublisher.PublishAsync(GlobalConst.TrackUpsertSubscribe, fillTrackEvent);
}
}
}
}

View File

@ -1,77 +0,0 @@
using DotNetCore.CAP;
using SJZY.Track.Application.Contract.Events;
using SJZY.Track.Domain.Aggregates.OrderAggregate;
using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate;
using SJZY.Track.Domain.Consts;
using Volo.Abp.DependencyInjection;
namespace SJZY.Track.Application.EventHandlers;
public class TrackChangeAfterUpdateOrderEventHandler : ITransientDependency, ICapSubscribe
{
private readonly IOrderRepository _orderRepository;
private readonly ITrackDetailsRepository _trackDetailsRepository;
public TrackChangeAfterUpdateOrderEventHandler(IOrderRepository orderRepository,
ITrackDetailsRepository trackDetailsRepository)
{
_orderRepository = orderRepository;
_trackDetailsRepository = trackDetailsRepository;
}
[CapSubscribe(GlobalConst.TrackChangeAfterUpdateOrderSubscribe)]
public async Task ConsumerAsync(TrackChangeAfterUpdateOrderEvent @event)
{
var order = await _orderRepository.GetFirstAsync(x => x.Id == @event.OrderId);
if (order == default) return;
if (@event.IsDelete)
{
var track = await _trackDetailsRepository.GetLatestAsync(@event.OrderId);
// 删除全部轨迹导致的重置行为
if (track == default)
{
if (order.LatestTrackTime == null) return;
// tips:订单是支持从已完成态到未完成态的
order.ResetTrackInfo();
await _orderRepository.UpdateAsync(order);
return;
}
// 删除引发的回退更新
order.IsCompleted = @event.IsCompleted;
order.LatestTrackContent = track.TrackContent;
order.LatestTrackEnContent = track.TrackEnContent;
order.LatestTrackPlace = track.TrackPlace;
order.LatestTrackTime = track.TrackTime;
order.LatestTrackTimeZone = track.TrackTimeZone;
}
else
{
if (order.LatestTrackTime != null &&
@event.TrackTime.Subtract(TimeSpan.FromHours(@event.TrackTimeZone)) <
order.LatestTrackTime.Value.Subtract(TimeSpan.FromHours(order.LatestTrackTimeZone ?? 0)))
{
return;
}
// upsert引发的更新
order.IsCompleted = @event.IsCompleted;
order.LatestTrackContent = @event.TrackContent;
order.LatestTrackEnContent = @event.TrackEnContent;
order.LatestTrackPlace = @event.TrackPlace;
order.LatestTrackTime = @event.TrackTime;
order.LatestTrackTimeZone = @event.TrackTimeZone;
}
await _orderRepository.UpdateColumnAsync(order, x => new
{
x.IsCompleted, x.LatestTrackContent, x.LatestTrackEnContent, x.LatestTrackPlace, x.LatestTrackTime,
x.LatestTrackTimeZone
});
}
}

View File

@ -1,86 +0,0 @@
using DotNetCore.CAP;
using SJZY.Expand.ABP.Core.Logger;
using SJZY.Expand.ABP.Core.PublicMethords;
using SJZY.Track.Application.Contract.Events;
using SJZY.Track.Domain.Aggregates.OrderAggregate;
using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate;
using SJZY.Track.Domain.Consts;
using SJZY.Track.Domain.Queries;
using Volo.Abp.DependencyInjection;
namespace SJZY.Track.Application.EventHandlers;
public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe
{
private readonly IOrderRepository _orderRepository;
private readonly IProvideQueries _provideQueries;
private readonly ICapPublisher _capPublisher;
private readonly ITrackDetailsRepository _trackDetailsRepository;
public TrackDeleteEventHandler(IProvideQueries provideQueries,
IOrderRepository orderRepository, ICapPublisher capPublisher, ITrackDetailsRepository trackDetailsRepository)
{
_provideQueries = provideQueries;
_orderRepository = orderRepository;
_capPublisher = capPublisher;
_trackDetailsRepository = trackDetailsRepository;
}
[CapSubscribe(GlobalConst.TrackDeleteSubscribe, Group = GlobalConst.TrackDeleteSubscribe)]
public async Task ConsumerAsync(TrackDeleteEvent @event)
{
if (@event.Visited.Contains(@event.OrderId))
{
LogHelper.Error($"删除轨迹订单时,发现循环依赖:{@event.Visited.JoinAsString(",")}");
return;
}
// 不同子系统需要做内容映射
if (@event.UpSubSystemId != 0 && @event.UpSubSystemId != @event.SubSystemId)
{
var mapCodes = await _provideQueries
.MapTrackCodesAsync(@event.UpSubSystemId, @event.SubSystemId, @event.TrackCodes.ToArray());
if (mapCodes.Count == 0)
{
//throw new BusinessException(message: "订单跨子系统,未配置内容映射,请联系管理员");
return;
}
@event.TrackCodes = mapCodes;
}
await DeleteTrackAsync(@event);
}
private async Task DeleteTrackAsync(TrackDeleteEvent @event)
{
// 删除轨迹
var isOk = await _trackDetailsRepository.DeleteTrackAsync(@event.OrderId, @event.TrackCodes);
if (isOk)
{
// 触发订单主表更新事件
await _capPublisher.PublishDelayAsync(TimeSpan.FromSeconds(3),
GlobalConst.TrackChangeAfterUpdateOrderSubscribe,
new TrackChangeAfterUpdateOrderEvent
{
OrderId = @event.OrderId,
IsDelete = true
});
}
// 根据订单关联关系异步向下刷
var linkedOrders = await _orderRepository.GetLinkedOrdersAsync(@event.OrderId);
foreach (var linkedOrder in linkedOrders)
{
var nextEvent = @event.DeepCopy();
nextEvent.Visited.Add(@event.OrderId);
nextEvent.TrackCodes = @event.TrackCodes;
nextEvent.SubSystemId = linkedOrder.SubSystemId;
nextEvent.UpSubSystemId = @event.SubSystemId;
await _capPublisher.PublishAsync(GlobalConst.TrackDeleteSubscribe, nextEvent);
}
}
}

View File

@ -1,50 +0,0 @@
using DotNetCore.CAP;
using SJZY.Track.Application.Contract.Dtos.TrackPush;
using SJZY.Track.Domain.Aggregates.OrderAggregate;
using SJZY.Track.Domain.Consts;
using SJZY.Track.Domain.Enums;
using Volo.Abp.DependencyInjection;
namespace SJZY.Track.Application.EventHandlers;
public class TrackPushAfterUpdateOrderEventHandler : ITransientDependency, ICapSubscribe
{
private readonly IOrderRepository _orderRepository;
public TrackPushAfterUpdateOrderEventHandler(IOrderRepository orderRepository)
{
_orderRepository = orderRepository;
}
[CapSubscribe(GlobalConst.TrackPushAfterUpdateOrderSubscribe)]
public async Task ConsumerAsync(TrackPushAfterUpdateOrderDto dto)
{
var order = await _orderRepository.GetFirstAsync(x => x.Id == dto.OrderId);
if (order == default) return;
/*
* 异常:最近一条推送是失败
*/
if (!dto.IsSuccess)
{
order.QuestionType = (int)TrackProblemType.Error;
order.LatestPushError = dto.Message;
}
else
{
order.QuestionType = (int)TrackProblemType.None;
order.LatestPushError = null;
if (dto.OptionType == OptionType.新增)
{
// tips:只有新增会影响推送进度
order.LatestPushTrackCode = dto.TrackCode;
order.LatestPushTime = DateTime.Now;
}
}
await _orderRepository.UpdateColumnAsync(order, x => new
{
x.LatestPushTrackCode, x.QuestionType, x.LatestPushError, x.LatestPushTime
});
}
}

View File

@ -1,232 +0,0 @@
using DotNetCore.CAP;
using Newtonsoft.Json;
using SJZY.Expand.ABP.Core.PublicMethords;
using SJZY.InterfaceRelay.Application.Contracts;
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.TrackPushAggregate;
using SJZY.Track.Domain.Consts;
using SJZY.Track.Domain.Enums;
using Volo.Abp.DependencyInjection;
using Volo.Abp.Http.Client.DynamicProxying;
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;
public TrackPushEventHandler(IHttpClientProxy<IInterfaceForwardServiceV2> httpClientProxy,
ITrackPushRecordsRepository trackPushRecordsRepository,
IOrderRepository orderRepository, ITrackPushCacheService trackPushCacheService)
{
_httpClientProxy = httpClientProxy;
_trackPushRecordsRepository = trackPushRecordsRepository;
_orderRepository = orderRepository;
_trackPushCacheService = trackPushCacheService;
}
[CapSubscribe(GlobalConst.TrackPushSubscribe, Group = GlobalConst.TrackPushSubscribe)]
public async Task ConsumerAsync(TrackPushEvent @event)
{
var order = await _orderRepository.GetFirstAsync(x =>
x.Id == @event.OrderId, x => new
{
x.SubSystemId, x.LatestPushTrackCode, x.QuestionType, x.CustomId
});
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;
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 is not { IsAutoPush: true })
{
config = null;
}
}
}
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);
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);
break;
default:
throw new ArgumentOutOfRangeException();
}
// 往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)
{
var order = new OrderEntity();
order.Id = dto.OrderId;
/*
* 异常:最近一条推送是失败
*/
if (!dto.IsSuccess)
{
order.QuestionType = (int)TrackProblemType.Error;
order.LatestPushError = dto.Message;
}
else
{
order.QuestionType = (int)TrackProblemType.None;
order.LatestPushError = null;
if (dto.OptionType == OptionType.新增)
{
// tips:只有新增会影响推送进度?
order.LatestPushTrackCode = dto.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
});
}
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 body = new TrackPushInput()
{
OrderNo = @event.OrderNo,
};
body.Data.Add(track);
if (config.IsAllPush)
{
var history = await _trackPushRecordsRepository.GetLatestTrackPushRecordAsync(@event.OrderId);
if (history != default)
{
body = JsonConvert.DeserializeObject<TrackPushInput>(history.PushContent);
body!.Data.Add(track);
}
}
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;
}
private async Task DoUpdateAsync(TrackPushEvent @event,
TrackPushContentConfigDto config, TrackPushRecordIndex log)
{
var data = new
{
orderNo = @event.OrderNo,
data = new[]
{
new
{
// 推真实的发生地和发生时间
location = @event.TrackPlace,
date = @event.TrackTime.AddHours(
config.TrackTimeZone ?? @event.TrackTimeZone - @event.TrackTimeZone),
timeZone = config.TrackTimeZone ?? @event.TrackTimeZone,
description = config.TrackEnContentValue,
description_cn = config.TrackContentValue,
code = config.TrackContentCodeValue,
status = config.TrackStatusValue,
extra = config.ExpandFieldsValue
}
}
};
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;
}
}

View File

@ -1,170 +0,0 @@
using DotNetCore.CAP;
using SJZY.Expand.ABP.Core.Logger;
using SJZY.Expand.ABP.Core.PublicMethords;
using SJZY.Track.Application.Contract.Events;
using SJZY.Track.Domain.Aggregates.OrderAggregate;
using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate;
using SJZY.Track.Domain.Consts;
using SJZY.Track.Domain.Queries;
using Volo.Abp.DependencyInjection;
namespace SJZY.Track.Application.EventHandlers;
public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe
{
private readonly IOrderRepository _orderRepository;
private readonly IProvideQueries _provideQueries;
private readonly ITrackDetailsRepository _trackDetailsRepository;
private readonly ICapPublisher _capPublisher;
public TrackUpsertEventHandler(IOrderRepository orderRepository, ICapPublisher capPublisher,
IProvideQueries provideQueries, ITrackDetailsRepository trackDetailsRepository)
{
_orderRepository = orderRepository;
_capPublisher = capPublisher;
_provideQueries = provideQueries;
_trackDetailsRepository = trackDetailsRepository;
}
[CapSubscribe(GlobalConst.TrackUpsertSubscribe, Group = GlobalConst.TrackUpsertSubscribe)]
public async Task ConsumerAsync(TrackUpsertEvent @event)
{
if (@event.Visited.Contains(@event.OrderId))
{
LogHelper.Error($"轨迹订单循环依赖预警:{@event.Visited.JoinAsString(",")}");
return;
}
// 不同子系统需要做内容映射
if (@event.UpSubSystemId != 0 && @event.UpSubSystemId != @event.SubSystemId)
{
var mapCode = (await _provideQueries
.MapTrackCodesAsync(@event.UpSubSystemId, @event.SubSystemId, @event.TrackCode))
.FirstOrDefault();
if (mapCode == default)
{
// 跨子系统没做映射的轨迹不处理
return;
}
@event.TrackCode = mapCode;
}
var track = await _trackDetailsRepository.GetFirstOrDefaultAsync(@event.OrderId, @event.TrackCode);
if (@event.IsAdd != null)
{
if (track == null && !@event.IsAdd!.Value) return; // 不存在不让改
if (track != null && @event.IsAdd!.Value) return; // 存在不让创建
}
// step1、upsert es轨迹
track = await UpsertTrackAsync(track, @event);
if (track == null) return; // 模板没配该轨迹代码,不再继续
// step2、触发轨迹推送事件(异步)
var trackPushEvent = new TrackPushEvent
{
OptionType = track.Id.IsNullOrWhiteSpace() ? 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,
};
await _capPublisher.PublishAsync(GlobalConst.TrackPushSubscribe, trackPushEvent);
// step3、触发订单主表更新事件(异步)
await _capPublisher.PublishAsync(GlobalConst.TrackChangeAfterUpdateOrderSubscribe,
new TrackChangeAfterUpdateOrderEvent
{
OrderId = track.OrderId,
IsCompleted = @event.IsCompleted,
TrackCode = track.TrackCode,
TrackContent = track.TrackContent,
TrackEnContent = track.TrackEnContent,
TrackPlace = track.TrackPlace,
TrackTime = track.TrackTime,
TrackTimeZone = track.TrackTimeZone
});
// step4、根据订单关联关系异步向下刷(异步)
var linkedOrders = await _orderRepository.GetLinkedOrdersAsync(@event.OrderId);
foreach (var linkedOrder in linkedOrders)
{
var nextEvent = @event.DeepCopy();
nextEvent.Visited.Add(@event.OrderId);
// tips:一定要重置为false,父级订单的完成不会使得关联订单完成,关联单也必须由刷新来完成或手动设置完成!!
nextEvent.IsCompleted = false; // 能够使得轨迹IsCompleted为true的点,全局只有一个:轨迹刷新
nextEvent.OrderId = linkedOrder.OrderId;
nextEvent.OrderType = linkedOrder.OrderType;
nextEvent.OrderNo = linkedOrder.OrderNo;
nextEvent.SubSystemId = linkedOrder.SubSystemId;
nextEvent.UpSubSystemId = @event.SubSystemId;
await _capPublisher.PublishAsync(GlobalConst.TrackUpsertSubscribe, nextEvent);
}
}
/// <summary>
/// 不存在则新增轨迹,存在则更新
/// </summary>
/// <param name="track"></param>
/// <param name="event"></param>
private async Task<TrackDetailsIndex?> UpsertTrackAsync(TrackDetailsIndex? track, TrackUpsertEvent @event)
{
if (track == default)
{
track = new TrackDetailsIndex
{
IndexName = $"{GlobalConst.TrackIndex}-{DateTime.Now:yyyy.MM.dd}",
OrderId = @event.OrderId,
OrderNo = @event.OrderNo,
OrderType = @event.OrderType,
TrackCode = @event.TrackCode,
TrackPlace = @event.TrackPlace,
TrackTime = @event.TrackTime,
TrackTimeZone = @event.TrackTimeZone,
ServiceProviderName = @event.ServiceProviderName,
OriginTrackContent = @event.OriginTrackContent,
OriginTrackData = @event.OriginTrackData,
IsCanDelete = false,
CreateUser = @event.OperationMan ?? GlobalConst.SystemUser,
CreateTime = DateTime.Now
};
//todo:可以优化
// 根据轨迹code在订单模板内容中寻找匹配的
var trackContent = await _provideQueries.MatchTrackContentAsync(@event.OrderId, @event.TrackCode);
if (trackContent == default)
{
// 订单所绑定的模板,没有一条内容可以匹配发生的轨迹
return null;
}
track.TrackStatus = trackContent.Status;
track.TrackCode = trackContent.TrackCode;
track.TrackContent = trackContent.Content;
track.TrackEnContent = trackContent.EnContent;
track.IsCanDelete = trackContent.IsCanDelete;
track.ZeroTrackTime = track.TrackTime.Subtract(TimeSpan.FromHours(track.TrackTimeZone));
}
else
{
track.TrackPlace = @event.TrackPlace;
track.TrackTime = @event.TrackTime;
track.TrackTimeZone = @event.TrackTimeZone;
track.UpdateUser = @event.OperationMan ?? GlobalConst.SystemUser;
track.UpdateTime = DateTime.Now;
}
await _trackDetailsRepository.UpsertAsync(track);
return track;
}
}

View File

@ -12,15 +12,17 @@
</ItemGroup>
<ItemGroup>
<PackageReference Include="SJZY.InterfaceRelay.Application.Contracts" Version="7.1.9"/>
<PackageReference Include="SJZY.InterfaceRelay.Application.Contracts" Version="7.2.1"/>
<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" />
<PackageReference Include="SJZY.UserCenter.HttpApi" Version="3.0.9-alpha"/>
<PackageReference Include="SJZY.Expand.CAP" Version="7.8.0" />
<PackageReference Include="DotNetCore.CAP" Version="8.0.1" />
<PackageReference Include="DotNetCore.CAP.Dashboard" Version="8.0.1" />
<PackageReference Include="DotNetCore.CAP.RabbitMQ" Version="8.0.1" />
<PackageReference Include="DotNetCore.CAP.MySql" Version="8.0.1" />
<PackageReference Include="DotNetCore.CAP" Version="8.1.2"/>
<PackageReference Include="DotNetCore.CAP.Dashboard" Version="8.1.2"/>
<PackageReference Include="DotNetCore.CAP.RabbitMQ" Version="8.1.2"/>
<PackageReference Include="DotNetCore.CAP.MySql" Version="8.1.2"/>
<PackageReference Include="DotNetCore.CAP.InMemoryStorage" Version="8.1.2"/>
<PackageReference Include="Savorboard.CAP.InMemoryMessageQueue" Version="7.0.0"/>
</ItemGroup>
<ItemGroup>

View File

@ -2,6 +2,7 @@
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.Consts;
using SJZY.Track.Domain.Queries;
@ -14,6 +15,7 @@ public partial class OrderService
{
private readonly IBasicQuery<OrderLinkEntity> _orderLinkQuery;
private readonly ICapPublisher _capPublisher;
private readonly ITrackDetailsRepository _trackDetailsRepository;
/// <summary>
/// 关联、解除关联订单
@ -52,23 +54,12 @@ public partial class OrderService
if (!input.IsLink) return await UnLinkOrderAsync(input);
// 关联订单
var linkedOrders = await DoLinkOrderAsync(input, order.Id);
// 补轨迹
if (linkedOrders.Count > 0)
{
await _capPublisher.PublishAsync(GlobalConst.OrderLinkAfterFillTrack, new OrderLinkAfterFillTrackEvent
{
OrderId = order.Id,
SubSystemId = order.SubSystemId,
LinkedOrders = linkedOrders,
});
}
await DoLinkOrderAsync(input, order.Id);
return new UnifyResultDto(true, true);
}
private async Task<List<LinkedOrder>> DoLinkOrderAsync(LinkOrderInput input, int orderId)
private async Task DoLinkOrderAsync(LinkOrderInput input, int orderId)
{
// 获取已关联的订单
var linkOrderNos = await _orderLinkQuery.GetListAsync(x =>
@ -81,9 +72,9 @@ 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.SubSystemId });
x => new { x.Id, x.OrderNo, x.OrderType, x.SubSystemId, x.CustomId });
if (canLinkOrders.Count == 0) return new List<LinkedOrder>();
if (canLinkOrders.Count == 0) return;
await _orderRepository.LinkOrderAsync(canLinkOrders.Select(x => new OrderLinkEntity
{
@ -95,13 +86,33 @@ public partial class OrderService
LinkOrderType = input.LinkOrderType
}).ToList());
return canLinkOrders.Select(x => new LinkedOrder
var tracks = await _trackDetailsRepository.GetListAsync(orderId);
// 补轨迹
foreach (var group in canLinkOrders.GroupBy(x => x.SubSystemId))
{
SubSystemId = x.SubSystemId,
OrderId = x.Id,
OrderNo = x.OrderNo,
OrderType = input.LinkOrderType
}).ToList();
foreach (var track in tracks)
{
var fillTrackEvent = new TrackUpsertEvent
{
SubSystemId = group.Key,
OrderIdentifiers = group.Select(x => new OrderIdentifier
{
OrderId = x.Id,
OrderNo = x.OrderNo,
OrderType = x.OrderType,
CustomId = x.CustomId
}).ToArray(),
TrackCode = track.TrackCode,
TrackPlace = track.TrackPlace,
TrackTime = track.TrackTime,
TrackTimeZone = track.TrackTimeZone,
IsAdd = true
};
await _capPublisher.PublishAsync(GlobalConst.TrackUpsertSubscribe, fillTrackEvent);
}
}
}
private async Task<UnifyResultDto> UnLinkOrderAsync(LinkOrderInput input)

View File

@ -2,6 +2,7 @@
using SJZY.Track.Application.Contract.Dtos.ProvideServices;
using SJZY.Track.Application.Contract.Services;
using SJZY.Track.Domain.Aggregates.OrderAggregate;
using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate;
using SJZY.Track.Domain.Queries;
using SJZY.UserCenter.HttpApi.Services;
@ -17,9 +18,10 @@ public partial class OrderService : ApplicationService, IOrderService
private readonly IOrderRepository _orderRepository;
public OrderService(IOrderRepository orderRepository, IBasicQuery<OrderLinkEntity> orderLinkQuery,
ICapPublisher capPublisher)
ICapPublisher capPublisher, ITrackDetailsRepository trackDetailsRepository)
{
_orderRepository = orderRepository;
_trackDetailsRepository = trackDetailsRepository;
_capPublisher = capPublisher;
_orderLinkQuery = orderLinkQuery;
}

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;
@ -77,42 +78,50 @@ public class TrackPullService : ApplicationService
/// <returns></returns>
[AllowAnonymous]
[HttpPost("Track/RefreshForTask")]
public async Task<bool> RefreshForTaskAsync()
public bool RefreshForTaskAsync()
{
var configs = await _trackPullCacheService.GetAllPullConfigsAsync();
foreach (var config in configs)
Task.Run(async () =>
{
var redisKey = _trackPullCacheService.TrackPullKey(config.SubSystemId, config.ServiceProviderId);
if (await RedisHelper.Client.ExistsAsync(redisKey))
{
continue;
}
var configs = await _trackPullCacheService.GetAllPullConfigsAsync();
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 }))
foreach (var config in configs)
{
var data = group.Select(x => new
var redisKey = _trackPullCacheService.TrackPullKey(config.SubSystemId, config.ServiceProviderId);
if (await RedisHelper.Client.ExistsAsync(redisKey))
{
x.OrderNo
}).ToList();
continue;
}
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);
var page = 1;
var orders = await _provideQueries.FindOrderWithServiceProviderAsync(config.ServiceProviderId, page);
while (orders.Any())
{
var serviceProviderCode = orders.FirstOrDefault()?.ServiceProviderCode;
var serviceProviderName = orders.FirstOrDefault()?.ServiceProviderName;
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);
}
await RedisHelper.Client.SetExAsync(redisKey, config.PullInterval * 60, orders.Count);
}
if (config.PullInterval < 60) config.PullInterval = 60;
await RedisHelper.Client.SetExAsync(redisKey, config.PullInterval * 60, orders.Count);
}
});
return true;
}

View File

@ -0,0 +1,45 @@
using SJZY.Track.Application.Contract.Dtos.ProvideServices;
using SJZY.Track.Application.Contract.Services;
using SJZY.Track.Domain.Aggregates.TrackPushAggregate;
using SJZY.Track.Domain.Dtos.TrackPush;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using Volo.Abp.ObjectMapping;
namespace SJZY.Track.Application.Services.ProvideServices
{
/// <summary>
/// 轨迹推送配置服务
/// </summary>
[ApiExplorerSettings(GroupName = "提供接口")]
[Route("/ProviderService")]
public class TrackPushConfigService : ApplicationService, ITrackPushConfigService
{
private readonly ITrackPushConfigRepository _trackPushConfigRepository;
public TrackPushConfigService(
ITrackPushConfigRepository trackPushConfigRepository
)
{
_trackPushConfigRepository = trackPushConfigRepository;
}
/// <summary>
/// 查询轨迹推送配置关系
/// </summary>
/// <param name="input"></param>
/// <returns></returns>
[HttpPost("Track/QueryTrackPullMappings")]
public async Task<UnifyResultDto> QueryTrackPullMappingsAsync(QueryTrackPullMappingsInput input)
{
var vo = ObjectMapper.Map<QueryTrackPullMappingsInput, QueryTrackPullMappingsInputVO>(input);
var result = await _trackPushConfigRepository.QueryTrackPullMappingsAsync(vo);
var output = ObjectMapper.Map<List<QueryTrackPullMappingsInfoVO>, List<QueryTrackPullMappingsInfo>>(result);
return new UnifyResultDto(true, output, "操作成功");
}
}
}

View File

@ -3,10 +3,10 @@ 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;
using Volo.Abp.Http.Client.DynamicProxying;
namespace SJZY.Track.Application.Services.ProvideServices;
@ -17,20 +17,20 @@ 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;
private readonly IHttpClientProxy<IRequestCurrentUserService> _requestCurrentUserService;
public TrackPushService(ITrackDetailsRepository trackDetailsRepository
, IOrderRepository orderRepository
public TrackPushService(IOrderRepository orderRepository
, ICapPublisher capPublisher
, ITrackPushRecordsRepository trackPushRecordsRepository)
, ITrackPushRecordsRepository trackPushRecordsRepository,
IHttpClientProxy<IRequestCurrentUserService> requestCurrentUserService)
{
_trackDetailsRepository = trackDetailsRepository;
_orderRepository = orderRepository;
_capPublisher = capPublisher;
_trackPushRecordsRepository = trackPushRecordsRepository;
_requestCurrentUserService = requestCurrentUserService;
}
/// <summary>
@ -43,48 +43,42 @@ public class TrackPushService : ApplicationService
{
input.OrderNos = input.OrderNos.Where(x => !x.IsNullOrWhiteSpace()).Distinct().ToList(); // 单号去重、去空
var orders = await _orderRepository.GetListAsync(x =>
var orders = await _orderRepository.GetArrayAsync(x =>
x.OrderType == input.OrderType && input.OrderNos.Contains(x.OrderNo), x => new
{
x.Id, x.OrderNo, x.OrderType, x.SubSystemId, x.CustomId
x.Id, x.SubSystemId, x.CustomId
});
if (orders.Count == 0)
if (!orders.Any())
{
throw new BusinessException(message: "一个推送配置都不存在");
throw new BusinessException(message: "一个订单都不存在");
}
var tracks = await _trackDetailsRepository.GetListAsync(input.OrderType, input.OrderNos, input.TrackCode);
if (tracks.Count == 0)
var userInfo = await _requestCurrentUserService.Service.GetCurrentUserBasicInfoAsync();
if (userInfo.Code != "200" || userInfo.Data is null)
{
throw new BusinessException(message: "一个能推的轨迹都不存在");
//TODO: log || ex
throw new BusinessException(message: $"调用用户中心,获取用户信息失败:{userInfo?.Msg}");
}
var res = 0;
var name = userInfo.Data.UserName;
foreach (var track in tracks)
foreach (var group in orders.GroupBy(x => new { x.SubSystemId, x.CustomId }))
{
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,
SubSystemId = group.Key.SubSystemId,
CustomId = group.Key.CustomId,
OrderIds = group.Select(x => x.Id).ToArray(),
TrackCode = input.TrackCode,
OperationMan = name,
};
await _capPublisher.PublishAsync(GlobalConst.TrackPushSubscribe, trackPushEvent);
res++;
await _capPublisher.PublishAsync(GlobalConst.TrackManualPushSubscribe, trackPushEvent);
}
return $"即将推送轨迹代码{input.TrackCode}给{res}个订单";
return $"即将推送轨迹代码{input.TrackCode}给{orders.Length}个订单";
}

View File

@ -17,7 +17,7 @@ public partial class TrackService
/// <returns></returns>
/// <exception cref="BusinessException"></exception>
[HttpPost("Track/Query")]
public async Task<List<GetTrackOutput>> QueryAsync(GetTrackInput input)
public async Task<UnifyResultDto> QueryAsync(GetTrackInput input)
{
if (input.OrderNos.Count > 500)
{
@ -50,7 +50,7 @@ public partial class TrackService
await TrackDetailSort(res, input);
return res;
return new UnifyResultDto(true, res, "操作成功");
}
private async Task TrackDetailSort(List<GetTrackOutput> list, GetTrackInput input)

View File

@ -7,6 +7,7 @@ using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate;
using SJZY.Track.Domain.Consts;
using SJZY.Track.Domain.Queries;
using SJZY.UserCenter.HttpApi.Services;
using Volo.Abp.Http.Client.DynamicProxying;
namespace SJZY.Track.Application.Services.ProvideServices;
@ -20,14 +21,17 @@ public partial class TrackService : ApplicationService, ITrackService
private readonly IOrderRepository _orderRepository;
private readonly IProvideQueries _provideQueries;
private readonly ICapPublisher _capPublisher;
private readonly IHttpClientProxy<IRequestCurrentUserService> _requestCurrentUserService;
public TrackService(IOrderRepository orderRepository, IProvideQueries provideQueries, ICapPublisher capPublisher,
ITrackDetailsRepository trackDetailsRepository)
ITrackDetailsRepository trackDetailsRepository,
IHttpClientProxy<IRequestCurrentUserService> requestCurrentUserService)
{
_orderRepository = orderRepository;
_provideQueries = provideQueries;
_provideQueries = provideQueries;
_capPublisher = capPublisher;
_requestCurrentUserService = requestCurrentUserService;
_trackDetailsRepository = trackDetailsRepository;
}
@ -43,18 +47,36 @@ 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.Id, x.SubSystemId, x.OrderNo, x.OrderType, x.CustomId
});
if (order == default)
{
throw new BusinessException(message: $"订单:{input.OrderNo}不存在");
}
var userInfo = await _requestCurrentUserService.Service.GetCurrentUserBasicInfoAsync();
if (userInfo.Code != "200" || userInfo.Data is null)
{
//TODO: log || ex
throw new BusinessException(message: $"调用用户中心,获取用户信息失败:{userInfo?.Msg}");
}
var name = userInfo.Data.UserName;
// 新增/更新轨迹
var @event = ObjectMapper.Map<UpsertTrackInput, TrackUpsertEvent>(input);
@event.OrderId = order.Id;
@event.SubSystemId = order.SubSystemId;
@event.OperationMan = CurrentUserRequest.CurrentUser.Name;
@event.OrderIdentifiers = new[]
{
new OrderIdentifier
{
OrderId = order.Id,
OrderNo = order.OrderNo,
OrderType = order.OrderType,
CustomId = order.CustomId
}
};
@event.OperationMan = name;
await _capPublisher.PublishAsync(GlobalConst.TrackUpsertSubscribe, @event);
return new UnifyResultDto(true, true);
}
@ -77,28 +99,42 @@ 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 order in orders)
var userInfo = await _requestCurrentUserService.Service.GetCurrentUserBasicInfoAsync();
if (userInfo.Code != "200" || userInfo.Data is null)
{
//TODO: log || ex
throw new BusinessException(message: $"调用用户中心,获取用户信息失败:{userInfo?.Msg}");
}
var name = userInfo.Data.UserName;
foreach (var group in orders.GroupBy(x => x.SubSystemId))
{
// 新增/更新轨迹
var @event = new TrackUpsertEvent
{
SubSystemId = group.Key,
OrderIdentifiers = group.Select(order => new OrderIdentifier
{
OrderId = order.Id,
OrderNo = order.OrderNo,
OrderType = input.OrderType,
CustomId = order.CustomId
}).ToArray(),
IsAdd = input.IsAdd,
TrackCode = input.TrackCode,
TrackPlace = input.TrackPlace,
TrackTime = input.TrackTime!.Value,
TrackTimeZone = input.TrackTimeZone!.Value,
OrderType = input.OrderType,
OrderNo = order.OrderNo,
OrderId = order.Id,
SubSystemId = order.SubSystemId,
OperationMan = CurrentUserRequest.CurrentUser.Name
OperationMan = name
};
await _capPublisher.PublishAsync(GlobalConst.TrackUpsertSubscribe, @event);
}
return new UnifyResultDto(true, true);
}
@ -129,13 +165,22 @@ public partial class TrackService : ApplicationService, ITrackService
throw new BusinessException(message: "没有一个轨迹代码是可删的");
}
var userInfo = await _requestCurrentUserService.Service.GetCurrentUserBasicInfoAsync();
if (userInfo.Code != "200" || userInfo.Data is null)
{
//TODO: log || ex
throw new BusinessException(message: $"调用用户中心,获取用户信息失败:{userInfo?.Msg}");
}
var name = userInfo.Data.UserName;
// 删除轨迹
var @event = new TrackDeleteEvent
{
OrderId = order.Id,
OrderIds = new[] { order.Id },
TrackCodes = codes.Select(x => x.TrackCode).ToList(),
SubSystemId = order.SubSystemId,
OperationMan = CurrentUserRequest.CurrentUser.Name
OperationMan = name
};
await _capPublisher.PublishAsync(GlobalConst.TrackDeleteSubscribe, @event);
@ -168,15 +213,24 @@ public partial class TrackService : ApplicationService, ITrackService
throw new BusinessException(message: "没有一个轨迹代码是可删的");
}
var userInfo = await _requestCurrentUserService.Service.GetCurrentUserBasicInfoAsync();
if (userInfo.Code != "200" || userInfo.Data is null)
{
//TODO: log || ex
throw new BusinessException(message: $"调用用户中心,获取用户信息失败:{userInfo?.Msg}");
}
var name = userInfo.Data.UserName;
// 删除轨迹
foreach (var group in codes.GroupBy(x => x.OrderId))
{
var @event = new TrackDeleteEvent
{
OrderId = group.Key,
TrackCodes = codes.Select(x => x.TrackCode).ToList(),
OrderIds = group.Select(x => x.OrderId).ToArray(),
TrackCodes = input.TrackCodes,
SubSystemId = orders.First(x => x.Id == group.Key).SubSystemId,
OperationMan = CurrentUserRequest.CurrentUser.Name
OperationMan = name
};
await _capPublisher.PublishAsync(GlobalConst.TrackDeleteSubscribe, @event);
}

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 =>

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(

View File

@ -1,12 +1,13 @@
using System.Text.Json;
using DotNetCore.CAP;
using DotNetCore.CAP;
using DotNetCore.CAP.Internal;
using Microsoft.Extensions.Configuration;
using Savorboard.CAP.InMemoryMessageQueue;
using SJZY.Expand.ABP.Core.Localization;
using SJZY.Expand.ABP.Core.Logger;
using SJZY.Expand.ABP.Core.PublicMethords;
using SJZY.InterfaceRelay.Application.Contracts;
using SJZY.Track.Application.CAP;
using SJZY.Track.Application.Contract;
using SJZY.Track.Application.EventHandlers;
using SJZY.Track.Application.Localization;
using SJZY.UserCenter.Contracts;
using SJZY.UserCenter.SharedModel;
@ -74,9 +75,14 @@ namespace SJZY.Track.Application
);
//CAP
context.Services.AddSingleton<ISubscribeExecutor, MySubscribeExecutor>();
context.Services.AddCapTableNameService();
context.Services.AddCap(options =>
{
#if DEBUG
options.UseInMemoryStorage();
options.UseInMemoryMessageQueue();
#else
options.UseMySql(configuration["ConnectionStrings:DefaultConnection"]!);
options.UseRabbitMQ(configure =>
{
@ -84,12 +90,17 @@ namespace SJZY.Track.Application
configure.UserName = configuration["RabbitMQ:UserName"]!;
configure.Password = configuration["RabbitMQ:Password"]!;
configure.Port = configuration.GetValue<int>("RabbitMQ:Port");
configure.BasicQosOptions =
new RabbitMQOptions.BasicQos(configuration.GetValue<ushort>("CAP:PrefetchCount"));
});
#endif
options.FailedRetryInterval = configuration.GetValue<int>("CAP:FailedRetryInterval");
options.FailedRetryCount = configuration.GetValue<int>("CAP:FailedRetryCount");
options.SucceedMessageExpiredAfter = configuration.GetValue<int>("CAP:SucceedMessageExpiredAfter");
options.FallbackWindowLookbackSeconds =
configuration.GetValue<int>("CAP:FallbackWindowLookbackSeconds");
options.EnableSubscriberParallelExecute = true;
// options.TopicNamePrefix = "v1";
options.SucceedMessageExpiredAfter = 6 * 60 * 60;
options.FailedRetryInterval = 60;
options.FailedRetryCount = 3;
options.UseDashboard(dashboardOptions =>
{
dashboardOptions.PathBase = System.Diagnostics.Debugger.IsAttached
@ -103,8 +114,7 @@ namespace SJZY.Track.Application
LogHelper.Error($@"
|-CAP失败回调事件:
|-处理类型:{(faileCalback.MessageType == DotNetCore.CAP.Messages.MessageType.Subscribe ? "消费消息达到最大重试次数,需要开发人员介入" : "生产消息推送失败")}
|-头部信息:{faileCalback.Message.Headers.ToJson()}
|-消息内容:{(faileCalback.Message.Value is string ? faileCalback.Message.Value : ((JsonElement?)faileCalback.Message.Value)?.GetRawText())}");
|-头部信息:{faileCalback.Message.Headers.ToJson()}");
}
catch (Exception ex)
{

View File

@ -38,7 +38,7 @@ public interface IOrderRepository : IBasicRepository<OrderEntity>
/// </summary>
/// <param name="orderId"></param>
/// <returns></returns>
Task<List<LinkedOrderVO>> GetLinkedOrdersAsync(int orderId);
Task<List<LinkedOrderVO>> GetLinkedOrdersAsync(int[] orderId);
/// <summary>
/// 关联订单

View File

@ -40,7 +40,7 @@ public class OrderFlightEntity : ICreateBaicEntity, ISoftDeleteBasicEntity, IEnt
/// 创建时间
/// </summary>
[SugarColumn(ColumnName = "CreateTime")]
public DateTime? CreateTime { get; set; }
public DateTime CreateTime { get; set; }
/// <summary>
/// 是否删除

View File

@ -59,7 +59,7 @@ public class OrderLinkEntity : ICreateBaicEntity, ISoftDeleteBasicEntity, IEntit
/// 创建时间
/// </summary>
[SugarColumn(ColumnName = "CreateTime")]
public DateTime? CreateTime { get; set; }
public DateTime CreateTime { get; set; }
/// <summary>
/// 是否删除

View File

@ -40,7 +40,7 @@ public class OrderTemplateEntity : ICreateBaicEntity, ISoftDeleteBasicEntity, IE
/// 创建时间
/// </summary>
[SugarColumn(ColumnName = "CreateTime")]
public DateTime? CreateTime { get; set; }
public DateTime CreateTime { get; set; }
/// <summary>
/// 是否删除

View File

@ -4,31 +4,26 @@ namespace SJZY.Track.Domain.Aggregates.TrackDetailsAggregate;
public interface ITrackDetailsRepository
{
Task<TrackDetailsIndex?> GetFirstOrDefaultAsync(int orderId, string trackCode);
Task<List<TrackDetailsIndex>> GetListAsync(int[] orderIds, string trackCode, bool forUpsert = false);
Task<List<TrackDetailsIndex>> GetLatestListAsync(int[] eventOrderIds, List<string> trackCodes);
Task<bool> AnyAsync(int orderId, DateTime trackTime);
Task UpsertAsync(TrackDetailsIndex index);
Task<bool> AnyAsync(int orderId, string originTrackData);
Task<bool> AnyByCodeAsync(int orderId, string trackCode);
Task UpdateAsync(TrackDetailsIndex index);
Task AddManyAsync(List<TrackDetailsIndex> index);
Task AddAsync(TrackDetailsIndex track);
Task<List<TrackDetailsIndex>> GetListAsync(int orderType, List<string> orderNos);
Task<List<TrackDetailsIndex>> GetListAsync(int orderType, List<string> orderNos, string trackCode);
Task<(List<TrackDetailsIndex> list, long total)> GetListAsync(TrackDetailsQueryInput input);
Task<List<TrackDetailsIndex>> GetListAsync(int orderId);
/// <summary>
/// 删除轨迹
/// </summary>
/// <param name="orderId"></param>
/// <param name="trackCodes"></param>
/// <returns></returns>
Task<bool> DeleteTrackAsync(int orderId, List<string> trackCodes);
/// <summary>
/// 获取最近一条发生的轨迹
/// </summary>
/// <param name="orderId"></param>
/// <returns></returns>
Task<TrackDetailsIndex?> GetLatestAsync(int orderId);
Task DeleteTrackAsync(int[] orderIds, List<string> trackCodes);
}

View File

@ -35,4 +35,5 @@ public interface ITrackPushConfigRepository : IBasicRepository<TrackPushConfigEn
Task<bool> UpdateContentAsync(Expression<Func<TrackPushContentConfigEntity, bool>> columns,
Expression<Func<TrackPushContentConfigEntity, bool>> whereExpression);
Task<List<QueryTrackPullMappingsInfoVO>> QueryTrackPullMappingsAsync(QueryTrackPullMappingsInputVO input);
}

View File

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

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>

View File

@ -7,6 +7,11 @@ public partial class GlobalConst
/// </summary>
public const string TrackUpsertSubscribe = "SJZY.Track.TrackUpsert";
/// <summary>
/// 轨迹刷新引发的新增/更新 专用
/// </summary>
public const string TrackUpsertForRefreshSubscribe = "SJZY.Track.TrackUpsertForRefresh";
/// <summary>
/// 无法映射的轨迹新增
/// </summary>
@ -27,6 +32,16 @@ public partial class GlobalConst
/// </summary>
public const string TrackPushSubscribe = "SJZY.Track.TrackPush";
/// <summary>
/// 轨迹人工手动推送
/// </summary>
public const string TrackManualPushSubscribe = "SJZY.Track.TrackManualPush";
/// <summary>
/// 轨迹推送完成(接口通回调)
/// </summary>
public const string TrackPushCompletedSubscribe = "SJZY.Track.TrackPushCompleted";
/// <summary>
/// 轨迹新增/更新/删除后触发更新订单主表
/// </summary>

View File

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

View File

@ -2,6 +2,11 @@
public class LinkedOrderVO
{
/// <summary>
/// 被关联订单id
/// </summary>
public int OwnerOrderId { get; set; }
/// <summary>
/// 关联订单id
/// </summary>
@ -21,4 +26,9 @@ public class LinkedOrderVO
/// 子系统id
/// </summary>
public int SubSystemId { get; set; }
/// <summary>
/// 客户id
/// </summary>
public int CustomId { get; set; }
}

View File

@ -0,0 +1,19 @@
namespace SJZY.Track.Domain.Dtos.TrackContents;
public class TrackContentVO
{
/// <summary>
/// 轨迹状态
/// </summary>
public string TrackStatus { get; set; } = null!;
/// <summary>
/// 轨迹内容
/// </summary>
public string TrackContent { get; set; } = null!;
/// <summary>
/// 轨迹内容英文
/// </summary>
public string TrackEnContent { get; set; } = null!;
}

View File

@ -3,9 +3,14 @@
public class MatchTrackContentDto
{
/// <summary>
/// 轨迹内容id
/// 订单Id
/// </summary>
public string TrackCode { get; set; }
public int OrderId { get; set; }
/// <summary>
/// 轨迹内容code
/// </summary>
public string TrackCode { get; set; } = null!;
/// <summary>
/// 轨迹内容

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

View File

@ -0,0 +1,60 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
namespace SJZY.Track.Domain.Dtos.TrackPush
{
/// <summary>
/// 查询轨迹推送关系入参
/// </summary>
public class QueryTrackPullMappingsInfoVO
{
/// <summary>
/// 轨迹代码
/// </summary>
public string? Code { get; set; }
/// <summary>
/// 订单类型
/// </summary>
public int OrderType { get; set; }
/// <summary>
/// 服务商轨迹代码
/// </summary>
public string? ServiceCode { get; set; }
/// <summary>
/// 服务商轨迹状态
/// </summary>
public string? Status { get; set; }
/// <summary>
/// 服务商轨迹内容
/// </summary>
public string? Content { get; set; }
/// <summary>
/// 服务商轨迹内容英文
/// </summary>
public string? ContentEn { get; set; }
}
/// <summary>
/// 查询轨迹推送关系返参项
/// </summary>
public class QueryTrackPullMappingsInputVO
{
/// <summary>
/// 子系统id
/// </summary>
public int SubsystemsId { get; set; }
/// <summary>
/// 客户id
/// </summary>
public int CustomerId { get; set; }
}
}

View File

@ -22,6 +22,9 @@ public interface IBasicQuery<T> where T : IEntity
Task<List<TResult>> GetListAsync<TResult>(Expression<Func<T, bool>> whereExpression,
Expression<Func<T, TResult>> resultExpression);
Task<TResult[]> GetArrayAsync<TResult>(Expression<Func<T, bool>> whereExpression,
Expression<Func<T, TResult>> resultExpression);
/// <summary>
/// count
/// </summary>

View File

@ -1,4 +1,5 @@
using SJZY.Track.Domain.Dtos.Order;
using SJZY.Track.Domain.Dtos.TrackContents;
using SJZY.Track.Domain.Dtos.TrackDetails;
using SJZY.Track.Domain.Dtos.TrackTemplate;
@ -27,10 +28,10 @@ public interface IProvideQueries
/// <summary>
/// 根据订单id找到订单模板,匹配模板命中的轨迹内容
/// </summary>
/// <param name="orderId">订单id</param>
/// <param name="orderIds">订单ids</param>
/// <param name="code">轨迹代码</param>
/// <returns></returns>
Task<MatchTrackContentDto?> MatchTrackContentAsync(int orderId, string code);
Task<List<MatchTrackContentDto>> MatchTrackContentAsync(int[] orderIds, string code);
/// <summary>
/// 获取可删除的订单轨迹代码
@ -60,6 +61,11 @@ 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);
Task<TrackContentVO> GetTrackContentAsync(int subSystemId, string trackCode);
}

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.8.6"/>
<PackageReference Include="SJZY.Expand.ElasticSearch" Version="7.0.17" />
</ItemGroup>

View File

@ -0,0 +1,22 @@
using Nest;
using SJZY.Expand.ElasticSearch;
namespace SJZY.Track.Repository;
public static class EsExtensions
{
public static List<T> ToIdList<T>(this ISearchResponse<T> response) where T : IdBaseIndex
{
var dtos = new List<T>();
foreach (var hit in response.Hits)
{
var dto = hit.Source;
dto.Id = hit.Id;
dtos.Add(dto);
}
return dtos;
}
}

View File

@ -24,4 +24,10 @@ public class BasicQuery<T> : SimpleClient<T>, IBasicQuery<T> where T : class, IE
{
return await Context.Queryable<T>().Where(whereExpression).Select(resultExpression).ToListAsync();
}
public async Task<TResult[]> GetArrayAsync<TResult>(Expression<Func<T, bool>> whereExpression,
Expression<Func<T, TResult>> resultExpression)
{
return await Context.Queryable<T>().Where(whereExpression).Select(resultExpression).ToArrayAsync();
}
}

View File

@ -3,6 +3,7 @@ using SJZY.Track.Domain.Aggregates.TrackContentAggregate;
using SJZY.Track.Domain.Aggregates.TrackStatusAggregate;
using SJZY.Track.Domain.Aggregates.TrackTemplateAggregate;
using SJZY.Track.Domain.Dtos.Order;
using SJZY.Track.Domain.Dtos.TrackContents;
using SJZY.Track.Domain.Dtos.TrackDetails;
using SJZY.Track.Domain.Dtos.TrackTemplate;
using SJZY.Track.Domain.Queries;
@ -50,22 +51,24 @@ public class ProvideQueries : ISugarRepository, IProvideQueries,
.ToListAsync();
}
public async Task<MatchTrackContentDto?> MatchTrackContentAsync(int orderId, string code)
public async Task<List<MatchTrackContentDto>> MatchTrackContentAsync(int[] orderIds, string code)
{
if (orderIds.Length == 0) return new List<MatchTrackContentDto>();
return await Context.Queryable<OrderTemplateEntity>()
.LeftJoin<TrackTemplateContentEntity>((x, a) => x.TrackTemplateId == a.TrackTemplateId)
.LeftJoin<TrackContentEntity>((x, a, b) => a.TrackContentId == b.Id)
.LeftJoin<TrackStatusEntity>((x, a, b, c) => b.TrackStatusId == c.Id)
.Where((x, a, b) => x.OrderId == orderId && b.Code == code)
.Where((x, a, b) => orderIds.Contains(x.OrderId) && b.Code == code)
.Select((x, a, b, c) => new MatchTrackContentDto
{
OrderId = x.OrderId,
TrackCode = b.Code,
Content = b.Content,
EnContent = b.EnContent,
Status = c.Status,
IsCanDelete = a.IsCanDelete
})
.FirstAsync();
.ToListAsync();
}
public async Task<List<CanDeleteTrackCodeVO>> GetCanDeleteTrackCodesAsync(List<int> orderIds,
@ -113,7 +116,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 +132,22 @@ public class ProvideQueries : ISugarRepository, IProvideQueries,
ServiceProviderName = c.Name,
SubSystemId = a.SubSystemId
})
.Skip((pageIndex - 1) * pageSize)
.Take(pageSize)
.ToListAsync();
}
public async Task<TrackContentVO> GetTrackContentAsync(int subSystemId, string trackCode)
{
return await Context.Queryable<TrackContentEntity>()
.LeftJoin<TrackStatusEntity>((x, y) => x.TrackStatusId == y.Id)
.Where(x => x.SubSystemId == subSystemId && x.Code == trackCode)
.Select((x, y) => new TrackContentVO
{
TrackStatus = y.Status,
TrackContent = x.Content,
TrackEnContent = x.EnContent
})
.FirstAsync();
}
}

View File

@ -67,17 +67,19 @@ AND `a`.`IsDeleted` = 0 AND `b`.`IsDeleted` = 0;");
#region 订单关联
public async Task<List<LinkedOrderVO>> GetLinkedOrdersAsync(int orderId)
public async Task<List<LinkedOrderVO>> GetLinkedOrdersAsync(int[] orderIds)
{
return await Context.Queryable<OrderLinkEntity>()
.LeftJoin<OrderEntity>((a, b) => a.LinkOrderId == b.Id)
.Where(a => a.OrderId == orderId)
.Where(a => orderIds.Contains(a.OrderId))
.Select((a, b) => new LinkedOrderVO
{
OwnerOrderId = a.OrderId,
OrderId = b.Id,
OrderType = b.OrderType,
OrderNo = b.OrderNo,
SubSystemId = b.SubSystemId
OrderType = b.OrderType,
SubSystemId = b.SubSystemId,
CustomId = b.CustomId
})
.ToListAsync();
}

View File

@ -1,7 +1,7 @@
using Nest;
using SJZY.Expand.ABP.Core.Logger;
using SJZY.Expand.ElasticSearch.Services;
using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate;
using SJZY.Track.Domain.Consts;
using SJZY.Track.Domain.Dtos.TrackDetails;
namespace SJZY.Track.Repository.Repositories;
@ -13,65 +13,197 @@ internal class TrackDetailsRepository : ElasticSearchContextAbstract<TrackDetail
{
}
public async Task<TrackDetailsIndex?> GetFirstOrDefaultAsync(int orderId, string trackCode)
public async Task<List<TrackDetailsIndex>> GetListAsync(int[] orderIds, string trackCode,
bool forUpsert = false)
{
var r = await Context.SearchAsync<TrackDetailsIndex>(s => s
.Source()
.Query(q =>
q.Term(f => f.OrderId, orderId)
&& q.Term(f => f.TrackCode.Suffix("keyword"), trackCode)
var r = !forUpsert
? await Context.SearchAsync<TrackDetailsIndex>(s => s
.Source()
.Size(1000)
.Query(q => q
.Bool(b => b
.Must(
mu => mu.Terms(t => t.Field(f => f.OrderId).Terms(orderIds)),
mu => mu.Term(t => t.Field(f => f.TrackCode.Suffix("keyword")).Value(trackCode))
)
)
)
)
: await Context.SearchAsync<TrackDetailsIndex>(s => s
.Source(s2 => s2.Includes(fd => fd
.Field(f => f.IndexName)
.Field(f => f.OrderId)
))
.Size(1000)
.Query(q => q
.Bool(b => b
.Must(
mu => mu.Terms(t => t.Field(f => f.OrderId).Terms(orderIds)),
mu => mu.Term(t => t.Field(f => f.TrackCode.Suffix("keyword")).Value(trackCode))
)
)
)
);
if (!r.IsValid)
{
throw new Exception($"轨迹查询异常,原因:{r.DebugInformation}。");
}
return r.ToIdList();
}
public async Task<List<TrackDetailsIndex>> GetLatestListAsync(int[] orderIds, List<string> trackCodes)
{
var excludeTrackCodes = trackCodes.ToList();
excludeTrackCodes.Add(GlobalConst.EmptyString);
var r = await Context.SearchAsync<TrackDetailsIndex>(x => x
.Source(s => s.Includes(fd => fd
.Field(f => f.TrackEnContent)
.Field(f => f.TrackContent)
.Field(f => f.TrackPlace)
.Field(f => f.TrackTime)
.Field(f => f.TrackTimeZone)
.Field(f => f.OrderId)
.Field(f => f.TrackCode)
))
.Query(q => q
.Bool(b => b
.Must(
mu => mu.Terms(t => t.Field(f => f.OrderId).Terms(orderIds))
)
.MustNot(mn => mn
.Terms(t => t
.Field(f => f.TrackCode.Suffix("keyword"))
.Terms(excludeTrackCodes)
)
)
)
)
.Size(orderIds.Length)
.Aggregations(a => a
.Terms("orders", t => t
.Field(f => f.OrderId)
.Aggregations(aa => aa
.TopHits("latest_track", th => th
.Size(1)
.Sort(so => so
.Field(f => f.ZeroTrackTime, SortOrder.Descending)
)
)
)
)
)
);
var res = r.Documents.FirstOrDefault();
if (res != default && r.Hits.FirstOrDefault() != default)
if (!r.IsValid)
{
res.Id = r.Hits.FirstOrDefault()!.Id;
throw new Exception($"轨迹查询异常,原因:{r.DebugInformation}。");
}
return res;
return r.Documents.ToList();
}
public async Task<bool> AnyAsync(int orderId, DateTime trackTime)
{
var r = await Context.SearchAsync<TrackDetailsIndex>(x => x
.Source(s => s.Includes(fd => fd.Field(f => f.Id)))
.Source(s => s.Includes(fd => fd.Field(f => f.TrackCode)))
.Query(q =>
q.Term(f => f.OrderId, orderId)
&& q.Term(f => f.TrackTime, trackTime)
)
);
if (!r.IsValid)
{
throw new Exception($"轨迹查询异常,原因:{r.DebugInformation}。");
}
return r.Hits.Count > 0;
}
public async Task UpsertAsync(TrackDetailsIndex index)
public async Task<bool> AnyAsync(int orderId, string originTrackData)
{
if (index.Id.IsNullOrWhiteSpace())
{
// 按天写入es
var r = await Context.IndexAsync(index, request => request
.Index(index.IndexName)
//.Refresh(Refresh.True) // 可选:立即刷新索引使更改立即可见
);
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.OriginTrackData.Suffix("keyword"), originTrackData)
)
);
if (!r.IsValid)
{
LogHelper.Error($"轨迹新增失败,原因:{r.DebugInformation}。");
}
if (!r.IsValid)
{
throw new Exception($"轨迹查询异常,原因:{r.DebugInformation}。");
}
else
{
var r = await Context.UpdateAsync<TrackDetailsIndex>(index.Id, u => u
.Index(index.IndexName)
.Doc(index)
//.Refresh(Refresh.True) // 可选:立即刷新索引使更改立即可见
);
if (!r.IsValid)
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;
}
public async Task UpdateAsync(TrackDetailsIndex track)
{
var r = await Context.UpdateAsync<TrackDetailsIndex, object>(track.Id, u => u
.Index(track.IndexName)
.Doc(new
{
LogHelper.Error($"轨迹更新失败,原因:{r.DebugInformation}。");
}
track.TrackPlace,
track.TrackTime,
track.TrackTimeZone,
track.UpdateUser,
UpdateTime = DateTime.Now,
})
);
if (!r.IsValid)
{
throw new Exception($"轨迹更新失败,原因:{r.DebugInformation}。");
}
}
public async Task AddManyAsync(List<TrackDetailsIndex> tracks)
{
if (!tracks.Any()) return;
// 按天写入es
var indexName = $"{GlobalConst.TrackIndex}-{DateTime.Now:yyyy.MM.dd}";
tracks.ForEach(x => x.IndexName = indexName);
var r = await Context.IndexManyAsync(tracks, indexName);
if (!r.IsValid)
{
throw new Exception($"{tracks.Count}条轨迹新增失败,原因:{r.DebugInformation}。");
}
}
public async Task AddAsync(TrackDetailsIndex track)
{
// 按天写入es
var indexName = $"{GlobalConst.TrackIndex}-{DateTime.Now:yyyy.MM.dd}";
track.IndexName = indexName;
var r = await Context.IndexAsync(track, s => s.Index(indexName));
if (!r.IsValid)
{
throw new Exception($"轨迹新增失败,原因:{r.DebugInformation}。");
}
}
@ -79,7 +211,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)
@ -96,85 +228,58 @@ internal class TrackDetailsRepository : ElasticSearchContextAbstract<TrackDetail
.Terms(orderNos))
&& q.Term(f => f.OrderType, orderType)
)
.Size(1000)
);
return data.Documents.ToList();
}
if (!r.IsValid)
{
throw new Exception($"轨迹查询异常,原因:{r.DebugInformation}。");
}
public async Task<List<TrackDetailsIndex>> GetListAsync(int orderType, List<string> orderNos, string trackCode)
{
if (orderNos.Count < 1) return new List<TrackDetailsIndex>();
var data = await Context.SearchAsync<TrackDetailsIndex>(x => x
.Source(s => s.Includes(fd => fd
.Field(f => f.OrderId)
.Field(f => f.OrderNo)
.Field(f => f.TrackStatus)
.Field(f => f.TrackCode)
.Field(f => f.TrackContent)
.Field(f => f.TrackPlace)
.Field(f => f.TrackTime)
.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.Term(f => f.TrackCode.Suffix("keyword"), trackCode)
)
);
return data.Documents.ToList();
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.TrackTime, SortOrder.Descending)));
return data.Documents.ToList();
}
public async Task<TrackDetailsIndex?> GetLatestAsync(int orderId)
{
var data = await Context.SearchAsync<TrackDetailsIndex>(x => x
.Source(s => s.Includes(fd => fd
.Field(f => f.TrackEnContent)
.Field(f => f.TrackContent)
.Field(f => f.TrackPlace)
.Field(f => f.TrackTime)
.Field(f => f.TrackTimeZone)
))
.Query(q =>
q.Term(f => f.OrderId, orderId)
)
.Sort(s => s.Field(f => f.TrackTime, SortOrder.Descending))
.Size(1)
);
return data.Documents.FirstOrDefault();
}
public async Task<bool> DeleteTrackAsync(int orderId, List<string> trackCodes)
{
var r = await Context.DeleteByQueryAsync<TrackDetailsIndex>(x => x
.Query(
q =>
q.Term(t => t.OrderId, orderId)
&& q.Term(t => t.IsCanDelete, true)
&& q.Terms(t => t.Field(f => f.TrackCode.Suffix("keyword")).Terms(trackCodes)))
//.Refresh(Refresh.True) // 可选:立即刷新索引使更改立即可见
);
.Sort(s => s.Field(f => f.ZeroTrackTime, SortOrder.Descending)));
if (!r.IsValid)
throw new Exception($"轨迹数据删除失败,原因:{r.DebugInformation}");
{
throw new Exception($"轨迹查询异常,原因:{r.DebugInformation}。");
}
return r.Deleted > 0;
return r.Documents.ToList();
}
public async Task DeleteTrackAsync(int[] orderIds, List<string> trackCodes)
{
var search = await Context.SearchAsync<TrackDetailsIndex>(x => x
.Source(s => s.Includes(fd => fd.Field(f => f.IndexName)))
.Size(orderIds.Length * trackCodes.Count * 2)
.Query(q =>
q.Terms(t => t.Field(f => f.OrderId).Terms(orderIds))
&& q.Terms(t => t.Field(f => f.TrackCode.Suffix("keyword")).Terms(trackCodes))
)
);
if (!search.IsValid)
{
throw new Exception($"轨迹查询异常,原因:{search.DebugInformation}。");
}
var tracks = search.ToIdList();
foreach (var track in tracks)
{
await Context.DeleteAsync<TrackDetailsIndex>(track.Id, u => u
.Index(track.IndexName)
);
}
}
public async Task<(List<TrackDetailsIndex> list, long total)> GetListAsync(TrackDetailsQueryInput input)

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,

View File

@ -166,4 +166,29 @@ public class TrackPushConfigRepository : BasicRepository<TrackPushConfigEntity>,
return await Context.Updateable<TrackPushContentConfigEntity>().SetColumns(columns).Where(whereExpression)
.ExecuteCommandHasChangeAsync();
}
/// <summary>
/// 查询轨迹推送关系
/// </summary>
/// <param name="input"></param>
/// <returns></returns>
public async Task<List<QueryTrackPullMappingsInfoVO>> QueryTrackPullMappingsAsync(QueryTrackPullMappingsInputVO input)
{
var result = await Context.Queryable<TrackPushConfigEntity>().LeftJoin<TrackPushContentConfigEntity>((a, b) => a.Id == b.TrackPushConfigId && !b.IsDeleted)
.LeftJoin<TrackContentEntity>((a, b, c) => b.TrackContentId == c.Id && !c.IsDeleted)
.Where((a, b, c) => !a.IsDeleted)
.Where((a, b, c) => a.SubSystemId == input.SubsystemsId && a.CustomId == input.CustomerId)
.Select((a, b, c) => new QueryTrackPullMappingsInfoVO
{
Code = c.Code,
OrderType = b.OrderNoType,
ServiceCode = b.TrackContentCodeValue,
Status = b.TrackStatusValue,
Content = b.TrackContentValue,
ContentEn = b.TrackEnContentValue
})
.ToListAsync();
return result;
}
}

View File

@ -16,6 +16,8 @@ public class TrackPushRecordsRepository : ElasticSearchContextAbstract<TrackDeta
public async Task AddAsync(TrackPushRecordIndex index)
{
index.IndexName = $"{GlobalConst.TrackPushRecordsIndex}-{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

View File

@ -3,13 +3,13 @@
<PropertyGroup>
<TargetFramework>net7.0</TargetFramework>
<GenerateDocumentationFile>True</GenerateDocumentationFile>
<LangVersion>preview</LangVersion>
</PropertyGroup>
<ItemGroup>
<PackageReference Include="SJZY.LoggerCenter.HttpApi" Version="7.0.5" />
<PackageReference Include="SJZY.OTel.Extension" Version="1.0.4"/>
<PackageReference Include="Volo.Abp.Autofac" Version="7.0.2" />
<PackageReference Include="SJZY.UserCenter.HttpApi" Version="3.0.3-alpha8" />
<PackageReference Include="SJZY.UserCenter.HttpApi" Version="3.0.9-alpha"/>
<PackageReference Include="SJZY.Expand.CAP" Version="7.8.0" />
</ItemGroup>