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

添加轨迹变更广播

Browse Source
This commit is contained in:
eli 2024-11-15 16:20:09 +08:00
parent 69e0731b7e
commit 23a1082d80
6 changed files with 250 additions and 44 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

111
src/SJZY.Track.Application.Contract/Events/TrackChangeEto.cs Normal file
View File

@ -0,0 +1,111 @@
using SJZY.Track.Application.Contract.Dtos.ProvideServices;
namespace SJZY.Track.Application.Contract.Events;
public class TrackChangeEto
{
/// <summary>
/// 操作类型：1新增，2删除，3修改
/// </summary>
public int Option { get; set; }
/// <summary>
/// 子系统id
/// </summary>
public int SubSystemId { get; set; }
/// <summary>
/// 客户id
/// </summary>
public int CustomId { get; set; }
/// <summary>
/// 订单id
/// </summary>
public OrderIdentifier[] OrderIdentifiers { get; set; } = null!;
/// <summary>
/// 操作人
/// </summary>
public string? OperationMan { get; set; }
/// <summary>
/// 轨迹内容代码
/// </summary>
public string TrackCode { get; set; } = null!;
/// <summary>
/// 轨迹内容
/// </summary>
public string? Content { get; set; }
/// <summary>
/// 轨迹内容英文
/// </summary>
public string? EnStatus { get; set; }
/// <summary>
/// 轨迹状态代码
/// </summary>
public string TrackStatusCode { get; set; }
/// <summary>
/// 轨迹发生地
/// </summary>
public string TrackPlace { get; set; } = null!;
/// <summary>
/// 轨迹发生时间
/// </summary>
public DateTime TrackTime { get; set; }
/// <summary>
/// 轨迹发生地时区
/// </summary>
public float TrackTimeZone { get; set; }
/// <summary>
/// 客户是否可见
/// </summary>
public bool IsVisibility { get; set; }
/// <summary>
/// 数据来源：1外部服务商创建，2内部系统创建，3人工创建
/// </summary>
public int Source { get; set; }
/// <summary>
/// 轨迹备注
/// </summary>
public string? Remarks { get; set; }
/// <summary>
/// 附件集合
/// </summary>
public List<FileItemDto>? Files { get; set; }
/// <summary>
/// 轨迹发生地（城市）
/// </summary>
public string? TrackPlaceCity { get; set; }
/// <summary>
/// 轨迹发生地（省/州）
/// </summary>
public string? TrackPlaceProvince { get; set; }
/// <summary>
/// 轨迹发生地（国家/地区）
/// </summary>
public string? TrackPlaceNation { get; set; }
/// <summary>
/// 轨迹发生地邮编
/// </summary>
public string? TrackPlacePostcode { get; set; }
/// <summary>
/// 预计到达时间
/// </summary>
public DateTime? EstimatedDeliveryDate { get; set; }
}

71
src/SJZY.Track.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs
View File

