update:广播服务抽象出接口

This commit is contained in:
hardstylewyl 2024-12-30 09:41:56 +08:00
parent c4ceca39c4
commit 6e0ff29614
8 changed files with 192 additions and 21 deletions

View File

@ -0,0 +1,89 @@
namespace SJZY.Track.Application.Contract.Dtos.TrackPush;
public class TrackPushRecordDto
{
/// <summary>
/// 子系统代码
/// </summary>
public string SubSystemCode { get; set; }
/// <summary>
/// 是否为手动推送
/// </summary>
public bool IsManualPush { get; set; }
/// <summary>
/// 推送时间
/// </summary>
public DateTime PushTime { get; set; }
/// <summary>
/// 操作人
/// </summary>
public string? OperationMan { get; set; }
/// <summary>
/// 订单号
/// </summary>
public string OrderNo { get; set; } = null!;
/// <summary>
/// 接口通请求Id
/// </summary>
public Guid RequestId { get; set; }
/// <summary>
/// 信息
/// </summary>
public string Message { get; set; } = null!;
/// <summary>
/// 操作类型
/// </summary>
public string OptionType { get; set; } = null!;
/// <summary>
/// 轨迹状态
/// </summary>
public string TrackStatus { get; set; } = null!;
/// <summary>
/// 轨迹内容代码
/// </summary>
public string TrackCode { get; set; } = null!;
/// <summary>
/// 轨迹内容
/// </summary>
public string TrackContent { 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; }
/// <summary>
/// 是否成功
/// </summary>
public bool IsSuccess { get; set; }
/// <summary>
/// 推送内容(json)
/// </summary>
public string PushContent { get; set; } = null!;
/// <summary>
/// 轨迹的额外新
/// </summary>
public string? Extra { get; set; }
}

View File

@ -0,0 +1,22 @@
using SJZY.Track.Application.Contract.Dtos.TrackPush;
using SJZY.Track.Application.Contract.Events;
using Volo.Abp.DependencyInjection;
namespace SJZY.Track.Application.Contract.Services;
public interface IBroadcastingService :ITransientDependency
{
/// <summary>
/// 轨迹变更广播
/// </summary>
/// <param name="eto"></param>
/// <param name="subSystemCode"></param>
/// <param name="trackStatusCode"></param>
void TrackChangePublicTopicAsync(TrackUpsertEvent eto, string subSystemCode, string trackStatusCode);
/// <summary>
/// 轨迹推送完成广播
/// </summary>
/// <param name="log"></param>
void TrackPushCompletedPublicTopicAsync(TrackPushRecordDto log);
}

View File

