diff --git a/src/SJZY.Track.Application.Contract/Events/TrackChangeEto.cs b/src/SJZY.Track.Application.Contract/Events/TrackChangeEto.cs
new file mode 100644
index 00000000..9d3a0a49
--- /dev/null
+++ b/src/SJZY.Track.Application.Contract/Events/TrackChangeEto.cs
@@ -0,0 +1,111 @@
+using SJZY.Track.Application.Contract.Dtos.ProvideServices;
+
+namespace SJZY.Track.Application.Contract.Events;
+
+public class TrackChangeEto
+{
+ ///
+ /// 操作类型:1新增,2删除,3修改
+ ///
+ public int Option { get; set; }
+
+ ///
+ /// 子系统id
+ ///
+ public int SubSystemId { get; set; }
+
+ ///
+ /// 客户id
+ ///
+ public int CustomId { get; set; }
+
+ ///
+ /// 订单id
+ ///
+ public OrderIdentifier[] OrderIdentifiers { get; set; } = null!;
+
+ ///
+ /// 操作人
+ ///
+ public string? OperationMan { get; set; }
+
+ ///
+ /// 轨迹内容代码
+ ///
+ public string TrackCode { get; set; } = null!;
+
+ ///
+ /// 轨迹内容
+ ///
+ public string? Content { get; set; }
+
+ ///
+ /// 轨迹内容英文
+ ///
+ public string? EnStatus { get; set; }
+
+ ///
+ /// 轨迹状态代码
+ ///
+ public string TrackStatusCode { get; set; }
+
+ ///
+ /// 轨迹发生地
+ ///
+ public string TrackPlace { get; set; } = null!;
+
+ ///
+ /// 轨迹发生时间
+ ///
+ public DateTime TrackTime { get; set; }
+
+ ///
+ /// 轨迹发生地时区
+ ///
+ public float TrackTimeZone { get; set; }
+
+ ///
+ /// 客户是否可见
+ ///
+ public bool IsVisibility { get; set; }
+
+ ///
+ /// 数据来源:1外部服务商创建,2内部系统创建,3人工创建
+ ///
+ public int Source { get; set; }
+
+ ///
+ /// 轨迹备注
+ ///
+ public string? Remarks { get; set; }
+
+ ///
+ /// 附件集合
+ ///
+ public List? Files { get; set; }
+
+ ///
+ /// 轨迹发生地(城市)
+ ///
+ public string? TrackPlaceCity { get; set; }
+
+ ///
+ /// 轨迹发生地(省/州)
+ ///
+ public string? TrackPlaceProvince { get; set; }
+
+ ///
+ /// 轨迹发生地(国家/地区)
+ ///
+ public string? TrackPlaceNation { get; set; }
+
+ ///
+ /// 轨迹发生地邮编
+ ///
+ public string? TrackPlacePostcode { get; set; }
+
+ ///
+ /// 预计到达时间
+ ///
+ public DateTime? EstimatedDeliveryDate { get; set; }
+}
diff --git a/src/SJZY.Track.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs b/src/SJZY.Track.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs
index a5035d63..9e8a148b 100644
--- a/src/SJZY.Track.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs
+++ b/src/SJZY.Track.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs
@@ -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();
- await HandleTrackAsync(@event, canNextOrderIds);
+
+ var content = await HandleTrackAsync(@event, canNextOrderIds);
+
+ // 处理广播事件
+ PublicTrackTopic(@event, content);
if (canNextOrderIds.Count == 0) return;
// 根据订单关联关系异步向下刷(异步)
await DownByOrderLink(@event, canNextOrderIds);
}
+ ///
+ /// 发送轨迹广播
+ ///
+ ///
+ ///
+ 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 canNextOrderIds)
{
// 根据订单关联关系异步向下刷(异步)
@@ -126,7 +183,7 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe
return true;
}
- private async Task HandleTrackAsync(TrackUpsertEvent @event, List canNextOrderIds)
+ private async Task HandleTrackAsync(TrackUpsertEvent @event, List 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> UpsertTrackAsync(TrackUpsertEvent @event, List canNextOrderIds,
diff --git a/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs b/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs
index 44486d40..e36b86ce 100644
--- a/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs
+++ b/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs
@@ -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(options =>
- {
- options.FileSets.AddEmbedded("SJZY.Track.Application");
- });
+ Configure(options => { options.FileSets.AddEmbedded("SJZY.Track.Application"); });
//添加多语言
var languageConfig = configuration.GetSection("Language");
@@ -82,45 +80,59 @@ public class SjzyTrackApplicationModule : AbpModule
//CAP
context.Services.AddSingleton();
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("RabbitMQ:Port");
- configure.BasicQosOptions =
- new RabbitMQOptions.BasicQos(configuration.GetValue("CAP:PrefetchCount"));
- });
- options.FailedRetryInterval = configuration.GetValue("CAP:FailedRetryInterval");
- options.FailedRetryCount = configuration.GetValue("CAP:FailedRetryCount");
- options.SucceedMessageExpiredAfter = configuration.GetValue("CAP:SucceedMessageExpiredAfter");
- options.FallbackWindowLookbackSeconds =
- configuration.GetValue("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("RabbitMQ:Port");
+ configure.BasicQosOptions =
+ new RabbitMQOptions.BasicQos(configuration.GetValue("CAP:PrefetchCount"));
+ });
+ options.FailedRetryInterval = configuration.GetValue("CAP:FailedRetryInterval");
+ options.FailedRetryCount = configuration.GetValue("CAP:FailedRetryCount");
+ options.SucceedMessageExpiredAfter = configuration.GetValue("CAP:SucceedMessageExpiredAfter");
+ options.FallbackWindowLookbackSeconds =
+ configuration.GetValue("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();
+ }
+ catch (Exception ex)
+ {
+ LogHelper.Error("【严重】CAP失败回调事件发生未处理异常" + ex);
+ }
+ };
+ }).AddSubscribeFilter();
+ }
+ else
+ {
+ //是本地调试环境,用内存存储和内存队列
+ context.Services.AddCap(capOptions =>
+ {
+ capOptions.UseInMemoryStorage();
+ capOptions.UseInMemoryMessageQueue();
+ capOptions.UseDashboard();
+ }).AddSubscribeFilter();
+ 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("TrackApp:TrackRefreshFetchCount"),
+ FailedRequeue = false,
+ Declaration = o =>
+ {
+ o.ExchangeDeclare(GlobalConst.TrackChangeExchange, ExchangeType.Topic, true);
+ }
}
};
}, typeof(SjzyTrackApplicationModule).Assembly);
diff --git a/src/SJZY.Track.Domain/Consts/GlobalConst.MQ.cs b/src/SJZY.Track.Domain/Consts/GlobalConst.MQ.cs
index d517bfc8..e7901efa 100644
--- a/src/SJZY.Track.Domain/Consts/GlobalConst.MQ.cs
+++ b/src/SJZY.Track.Domain/Consts/GlobalConst.MQ.cs
@@ -51,4 +51,11 @@ public partial class GlobalConst
/// 预警计算任务
///
public const string TrackWarningCalcSubscribe = "SJZY.Track.TrackWarningCalc";
+
+ ///
+ /// 轨迹变更通知交换机
+ ///
+ public const string TrackChangeExchange = "sjzy.track.on_track_change_ex";
+
+
}
diff --git a/src/SJZY.Track.Domain/Dtos/TrackContents/TrackContentVO.cs b/src/SJZY.Track.Domain/Dtos/TrackContents/TrackContentVO.cs
index f9f6da60..f084b0a8 100644
--- a/src/SJZY.Track.Domain/Dtos/TrackContents/TrackContentVO.cs
+++ b/src/SJZY.Track.Domain/Dtos/TrackContents/TrackContentVO.cs
@@ -51,4 +51,9 @@ public class TrackContentVO
/// 是否允许手动创建
///
public bool IsAllowManuallyCreate { get; set; }
+
+ ///
+ /// 是否广播
+ ///
+ public bool IsToMq { get; set; }
}
diff --git a/src/SJZY.Track.Repository/Queries/ProvideQueries.cs b/src/SJZY.Track.Repository/Queries/ProvideQueries.cs
index 25d2c0d9..710dd407 100644
--- a/src/SJZY.Track.Repository/Queries/ProvideQueries.cs
+++ b/src/SJZY.Track.Repository/Queries/ProvideQueries.cs
@@ -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();
}