@ -1,9 +1,12 @@
using System.Text;
using BSI.Stack.Rabbit;
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.Aggregates.TrackStatusAggregate;
using SJZY.Track.Domain.Dtos.TrackContents;
using SJZY.Track.Domain.Enums;
using SJZY.Track.Domain.Queries;
@ -18,14 +21,21 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe
private readonly IOrderRepository _orderRepository;
private readonly IProvideQueries _provideQueries;
private readonly ITrackDetailsEsRepository _trackDetailsEsRepository;
private readonly RabbitClientBus _rabbitClientBus;
public TrackUpsertEventHandler(IOrderRepository orderRepository, ICapPublisher capPublisher,
IProvideQueries provideQueries, ITrackDetailsEsRepository trackDetailsEsRepository)
public TrackUpsertEventHandler(
IOrderRepository orderRepository,
ICapPublisher capPublisher,
IProvideQueries provideQueries,
ITrackDetailsEsRepository trackDetailsEsRepository,
ITrackStatusRepository trackStatusRepository,
RabbitClientBus rabbitClientBus)
{
_orderRepository = orderRepository;
_capPublisher = capPublisher;
_provideQueries = provideQueries;
_trackDetailsEsRepository = trackDetailsEsRepository;
_rabbitClientBus = rabbitClientBus;
}
[CapSubscribe(GlobalConst.TrackUpsertSubscribe, Group = GlobalConst.TrackUpsertSubscribe)]
@ -35,13 +45,60 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe
if (!CanDelete(@event)) return;
var canNextOrderIds = new List<int>();
await HandleTrackAsync(@event, canNextOrderIds);
var content = await HandleTrackAsync(@event, canNextOrderIds);
// 处理广播事件
PublicTrackTopic(@event, content);
if (canNextOrderIds.Count == 0) return;
// 根据订单关联关系异步向下刷（异步）
await DownByOrderLink(@event, canNextOrderIds);
}
/// <summary>
/// 发送轨迹广播
/// </summary>
/// <param name="eto"></param>
/// <param name="content"></param>
private void PublicTrackTopic(TrackUpsertEvent eto, TrackContentVO? content)
{
// var trackStatus = await _trackStatusRepository.GetFirstAsync(m => m.Code == content.TrackStatusCode);
if (content == null || !content.IsToMq)
return;
var data = new TrackChangeEto()
{
EstimatedDeliveryDate = eto.EstimatedDeliveryDate,
TrackPlacePostcode = eto.TrackPlacePostcode,
TrackPlaceNation = eto.TrackPlaceNation,
TrackPlaceProvince = eto.TrackPlaceProvince,
TrackPlaceCity = eto.TrackPlaceCity,
Files = eto.Files,
Remarks = eto.Remarks,
Source = eto.Source,
IsVisibility = eto.IsVisibility,
TrackTimeZone = eto.TrackTimeZone,
TrackTime = eto.TrackTime,
TrackPlace = eto.TrackPlace,
TrackStatusCode = content.TrackStatusCode,
EnStatus = eto.EnStatus,
Content = eto.Content,
TrackCode = eto.TrackCode,
OperationMan = eto.OperationMan,
OrderIdentifiers = eto.OrderIdentifiers,
CustomId = eto.CustomId,
SubSystemId = eto.SubSystemId
};
if (eto.IsAdd)
data.Option = 1;
else
data.Option = 3;
_rabbitClientBus.Publish(GlobalConst.TrackChangeExchange, $"TrackChange.{content.SubSystemCode}", Encoding.UTF8.GetBytes(data.ToJson()));
}
private async Task DownByOrderLink(TrackUpsertEvent @event, List<int> canNextOrderIds)
{
// 根据订单关联关系异步向下刷（异步）
@ -126,7 +183,7 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe
return true;
}
private async Task HandleTrackAsync(TrackUpsertEvent @event, List<int> canNextOrderIds)
private async Task<TrackContentVO?> HandleTrackAsync(TrackUpsertEvent @event, List<int> canNextOrderIds)
{
var trackContent = await _provideQueries.GetTrackContentAsync(@event.SubSystemId, @event.TrackCode);
@ -134,12 +191,12 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe
{
// 找不到轨迹内容，直接结束
LogHelper.Warn($"找不到轨迹内容：{@event}", true);
return;
return default;
}
var insertTracks = await UpsertTrackAsync(@event, canNextOrderIds, trackContent);
if (canNextOrderIds.Count == 0) return;
if (canNextOrderIds.Count == 0) return trackContent;
if (insertTracks.Any())
{
@ -195,6 +252,8 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe
}
};
await _capPublisher.PublishAsync(GlobalConst.TrackPushSubscribe, trackPushEvent);
return trackContent;
}
private async Task<List<TrackDetailsIndex>> UpsertTrackAsync(TrackUpsertEvent @event, List<int> canNextOrderIds,

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

@ -3,6 +3,7 @@ using DotNetCore.CAP;
using DotNetCore.CAP.Internal;
using Microsoft.Extensions.Configuration;
using RabbitMQ.Client;
using Savorboard.CAP.InMemoryMessageQueue;
using SJZY.Expand.ABP.Core.Localization;
using SJZY.Expand.ABP.Core.Logger;
using SJZY.Expand.ABP.Core.PublicMethords;
@ -50,10 +51,7 @@ public class SjzyTrackApplicationModule : AbpModule
});
//添加虚拟文件系统
Configure<AbpVirtualFileSystemOptions>(options =>
{
options.FileSets.AddEmbedded<SjzyTrackApplicationModule>("SJZY.Track.Application");
});
Configure<AbpVirtualFileSystemOptions>(options => { options.FileSets.AddEmbedded<SjzyTrackApplicationModule>("SJZY.Track.Application"); });
//添加多语言
var languageConfig = configuration.GetSection("Language");
@ -82,45 +80,59 @@ public class SjzyTrackApplicationModule : AbpModule
//CAP
context.Services.AddSingleton<ISubscribeExecutor, MySubscribeExecutor>();
context.Services.AddCapTableNameService();
context.Services.AddCap(options =>
if (!System.Diagnostics.Debugger.IsAttached)
{
options.UseMySql(configuration["ConnectionStrings:DefaultConnection"]!);
options.UseRabbitMQ(configure =>
context.Services.AddCap(options =>
{
configure.HostName = configuration["RabbitMQ:HostName"]!;
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"));
});
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.UseDashboard(dashboardOptions =>
{
dashboardOptions.PathBase = System.Diagnostics.Debugger.IsAttached
? ""
: configuration["Swagger:JsonUrlPrefix"]!;
});
options.FailedThresholdCallback = faileCalback =>
{
try
options.UseMySql(configuration["ConnectionStrings:DefaultConnection"]!);
options.UseRabbitMQ(configure =>
{
LogHelper.Error($@"
configure.HostName = configuration["RabbitMQ:HostName"]!;
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"));
});
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.UseDashboard(dashboardOptions =>
{
dashboardOptions.PathBase = System.Diagnostics.Debugger.IsAttached
? ""
: configuration["Swagger:JsonUrlPrefix"]!;
});
options.FailedThresholdCallback = faileCalback =>
{
try
{
LogHelper.Error($@"
|-CAP失败回调事件：
|-处理类型：{(faileCalback.MessageType == DotNetCore.CAP.Messages.MessageType.Subscribe ? "消费消息达到最大重试次数，需要开发人员介入" : "生产消息推送失败")}
|-头部信息：{faileCalback.Message.Headers.ToJson()}");
}
catch (Exception ex)
{
LogHelper.Error("【严重】CAP失败回调事件发生未处理异常" + ex);
}
};
}).AddSubscribeFilter<CapFilter>();
}
catch (Exception ex)
{
LogHelper.Error("【严重】CAP失败回调事件发生未处理异常" + ex);
}
};
}).AddSubscribeFilter<CapFilter>();
}
else
{
//是本地调试环境，用内存存储和内存队列
context.Services.AddCap(capOptions =>
{
capOptions.UseInMemoryStorage();
capOptions.UseInMemoryMessageQueue();
capOptions.UseDashboard();
}).AddSubscribeFilter<CapFilter>();
Console.WriteLine("-- 当前为VS本地环境，用内存存储和内存队列");
}
context.Services.AddRabbitBoot(options =>
@ -142,6 +154,17 @@ public class SjzyTrackApplicationModule : AbpModule
o.QueueBind(GlobalConst.TrackRefreshSubscribe, GlobalConst.InterfaceRelayExchange,
GlobalConst.TrackRefreshSubscribe);
}
},
new()
{
//广播轨迹
Queue = GlobalConst.TrackRefreshSubscribe,
FetchCount = configuration.GetValue<ushort>("TrackApp:TrackRefreshFetchCount"),
FailedRequeue = false,
Declaration = o =>
{
o.ExchangeDeclare(GlobalConst.TrackChangeExchange, ExchangeType.Topic, true);
}
}
};
}, typeof(SjzyTrackApplicationModule).Assembly);

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

@ -51,4 +51,11 @@ public partial class GlobalConst
/// 预警计算任务
/// </summary>
public const string TrackWarningCalcSubscribe = "SJZY.Track.TrackWarningCalc";
/// <summary>
/// 轨迹变更通知交换机
/// </summary>
public const string TrackChangeExchange = "sjzy.track.on_track_change_ex";
}

5
src/SJZY.Track.Domain/Dtos/TrackContents/TrackContentVO.cs
View File

@ -51,4 +51,9 @@ public class TrackContentVO
/// 是否允许手动创建
/// </summary>
public bool IsAllowManuallyCreate { get; set; }
/// <summary>
/// 是否广播
/// </summary>
public bool IsToMq { get; set; }
}

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

@ -93,7 +93,8 @@ public class ProvideQueries : ISugarRepository, IProvideQueries, ITransientDepen
IsCompleted = y.IsCompleted,
IsRepeatable = x.IsRepeatable,
IsAllowManuallyUpdate = x.IsAllowManuallyUpdate,
IsAllowManuallyCreate = x.IsAllowManuallyCreate
IsAllowManuallyCreate = x.IsAllowManuallyCreate,
IsToMq = y.IsToMq,
})
.FirstAsync();
}