@ -2,7 +2,7 @@ using DotNetCore.CAP;
using SJZY.Expand.ABP.Core.Logger; using SJZY.Expand.ABP.Core.Logger;
using SJZY.Expand.ABP.Core.PublicMethords; using SJZY.Expand.ABP.Core.PublicMethords;
using SJZY.Track.Application.Contract.Events; using SJZY.Track.Application.Contract.Events;
using SJZY.Track.Application.Services.Broadcasting; using SJZY.Track.Application.Contract.Services;
using SJZY.Track.Domain.Aggregates.OrderAggregate; using SJZY.Track.Domain.Aggregates.OrderAggregate;
using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate; using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate;
using SJZY.Track.Domain.Dtos.TrackContents; using SJZY.Track.Domain.Dtos.TrackContents;
@ -19,7 +19,7 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe
private readonly IOrderRepository _orderRepository; private readonly IOrderRepository _orderRepository;
private readonly IProvideQueries _provideQueries; private readonly IProvideQueries _provideQueries;
private readonly ITrackDetailsEsRepository _trackDetailsEsRepository; private readonly ITrackDetailsEsRepository _trackDetailsEsRepository;
private readonly BroadcastingService _broadcastingService; private readonly IBroadcastingService _broadcastingService;
private readonly IOrderOperationRecordEsRepository _orderOperationRecordEsRepository; private readonly IOrderOperationRecordEsRepository _orderOperationRecordEsRepository;
public TrackUpsertEventHandler( public TrackUpsertEventHandler(
@ -28,7 +28,7 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe
IProvideQueries provideQueries, IProvideQueries provideQueries,
ITrackDetailsEsRepository trackDetailsEsRepository, ITrackDetailsEsRepository trackDetailsEsRepository,
IOrderOperationRecordEsRepository orderOperationRecordEsRepository, IOrderOperationRecordEsRepository orderOperationRecordEsRepository,
BroadcastingService broadcastingService) IBroadcastingService broadcastingService)
{ {
_orderRepository = orderRepository; _orderRepository = orderRepository;
_capPublisher = capPublisher; _capPublisher = capPublisher;
@ -50,8 +50,7 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe
// 处理广播事件 // 处理广播事件
if (content != null && (content.IsToMq || @event.IsToMq)) if (content != null && (content.IsToMq || @event.IsToMq))
// todo: review BroadcastingService抽一层接口,面向抽象编程 _broadcastingService.TrackChangePublicTopicAsync(@event, content.SubSystemCode, content.TrackStatusCode);
_broadcastingService.TrackChangePublicTopicAsync(@event, content);
if (canNextOrderIds.Count == 0) return; if (canNextOrderIds.Count == 0) return;
// 根据订单关联关系异步向下刷(异步) // 根据订单关联关系异步向下刷(异步)

View File

@ -6,8 +6,8 @@ using RabbitMQ.Client.Events;
using SJZY.Expand.ABP.Core.Logger; using SJZY.Expand.ABP.Core.Logger;
using SJZY.InterfaceRelay.Application.Contracts; using SJZY.InterfaceRelay.Application.Contracts;
using SJZY.Track.Application.Contract.Dtos.TrackPush; using SJZY.Track.Application.Contract.Dtos.TrackPush;
using SJZY.Track.Application.Contract.Services;
using SJZY.Track.Application.Helpers; using SJZY.Track.Application.Helpers;
using SJZY.Track.Application.Services.Broadcasting;
using SJZY.Track.Domain.Aggregates.OrderAggregate; using SJZY.Track.Domain.Aggregates.OrderAggregate;
using SJZY.Track.Domain.Aggregates.TrackPushAggregate; using SJZY.Track.Domain.Aggregates.TrackPushAggregate;
using SJZY.Track.Domain.Enums; using SJZY.Track.Domain.Enums;
@ -19,11 +19,11 @@ public class TrackPushCompletedEventHandler : ITransientDependency, IRabbitHandl
{ {
private readonly IOrderRepository _orderRepository; private readonly IOrderRepository _orderRepository;
private readonly ITrackPushRecordEsRepository _trackPushRecordEsRepository; private readonly ITrackPushRecordEsRepository _trackPushRecordEsRepository;
private readonly BroadcastingService _broadcastingService; private readonly IBroadcastingService _broadcastingService;
public TrackPushCompletedEventHandler(IOrderRepository orderRepository, public TrackPushCompletedEventHandler(IOrderRepository orderRepository,
ITrackPushRecordEsRepository trackPushRecordEsRepository, ITrackPushRecordEsRepository trackPushRecordEsRepository,
BroadcastingService broadcastingService) IBroadcastingService broadcastingService)
{ {
_orderRepository = orderRepository; _orderRepository = orderRepository;
_trackPushRecordEsRepository = trackPushRecordEsRepository; _trackPushRecordEsRepository = trackPushRecordEsRepository;
@ -70,7 +70,7 @@ public class TrackPushCompletedEventHandler : ITransientDependency, IRabbitHandl
// 推送广播 // 推送广播
if (log.IsToMq) if (log.IsToMq)
_broadcastingService.TrackPushCompletedPublicTopicAsync(log); _broadcastingService.TrackPushCompletedPublicTopicAsync(BuildDto(log));
// 往es 写推送日志 // 往es 写推送日志
await _trackPushRecordEsRepository.AddAsync(log); await _trackPushRecordEsRepository.AddAsync(log);
@ -119,4 +119,28 @@ public class TrackPushCompletedEventHandler : ITransientDependency, IRabbitHandl
}); });
} }
} }
private TrackPushRecordDto BuildDto(TrackPushRecordIndex log)
{
return new TrackPushRecordDto
{
SubSystemCode = log.SubSystemCode,
RequestId = log.RequestId,
IsSuccess = log.IsSuccess,
Message = log.Message,
OperationMan = log.CreateUser,
PushTime = DateTime.Now,
OptionType = log.OptionType,
IsManualPush = !string.Equals(log.CreateUser, GlobalConst.SystemUser), //操作人不为系统用户,则认为是仍推送
PushContent = log.PushContent,
OrderNo = log.OrderNo,
TrackStatus = log.TrackStatus,
TrackContent = log.TrackContent,
TrackTimeZone = log.TrackTimeZone,
TrackTime = log.TrackTime,
TrackPlace = log.TrackPlace,
TrackCode = log.TrackCode,
Extra = log.Extra
};
}
} }

View File

@ -1,17 +1,16 @@
using System.Text; using System.Text;
using BSI.Stack.Rabbit; using BSI.Stack.Rabbit;
using SJZY.Expand.ABP.Core.PublicMethords; using SJZY.Expand.ABP.Core.PublicMethords;
using SJZY.Track.Application.Contract.Dtos.TrackPush;
using SJZY.Track.Application.Contract.Events; using SJZY.Track.Application.Contract.Events;
using SJZY.Track.Domain.Aggregates.TrackPushAggregate; using SJZY.Track.Application.Contract.Services;
using SJZY.Track.Domain.Dtos.TrackContents;
using Volo.Abp.DependencyInjection;
namespace SJZY.Track.Application.Services.Broadcasting; namespace SJZY.Track.Application.Services.Broadcasting;
/// <summary> /// <summary>
/// 广播服务 /// 广播服务
/// </summary> /// </summary>
public class BroadcastingService : ITransientDependency public class BroadcastingService : IBroadcastingService
{ {
private readonly RabbitClientBus _rabbitClientBus; private readonly RabbitClientBus _rabbitClientBus;
@ -24,9 +23,10 @@ public class BroadcastingService : ITransientDependency
/// 轨迹变更广播 /// 轨迹变更广播
/// </summary> /// </summary>
/// <param name="eto"></param> /// <param name="eto"></param>
/// <param name="content"></param> /// <param name="subSystemCode">子系统代码</param>
/// <param name="trackStatusCode">轨迹状态代码</param>
/// <returns></returns> /// <returns></returns>
public void TrackChangePublicTopicAsync(TrackUpsertEvent eto, TrackContentVO content) public void TrackChangePublicTopicAsync(TrackUpsertEvent eto, string subSystemCode, string trackStatusCode)
{ {
var data = new TrackChangeEto var data = new TrackChangeEto
{ {
@ -45,7 +45,7 @@ public class BroadcastingService : ITransientDependency
TrackTimeZone = eto.TrackTimeZone, TrackTimeZone = eto.TrackTimeZone,
TrackTime = eto.TrackTime, TrackTime = eto.TrackTime,
TrackPlace = eto.TrackPlace, TrackPlace = eto.TrackPlace,
TrackStatusCode = content.TrackStatusCode, TrackStatusCode = trackStatusCode,
EnStatus = eto.EnStatus, EnStatus = eto.EnStatus,
Content = eto.Content, Content = eto.Content,
TrackCode = eto.TrackCode, TrackCode = eto.TrackCode,
@ -61,7 +61,7 @@ public class BroadcastingService : ITransientDependency
else else
data.Option = 3; data.Option = 3;
_rabbitClientBus.Publish(GlobalConst.TrackChangeExchange, $"TrackChange.{content.SubSystemCode}", _rabbitClientBus.Publish(GlobalConst.TrackChangeExchange, $"TrackChange.{subSystemCode}",
Encoding.UTF8.GetBytes(data.ToJson())); Encoding.UTF8.GetBytes(data.ToJson()));
} }
@ -70,17 +70,17 @@ public class BroadcastingService : ITransientDependency
/// </summary> /// </summary>
/// <param name="log"></param> /// <param name="log"></param>
/// <returns></returns> /// <returns></returns>
public void TrackPushCompletedPublicTopicAsync(TrackPushRecordIndex log) public void TrackPushCompletedPublicTopicAsync(TrackPushRecordDto log)
{ {
var data = new TrackPushCompletedEvent var data = new TrackPushCompletedEvent
{ {
RequestId = log.RequestId, RequestId = log.RequestId,
IsSuccess = log.IsSuccess, IsSuccess = log.IsSuccess,
Message = log.Message, Message = log.Message,
OperationMan = log.CreateUser, OperationMan = log.OperationMan,
PushTime = DateTime.Now, PushTime = log.PushTime,
OptionType = log.OptionType, OptionType = log.OptionType,
IsManualPush = !string.Equals(log.CreateUser, GlobalConst.SystemUser), //操作人不为系统用户,则认为是仍推送 IsManualPush = log.IsManualPush, //操作人不为系统用户,则认为是仍推送
PushContent = log.PushContent, PushContent = log.PushContent,
OrderNo = log.OrderNo, OrderNo = log.OrderNo,
TrackStatus = log.TrackStatus, TrackStatus = log.TrackStatus,

View File

@ -2,6 +2,11 @@ namespace SJZY.Track.Domain.Dtos.TrackContents;
public class TrackContentVO public class TrackContentVO
{ {
/// <summary>
/// 子系统Id
/// </summary>
public int SubSystemId { get; set; }
/// <summary> /// <summary>
/// 子系统代码 /// 子系统代码
/// </summary> /// </summary>

View File

@ -52,6 +52,14 @@ public interface IProvideQueries
/// <returns></returns> /// <returns></returns>
Task<TrackContentVO?> GetTrackContentAsync(int subSystemId, string trackCode); Task<TrackContentVO?> GetTrackContentAsync(int subSystemId, string trackCode);
/// <summary>
/// 获取轨迹内容
/// </summary>
/// <param name="subSystemIds">子系统ids</param>
/// <param name="trackCodes">轨迹代码集合</param>
/// <returns></returns>
Task<List<TrackContentVO>> GetTrackContentsAsync(int[] subSystemIds, string[] trackCodes);
/// <summary> /// <summary>
/// 获取所有拉取配置 /// 获取所有拉取配置
/// </summary> /// </summary>

View File

@ -117,6 +117,30 @@ public class ProvideQueries : ISugarRepository, IProvideQueries, ITransientDepen
.FirstAsync(); .FirstAsync();
} }
public async Task<List<TrackContentVO>> GetTrackContentsAsync(int[] subSystemIds, string[] trackCodes)
{
return await Context.Queryable<TrackContentEntity>()
.LeftJoin<TrackStatusEntity>((x, y) => x.TrackStatusId == y.Id)
.LeftJoin<VSystemEntity>((x, y, z) => x.SubSystemId == z.Id)
.Where(x => subSystemIds.Contains(x.SubSystemId) && trackCodes.Contains(x.Code))
.Select((x, y, z) => new TrackContentVO
{
SubSystemId = x.SubSystemId,
SubSystemCode = z.Key,
TrackStatusId = x.TrackStatusId,
TrackStatus = y.Status,
TrackStatusCode = y.Code,
TrackContent = x.Content,
TrackEnContent = x.EnContent,
IsCompleted = y.IsCompleted,
IsRepeatable = x.IsRepeatable,
IsAllowManuallyUpdate = x.IsAllowManuallyUpdate,
IsAllowManuallyCreate = x.IsAllowManuallyCreate,
IsToMq = y.IsToMq,
})
.ToListAsync();
}
public async Task<List<TrackPullConfigVo>> GetAllPullConfigsAsync() public async Task<List<TrackPullConfigVo>> GetAllPullConfigsAsync()
{ {
return await Context.Queryable<TrackPullConfigEntity>() return await Context.Queryable<TrackPullConfigEntity>()