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

移除自定义 CAP / Rabbit 相关代码 事件处理迁到 Messaging 目录 新增 TrackUpsertForRefreshEvent、Directory.Build.targets、Fake.Local.props #7

Merged
zzh merged 5 commits from test into master 2026-10-27 17:49:00 +08:00
Conversation 0 Commits 5 Files Changed 46 +204 -518
28 changed files with 204 additions and 518 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit 5c7829bdbd - Show all commits

2
Directory.Build.props
View File

@ -4,6 +4,6 @@
<PropertyGroup>
<!-- All Fake packages -->
<!-- 清本地nuget缓存：dotnet nuget locals -c http-cache -->
<FakePackageVersion>8.0.6</FakePackageVersion>
<FakePackageVersion>8.0.6.2</FakePackageVersion>
</PropertyGroup>
</Project>

3
src/TrackCenter.Api/TrackCenter.Api.csproj
View File

@ -20,8 +20,7 @@
<ProjectReference Include="..\TrackCenter.Repository\TrackCenter.Repository.csproj" />
<PackageReference Include="Fake.Autofac" Version="$(FakePackageVersion)" />
<PackageReference Include="Fake.Consul" Version="$(FakePackageVersion)" />
<PackageReference Include="Fake.EventBus.IntegrationEventLog" Version="$(FakePackageVersion)" />
<PackageReference Include="Fake.EventBus.Dashboard" Version="$(FakePackageVersion)" />
<PackageReference Include="Fake.Messaging.Dashboard" Version="$(FakePackageVersion)" />
<PackageReference Include="Pomelo.EntityFrameworkCore.MySql" Version="8.0.3" />
<PackageReference Include="Serilog.AspNetCore" Version="8.0.1" />

25
src/TrackCenter.Api/TrackCenterApiModule.cs
View File

@ -5,8 +5,8 @@ using Fake.AspNetCore.ExceptionHandling;
using Fake.AspNetCore.Newtonsoft;
using Fake.AspNetCore.Swagger;
using Fake.Autofac;
using Fake.EventBus.Dashboard;
using Fake.EventBus.IntegrationEventLog;
using Fake.Messaging.Dashboard;
using Fake.Messaging.EntityFrameworkCore;
using Fake.FeiShu;
using Fake.Json;
using Fake.Modularity;
@ -33,9 +33,8 @@ namespace TrackCenter.Api;
[DependsOn(typeof(FakeAspNetCoreModule))]
[DependsOn(typeof(TrackCenterApplicationModule))]
[DependsOn(typeof(TrackCenterRepositoryModule))]
// RabbitMQ 物理总线之后 Replace IDistributedEventBus → Outbox（与 admin 一致）
[DependsOn(typeof(FakeEventBusIntegrationEventLogModule))]
[DependsOn(typeof(FakeEventBusDashboardModule))]
// Messaging 独立队列（Outbox/Inbox + Dashboard）；物理传输由 FakeMessagingRabbitMqModule 提供
[DependsOn(typeof(FakeMessagingDashboardModule))]
[DependsOn(typeof(FakeSerilogModule))]
public class TrackCenterApiModule : FakeModule
{
@ -59,10 +58,11 @@ public class TrackCenterApiModule : FakeModule
// 配置签名验证选项
services.Configure<SignatureValidationOptions>(configuration.GetSection("SignatureValidation"));
// Outbox / Inbox 与业务同库（EF），通过 DbTransaction 征用 SqlSugar 事务
services.AddDbContext<IntegrationEventLogContext>(options =>
// Outbox / Inbox（fk_msg_*）与业务同库（EF），通过 DbTransaction 征用 SqlSugar 事务
services.AddDbContext<MessagingDbContext>(options =>
{
var connectionString = configuration.GetConnectionString("IntegrationEventLog")
var connectionString = configuration.GetConnectionString("Messaging")
?? configuration.GetConnectionString("IntegrationEventLog")
?? configuration.GetConnectionString("Default")
?? configuration["ConnectionStrings:Default"]
?? string.Empty;
@ -73,6 +73,13 @@ public class TrackCenterApiModule : FakeModule
#endif
});
// Dashboard 路径保持 /eventbus，兼容旧书签与网关
services.Configure<MessagingDashboardOptions>(o =>
{
if (string.IsNullOrWhiteSpace(o.PathMatch) || o.PathMatch == "/messaging")
o.PathMatch = "/eventbus";
});
// 全局授权配置：所有接口默认需要授权
services.AddMvc(options =>
{
@ -207,7 +214,7 @@ public class TrackCenterApiModule : FakeModule
});
// 必须在 UsePathBase 之后：完整对外路径为 {PathBase}/eventbus
app.UseFakeEventBusDashboard();
app.UseFakeMessagingDashboard();
app.Services.GetRequiredService<IFeiShuNotificationService>().Send("Application started");
}

10
src/TrackCenter.Application/Rabbit/InterfaceRelayForwardReplayEvent.cs → src/TrackCenter.Application/Events/InterfaceRelayForwardReplayEvent.cs
View File

@ -1,6 +1,12 @@
namespace TrackCenter.Application.Rabbit;
using Fake.Messaging;
public class InterfaceRelayForwardReplayEvent
namespace TrackCenter.Application.Events;
/// <summary>
/// 接口通回调报文（外部生产，无 Fake MessageId / DLX）。
/// </summary>
[MessageRoute(Topic = GlobalConst.InterfaceRelayExchange, TopicType = "direct")]
public class InterfaceRelayForwardReplayEvent : Message
{
/// <summary>
/// 请求Id

5
src/TrackCenter.Application/Events/OrderPodPullEvent.cs
View File

@ -1,4 +1,4 @@
using Fake.EventBus.Distributed;
using Fake.Messaging;
namespace TrackCenter.Application.Events;
@ -6,7 +6,8 @@ namespace TrackCenter.Application.Events;
/// 订单 POD 拉取（当前仓库无订阅方，仍走 Outbox 投递以兼容后续消费者）。
/// </summary>
[Serializable]
public class OrderPodPullEvent : IntegrationEvent
[MessageRoute(nameof(OrderPodPullEvent))]
public class OrderPodPullEvent : Message
{
// 确定应用
public string AppKey { get; set; }

5
src/TrackCenter.Application/Events/TrackChangeAfterUpdateOrderEvent.cs
View File

@ -1,9 +1,10 @@
using Fake.EventBus.Distributed;
using Fake.Messaging;
namespace TrackCenter.Application.Events;
[Serializable]
public class TrackChangeAfterUpdateOrderEvent : IntegrationEvent
[MessageRoute(nameof(TrackChangeAfterUpdateOrderEvent))]
public class TrackChangeAfterUpdateOrderEvent : Message
{
/// <summary>
/// 订单Id

5
src/TrackCenter.Application/Events/TrackDeleteEvent.cs
View File

@ -1,10 +1,11 @@
using Fake.EventBus.Distributed;
using Fake.Messaging;
using TrackCenter.Application.Dtos.ProvideServices;
namespace TrackCenter.Application.Events;
[Serializable]
public class TrackDeleteEvent : IntegrationEvent
[MessageRoute(nameof(TrackDeleteEvent))]
public class TrackDeleteEvent : Message
{
/// <summary>
/// 根据esId定位轨迹，优先级最高

5
src/TrackCenter.Application/Events/TrackPushEvent.cs
View File

@ -1,4 +1,4 @@
using Fake.EventBus.Distributed;
using Fake.Messaging;
namespace TrackCenter.Application.Events;
@ -7,7 +7,8 @@ namespace TrackCenter.Application.Events;
/// 每次可传递多个订单，但仅能传递同一个轨迹内容
/// </summary>
[Serializable]
public class TrackPushEvent : IntegrationEvent
[MessageRoute(nameof(TrackPushEvent))]
public class TrackPushEvent : Message
{
/// <summary>
/// 操作类型

5
src/TrackCenter.Application/Events/TrackUpsertEvent.cs
View File

@ -1,10 +1,11 @@
using Fake.EventBus.Distributed;
using Fake.Messaging;
using TrackCenter.Application.Dtos.ProvideServices;
namespace TrackCenter.Application.Events;
[Serializable]
public class TrackUpsertEvent : IntegrationEvent
[MessageRoute(nameof(TrackUpsertEvent))]
public class TrackUpsertEvent : Message
{
/// <summary>
/// 应用

2
src/TrackCenter.Application/Events/TrackUpsertForRefreshEvent.cs
View File

@ -1,9 +1,11 @@
using Fake.Messaging;
namespace TrackCenter.Application.Events;
/// <summary>
/// 轨迹刷新引发的新增/更新（独立队列，与 <see cref="TrackUpsertEvent"/> 隔离）。
/// </summary>
[Serializable]
[MessageRoute(nameof(TrackUpsertForRefreshEvent))]
public class TrackUpsertForRefreshEvent : TrackUpsertEvent
{
}

6
src/TrackCenter.Application/Messaging/TrackChangeAfterUpdateOrderEventHandler.cs
View File

@ -1,6 +1,4 @@
using Fake.EventBus;
using Fake.EventBus.Distributed;
using Fake.UnitOfWork;
using Fake.Messaging;
using TrackCenter.Application.Events;
using TrackCenter.Application.Services.Warning.Handlers;
using TrackCenter.Domain.Aggregates.OrderAggregate;
@ -8,7 +6,7 @@ using TrackCenter.Domain.Aggregates.OrderAggregate;
namespace TrackCenter.Application.Messaging;
public class TrackChangeAfterUpdateOrderEventHandler(IOrderRepository orderRepository, IServiceProvider serviceProvider)
: IEventHandler<TrackChangeAfterUpdateOrderEvent>, ITransientDependency
: IMessageHandle<TrackChangeAfterUpdateOrderEvent>, ITransientDependency
{
public virtual async Task HandleAsync(
TrackChangeAfterUpdateOrderEvent @event,

17
src/TrackCenter.Application/Messaging/TrackDeleteEventHandler.cs
View File

@ -1,5 +1,4 @@
using Fake.EventBus;
using Fake.EventBus.Distributed;
using Fake.Messaging;
using Fake.UnitOfWork;
using InterfaceForward.Domain.Shared.Helpers;
using TrackCenter.Application.Dtos.ProvideServices;
@ -17,23 +16,23 @@ namespace TrackCenter.Application.Messaging;
public class TrackDeleteEventHandler(
ProvideQueries provideQueries,
IOrderRepository orderRepository,
IDistributedEventBus eventBus,
IMessagePublisher messagePublisher,
ITrackDetailsEsRepository trackDetailsEsRepository,
IOrderOperationRecordEsRepository orderOperationRecordEsRepository,
BroadcastingService broadcastingService,
WarningSignalPublisher warningSignalPublisher)
: IEventHandler<TrackDeleteEvent>, ITransientDependency
: IMessageHandle<TrackDeleteEvent>, ITransientDependency
{
[UnitOfWork(IsTransactional = true)]
public virtual async Task HandleAsync(TrackDeleteEvent @event, CancellationToken cancellationToken = default)
public virtual async Task HandleAsync(TrackDeleteEvent message, CancellationToken cancellationToken = default)
{
if (!CanDelete(@event)) return;
if (!CanDelete(message)) return;
// 处理删除轨迹
await HandleDeleteTrackAsync(@event);
await HandleDeleteTrackAsync(message);
// 根据订单关联关系异步向下刷（异步）
await DownByOrderLink(@event);
await DownByOrderLink(message);
}
private async Task DownByOrderLink(TrackDeleteEvent @event)
@ -77,7 +76,7 @@ public class TrackDeleteEventHandler(
var subEvent = nextEvent.DeepCopy();
subEvent.OrderIds = subOrderIds;
// 关联单继续走删除链路（原 CAP 误发到 Upsert topic，此处修正）
await eventBus.PublishAsync(subEvent);
await messagePublisher.PublishAsync(subEvent);
}
}
}

71
src/TrackCenter.Application/Rabbit/TrackPushCompletedEventHandler.cs → src/TrackCenter.Application/Messaging/TrackPushCompletedEventHandler.cs
View File

@ -1,63 +1,44 @@
using System.Text;
using Fake.Messaging;
using Fake.UnitOfWork;
using InterfaceForward.Domain.Shared.Helpers;
using Newtonsoft.Json;
using RabbitMQ.Client.Events;
using TrackCenter.Application.Dtos.TrackPush;
using TrackCenter.Application.Events;
using TrackCenter.Application.Helpers;
using TrackCenter.Application.Services.Broadcasting;
using TrackCenter.Application.Services.Warning.Handlers;
using TrackCenter.Domain.Aggregates.OrderAggregate;
using TrackCenter.Domain.Aggregates.TrackPushAggregate;
namespace TrackCenter.Application.Rabbit;
namespace TrackCenter.Application.Messaging;
[ExposeServices(typeof(IRabbitHandler))]
public class TrackPushCompletedEventHandler : ITransientDependency, IRabbitHandler
/// <summary>
/// 接口通轨迹推送完成回调 → 独立队列 <see cref="GlobalConst.TrackPushCompletedSubscribe"/>。
/// </summary>
[MessageSubscribe(GlobalConst.TrackPushCompletedSubscribe, ExternalQueue = true)]
public class TrackPushCompletedEventHandler(
IOrderRepository orderRepository,
ITrackPushRecordEsRepository trackPushRecordEsRepository,
BroadcastingService broadcastingService,
IReturnWarningHandler returnWarningHandler)
: IMessageHandle<InterfaceRelayForwardReplayEvent>, ITransientDependency
{
private readonly IOrderRepository _orderRepository;
private readonly ITrackPushRecordEsRepository _trackPushRecordEsRepository;
private readonly BroadcastingService _broadcastingService;
private readonly IReturnWarningHandler _returnWarningHandler;
public TrackPushCompletedEventHandler(IOrderRepository orderRepository,
ITrackPushRecordEsRepository trackPushRecordEsRepository,
BroadcastingService broadcastingService, IReturnWarningHandler returnWarningHandler)
public virtual async Task HandleAsync(
InterfaceRelayForwardReplayEvent message,
CancellationToken cancellationToken = default)
{
_orderRepository = orderRepository;
_trackPushRecordEsRepository = trackPushRecordEsRepository;
_broadcastingService = broadcastingService;
_returnWarningHandler = returnWarningHandler;
}
public bool Enable(ConsumeOptions options)
{
return options.Queue.StartsWith(GlobalConst.TrackPushCompletedSubscribe);
}
[UnitOfWork(IsTransactional = true)]
public virtual async Task Handle(BasicDeliverEventArgs args)
{
var msg = string.Empty;
try
{
msg = Encoding.UTF8.GetString(args.Body.ToArray());
var @event = System.Text.Json.JsonSerializer.Deserialize<InterfaceRelayForwardReplayEvent>(msg);
if (@event == null)
{
Log.Warning(
$"rabbit on queue received msg can't convert to {nameof(InterfaceRelayForwardReplayEvent)}");
return;
}
await ConsumerAsync(@event);
await ConsumerAsync(message);
}
catch (Exception ex)
{
Log.Error($"TrackPushCompletedEventHandler出错，错误信息:{ex}\r\n-----\r\n消息:{msg}");
Log.Error($"TrackPushCompletedEventHandler出错，错误信息:{ex}\r\n-----\r\n消息:{message.ToJson()}");
throw;
}
}
public async Task ConsumerAsync(InterfaceRelayForwardReplayEvent @event)
private async Task ConsumerAsync(InterfaceRelayForwardReplayEvent @event)
{
var optionType = Base64Helper.DeSerialize<string>(@event.Attach);
var pushInputs = new List<TrackPushInput>();
@ -91,16 +72,16 @@ public class TrackPushCompletedEventHandler : ITransientDependency, IRabbitHandl
// 推送广播
if (logs.First().IsToMq)
_broadcastingService.TrackPushCompletedPublicTopic(logs.Select(log => BuildDto(log, @event.Data)).ToList());
broadcastingService.TrackPushCompletedPublicTopic(logs.Select(log => BuildDto(log, @event.Data)).ToList());
// 写入推送日志
await _trackPushRecordEsRepository.AddManyAsync(logs);
await trackPushRecordEsRepository.AddManyAsync(logs);
// 同步更新订单主表
await TrackPushAfterUpdateOrderAsync(isAdd, @event.IsSuccess, logs);
// 引发预警计算
await _returnWarningHandler.ReceiveTrackPushRecordHandleAsync(@event.IsSuccess, @event.RequestParams, logs, pushInputs, updateInputs);
await returnWarningHandler.ReceiveTrackPushRecordHandleAsync(@event.IsSuccess, @event.RequestParams, logs, pushInputs, updateInputs);
}
private async Task TrackPushAfterUpdateOrderAsync(bool isAdd, bool isSuccess, List<TrackPushRecordIndex> logs)
@ -118,7 +99,7 @@ public class TrackPushCompletedEventHandler : ITransientDependency, IRabbitHandl
LatestPushError = x.Message
}).ToList();
await _orderRepository.UpdateRangeAsync(orders, x => new
await orderRepository.UpdateRangeAsync(orders, x => new
{
x.QuestionType,
x.LatestPushError
@ -136,7 +117,7 @@ public class TrackPushCompletedEventHandler : ITransientDependency, IRabbitHandl
LatestPushTime = x.CreateTime,
}).ToList();
await _orderRepository.UpdateRangeAsync(orders, x => new
await orderRepository.UpdateRangeAsync(orders, x => new
{
x.QuestionType,
x.LatestPushError,

5
src/TrackCenter.Application/Messaging/TrackPushEventHandler.cs
View File

@ -1,7 +1,6 @@
using System.Collections.Immutable;
using System.Text.Json;
using Fake.EventBus;
using Fake.EventBus.Distributed;
using Fake.Messaging;
using Fake.UnitOfWork;
using InterfaceForward.Domain.Shared.Helpers;
using Microsoft.Extensions.Options;
@ -29,7 +28,7 @@ public class TrackPushEventHandler(
BroadcastingService broadcastingService,
ITrackDetailsEsRepository trackDetailsEsRepository,
InterfaceForwardService interfaceForwardService)
: IEventHandler<TrackPushEvent>, ITransientDependency
: IMessageHandle<TrackPushEvent>, ITransientDependency
{
private List<OrderEntity> _orders = null!;
private List<TrackPushContentConfigDto> _pushConfigs = null!;

54
src/TrackCenter.Application/Rabbit/TrackRefreshEventHandler.cs → src/TrackCenter.Application/Messaging/TrackRefreshEventHandler.cs
View File

@ -1,13 +1,11 @@
using System.Diagnostics;
using System.Text;
using Fake.EventBus.Distributed;
using Fake.Messaging;
using Fake.UnitOfWork;
using InterfaceForward.Domain.Shared.Helpers;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using Newtonsoft.Json;
using Newtonsoft.Json.Linq;
using RabbitMQ.Client.Events;
using TrackCenter.Application.Dtos.TrackPull;
using TrackCenter.Application.Events;
using TrackCenter.Application.Helpers;
@ -16,47 +14,37 @@ using TrackCenter.Application.Services.TrackPull;
using TrackCenter.Domain.Aggregates.OrderAggregate;
using TrackCenter.Domain.Aggregates.TrackDetailsAggregate;
namespace TrackCenter.Application.Rabbit;
namespace TrackCenter.Application.Messaging;
[ExposeServices(typeof(IRabbitHandler))]
/// <summary>
/// 接口通轨迹刷新回调 → 独立队列 <see cref="GlobalConst.TrackRefreshSubscribe"/>。
/// </summary>
[MessageSubscribe(GlobalConst.TrackRefreshSubscribe, ExternalQueue = true)]
public class TrackRefreshEventHandler(
IOrderRepository orderRepository,
IDistributedEventBus eventBus,
IMessagePublisher messagePublisher,
TrackPullCacheService trackPullCacheService,
ITrackDetailsEsRepository trackDetailsEsRepository,
ILogger<TrackRefreshEventHandler> logger,
IOptionsMonitor<InterfaceRelayParamsConfigOptions> options)
: ITransientDependency, IRabbitHandler
: IMessageHandle<InterfaceRelayForwardReplayEvent>, ITransientDependency
{
public bool Enable(ConsumeOptions coptions)
public virtual async Task HandleAsync(
InterfaceRelayForwardReplayEvent message,
CancellationToken cancellationToken = default)
{
return coptions.Queue.StartsWith(GlobalConst.TrackRefreshSubscribe);
}
[UnitOfWork(IsTransactional = true)]
public virtual async Task Handle(BasicDeliverEventArgs args)
{
var msg = string.Empty;
try
{
msg = Encoding.UTF8.GetString(args.Body.ToArray());
var @event = System.Text.Json.JsonSerializer.Deserialize<InterfaceRelayForwardReplayEvent>(msg);
if (@event == null)
{
logger.LogWarning(
$"rabbit on queue received msg can't convert to {nameof(InterfaceRelayForwardReplayEvent)}");
return;
}
await ConsumerAsync(@event);
await ConsumerAsync(message);
}
catch (Exception ex)
{
Log.Error($"TrackRefreshEventHandler出错，错误信息:{ex}\r\n-----\r\n消息:{msg}");
Log.Error($"TrackRefreshEventHandler出错，错误信息:{ex}\r\n-----\r\n消息:{message.ToJson()}");
throw;
}
}
public async Task ConsumerAsync(InterfaceRelayForwardReplayEvent @event)
private async Task ConsumerAsync(InterfaceRelayForwardReplayEvent @event)
{
if (!@event.IsSuccess)
{
@ -71,15 +59,6 @@ public class TrackRefreshEventHandler(
return;
}
// 最优先级，最匹配的 权重 sort
// 客户、渠道、产品
// */*/* 全部
// 客户A/渠道A/产品A
// 客户A
// 客户A 渠道B 产品A
var config = await trackPullCacheService.GetPullParamsWithContentAsync(attach.AppKey,
attach.ServiceProviderCode, attach.OrderType);
if (config is null)
@ -134,7 +113,6 @@ public class TrackRefreshEventHandler(
ServiceProviderCode = attach.ServiceProviderCode
};
if (trackToken is JArray trackTokens)
{
foreach (var item in trackTokens)
@ -284,7 +262,7 @@ public class TrackRefreshEventHandler(
Extra = trackExtra
};
await eventBus.PublishAsync(trackUpsertEvent);
await messagePublisher.PublishAsync(trackUpsertEvent);
}
private async Task InsertNotMappedTrackAsync(JToken trackToken, TrackRefreshContext context, string? trackPlace,

33
src/TrackCenter.Application/Messaging/TrackUpsertEventHandler.cs
View File

@ -1,5 +1,4 @@
using Fake.EventBus;
using Fake.EventBus.Distributed;
using Fake.Messaging;
using Fake.UnitOfWork;
using InterfaceForward.Domain.Shared.Helpers;
using TrackCenter.Application.Events;
@ -18,30 +17,22 @@ namespace TrackCenter.Application.Messaging;
public class TrackUpsertEventHandler(
IOrderRepository orderRepository,
IDistributedEventBus eventBus,
IMessagePublisher messagePublisher,
ProvideQueries provideQueries,
ITrackDetailsEsRepository trackDetailsEsRepository,
IOrderOperationRecordEsRepository orderOperationRecordEsRepository,
BroadcastingService broadcastingService,
WarningSignalPublisher warningSignalPublisher)
:
IEventHandler<TrackUpsertEvent>,
IEventHandler<TrackUpsertForRefreshEvent>,
IMessageHandle<TrackUpsertEvent>,
IMessageHandle<TrackUpsertForRefreshEvent>,
ITransientDependency
{
Task IEventHandler.HandleAsync(Event eventData, CancellationToken cancellationToken)
=> eventData switch
{
TrackUpsertForRefreshEvent refresh => HandleAsync(refresh, cancellationToken),
TrackUpsertEvent upsert => HandleAsync(upsert, cancellationToken),
_ => Task.CompletedTask
};
public Task HandleAsync(TrackUpsertEvent message, CancellationToken cancellationToken = default)
=> ConsumerAsync(message);
public Task HandleAsync(TrackUpsertEvent @event, CancellationToken cancellationToken = default)
=> ConsumerAsync(@event);
public Task HandleAsync(TrackUpsertForRefreshEvent @event, CancellationToken cancellationToken = default)
=> ConsumerAsync(@event);
public Task HandleAsync(TrackUpsertForRefreshEvent message, CancellationToken cancellationToken = default)
=> ConsumerAsync(message);
private async Task ConsumerAsync(TrackUpsertEvent @event)
{
@ -107,7 +98,7 @@ public class TrackUpsertEventHandler(
{
var subEvent = nextEvent.DeepCopy();
subEvent.OrderIdentifiers = subOrderIds.ToArray();
await eventBus.PublishAsync(subEvent);
await messagePublisher.PublishAsync(subEvent);
}
}
}
@ -200,7 +191,7 @@ public class TrackUpsertEventHandler(
// 如果存在轨迹标记，则触发订单pod拉取
if (!string.IsNullOrWhiteSpace(trackContent.TrackFlagId))
{
await eventBus.PublishAsync(new OrderPodPullEvent
await messagePublisher.PublishAsync(new OrderPodPullEvent
{
AppKey = @event.AppKey,
OrderIdEsIdDict = insertTracks.ToDictionary(x => x.OrderId, y => y.Id),
@ -213,7 +204,7 @@ public class TrackUpsertEventHandler(
await TrackChangePublicTopicAsync(trackContent, trackChangeEtos, true);
// 触发订单主表更新事件（异步）
await eventBus.PublishAsync(new TrackChangeAfterUpdateOrderEvent
await messagePublisher.PublishAsync(new TrackChangeAfterUpdateOrderEvent
{
OrderIds = canNextOrderIds.ToArray(),
IsCompleted = trackContent.IsCompleted,
@ -248,7 +239,7 @@ public class TrackUpsertEventHandler(
ForcedPush = @event.ForcedPush,
IsToMq = (trackContent.IsToMq || @event.IsToMq)
};
await eventBus.PublishAsync(trackPushEvent);
await messagePublisher.PublishAsync(trackPushEvent);
}
private async Task<List<TrackDetailsIndex>> UpsertTrackAsync(

10
src/TrackCenter.Application/Rabbit/IRabbitHandler.cs
View File

@ -1,10 +0,0 @@
using RabbitMQ.Client.Events;
namespace TrackCenter.Application.Rabbit;
public interface IRabbitHandler
{
bool Enable(ConsumeOptions options);
Task Handle(BasicDeliverEventArgs args);
}

166
src/TrackCenter.Application/Rabbit/RabbitClient.cs
View File

@ -1,166 +0,0 @@
using Fake.RabbitMQ;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
namespace TrackCenter.Application.Rabbit;
public class RabbitClient(
IRabbitMqChannelPool channelPool,
IServiceScopeFactory serviceScopeFactory,
ILogger<RabbitClient> logger,
IOptions<RabbitOptions> options) : ISingletonDependency, IHostedService
{
public void Subscribe(ConsumeOptions consumeOptions)
{
using var channelAccessor = channelPool.Acquire(consumeOptions.Queue);
if (!channelAccessor.IsNew) return;
var channel = channelAccessor.Channel;
var decl = new DefaultDeclaration(channel);
consumeOptions.Declaration.Invoke(decl);
if (consumeOptions.FetchCount > 0)
{
channel.BasicQos(0, consumeOptions.FetchCount, false);
}
var consumer = new AsyncEventingBasicConsumer(channel);
consumer.Received += (model, ea) =>
{
_ = Task.Run(async () =>
{
try
{
await using var scope = serviceScopeFactory.CreateAsyncScope();
foreach (var handler in scope.ServiceProvider.GetServices<IRabbitHandler>())
{
if (handler.Enable(consumeOptions))
{
await handler.Handle(ea);
}
}
channel.BasicAck(ea.DeliveryTag, false);
}
catch (Exception ex)
{
logger.LogError($"rabbit on queue({consumeOptions.Queue}) received error: {ex}");
channel.BasicNack(ea.DeliveryTag, false, consumeOptions.FailedRequeue);
}
});
return Task.CompletedTask;
};
channel.CallbackException += (sender, args) =>
{
logger.LogError(args.Exception, "RabbitMQ channel exception");
};
channel.BasicConsume(queue: consumeOptions.Queue,
autoAck: false,
consumer: consumer);
}
public void Publish(string exchange, string routingKey, byte[] data, Action<IBasicProperties>? options = null)
{
using var channelAccessor = channelPool.Acquire(routingKey);
var props = channelAccessor.Channel.CreateBasicProperties();
options?.Invoke(props);
channelAccessor.Channel.BasicPublish(exchange, routingKey, props, data);
}
public void PublishByConfirmSelect(
string exchange,
string routingKey,
byte[] data,
Action<IBasicProperties>? options = null
)
{
using var activity = RabbitInstrumentation.ActivitySource.StartActivity("/" + exchange + "/" + routingKey);
activity?.SetTag("kind", "publish");
using var channelAccessor = channelPool.Acquire(routingKey);
var basicProperties = channelAccessor.Channel.CreateBasicProperties();
// 配置属性
if (options != null)
options(basicProperties);
var model = channelAccessor.Channel;
// 启用发布确认
model.ConfirmSelect();
var readOnlyMemory = (ReadOnlyMemory<byte>) data;
model.BasicPublish(exchange, routingKey, basicProperties, readOnlyMemory);
model.WaitForConfirmsOrDie();
}
public void PublishByConfirmSelect(
string exchange,
string routingKey,
IEnumerable<byte[]> datas,
Action<IBasicProperties>? options = null
)
{
using var activity = RabbitInstrumentation.ActivitySource.StartActivity("/" + exchange + "/" + routingKey);
activity?.SetTag("kind", "publish");
using var channelAccessor = channelPool.Acquire(routingKey);
var basicProperties = channelAccessor.Channel.CreateBasicProperties();
// 配置属性
if (options != null)
options(basicProperties);
var model = channelAccessor.Channel;
// 启用发布确认
model.ConfirmSelect();
foreach (var data in datas)
{
var readOnlyMemory = (ReadOnlyMemory<byte>) data;
model.BasicPublish(exchange, routingKey, basicProperties, readOnlyMemory);
}
model.WaitForConfirmsOrDie();
}
public void UnSubscribe(string queueName, string exchange, string routingKey, bool deleteQueue = false)
{
logger.LogInformation(
$"UnSubscribe queue: {queueName} exchange: {exchange} routingKey: {routingKey} deleteQueue: {deleteQueue}");
using var channelAccessor = channelPool.Acquire(queueName);
if (deleteQueue)
{
var passive = channelAccessor.Channel.QueueDeclarePassive(queueName);
logger.LogInformation($"Queue {queueName} exists, deleting... {passive.MessageCount} messages in queue");
channelAccessor.Channel.QueueDelete(queueName);
}
channelPool.Release(queueName);
}
public Task StartAsync(CancellationToken cancellationToken)
{
if (options.Value.Declarations.Count > 0)
{
using var channelAccessor = channelPool.Acquire();
var channel = channelAccessor.Channel;
var decl = new DefaultDeclaration(channel);
options.Value.Declarations.ForEach(q => q.Invoke(decl));
}
foreach (var opt in options.Value.Consumes)
{
Subscribe(opt);
}
return Task.CompletedTask;
}
public Task StopAsync(CancellationToken cancellationToken)
{
channelPool.Dispose();
return Task.CompletedTask;
}
}

21
src/TrackCenter.Application/Rabbit/RabbitInstrumentation.cs
View File

@ -1,21 +0,0 @@
using System.Diagnostics;
using System.Diagnostics.Metrics;
namespace TrackCenter.Application.Rabbit;
public class RabbitInstrumentation
{
public const string MeterName = "Fake.Rabbit.Meter";
public const string ActivitySourceName = "Fake.Rabbit.ActivitySource";
static RabbitInstrumentation()
{
string version = typeof (RabbitInstrumentation).Assembly.GetName().Version?.ToString();
Meter = new Meter(MeterName, version);
ActivitySource = new ActivitySource(ActivitySourceName, version);
}
public static Meter Meter { get; }
public static ActivitySource ActivitySource { get; }
}

84
src/TrackCenter.Application/Rabbit/RabbitOptions.cs
View File

@ -1,84 +0,0 @@
using RabbitMQ.Client;
namespace TrackCenter.Application.Rabbit;
public class RabbitOptions
{
public List<ConsumeOptions> Consumes { get; set; } = new();
public List<Action<IDeclaration>> Declarations { get; set; } = new();
}
public class ConsumeOptions
{
/// <summary>
/// 队列名
/// </summary>
public string Queue { get; set; }
/// <summary>
/// 并行度
/// </summary>
public ushort FetchCount { get; set; } = 1;
/// <summary>
/// handle发生异常，是否重回队列
/// </summary>
public bool FailedRequeue { get; set; }
public Action<IDeclaration> Declaration { get; set; }
}
public interface IDeclaration
{
void QueueDeclare(string queue, bool durable, bool autoDelete = false, bool exclusive = false,
IDictionary<string, object> arguments = null);
void ExchangeDeclare(string exchange, string type, bool durable, bool autoDelete = false,
IDictionary<string, object> arguments = null);
void QueueBind(string queue, string exchange, string routingKey,
IDictionary<string, object> arguments = null);
}
public class DefaultDeclaration : IDeclaration
{
private readonly IModel _chnl;
// ReSharper disable once ConvertToPrimaryConstructor
public DefaultDeclaration(IModel chnl)
{
_chnl = chnl;
}
public void QueueDeclare(string queue, bool durable, bool autoDelete = false, bool exclusive = false,
IDictionary<string, object> arguments = null)
{
_chnl.QueueDeclare(
queue: queue,
durable: durable,
autoDelete: autoDelete,
exclusive: exclusive,
arguments: arguments);
}
public void ExchangeDeclare(string exchange, string type, bool durable, bool autoDelete = false,
IDictionary<string, object> arguments = null)
{
_chnl.ExchangeDeclare(
exchange: exchange,
type: type,
durable: durable,
autoDelete: autoDelete,
arguments: arguments);
}
public void QueueBind(string queue, string exchange, string routingKey, IDictionary<string, object> arguments)
{
_chnl.QueueBind(
queue: queue,
exchange: exchange,
routingKey: routingKey,
arguments: arguments);
}
}

76
src/TrackCenter.Application/Services/Broadcasting/BroadcastingService.cs
View File

@ -1,74 +1,102 @@
using System.Text;
using Fake.RabbitMQ;
using InterfaceForward.Domain.Shared.Helpers;
using RabbitMQ.Client;
using TrackCenter.Application.Dtos.TrackPush;
using TrackCenter.Application.Events;
using TrackCenter.Application.Rabbit;
namespace TrackCenter.Application.Services.Broadcasting;
/// <summary>
/// 广播服务
/// 对外 Topic 广播（下游仍按 Newtonsoft PascalCase 契约消费，不经 Messaging 信封）。
/// </summary>
public class BroadcastingService : ITransientDependency
public class BroadcastingService(IRabbitMqChannelPool channelPool) : ITransientDependency
{
private readonly RabbitClient _rabbitClient;
public BroadcastingService(RabbitClient rabbitClient)
public void TrackChangePublicTopic(TrackChangeEto eto, string AppCode)
{
_rabbitClient = rabbitClient;
}
public void TrackChangePublicTopic(TrackChangeEto eto, string AppCode)
{
_rabbitClient.PublishByConfirmSelect(
PublishByConfirmSelect(
GlobalConst.TrackChangeExchange,
ExchangeType.Topic,
$"TrackChange.{AppCode}",
Encoding.UTF8.GetBytes(eto.ToJson()),
prop => { prop.Persistent = true; }
);
prop => { prop.Persistent = true; });
}
/// <summary>
/// 轨迹推送完成广播（批量）
/// </summary>
/// <param name="list"></param>
/// <returns></returns>
public void TrackPushCompletedPublicTopic(List<TrackPushRecordDto> list)
{
var data = list
.Select(TrackPushRecordDtoToEvent)
.Select(x => Encoding.UTF8.GetBytes(x.ToJson()));
var log = list.First();
_rabbitClient.PublishByConfirmSelect(
PublishByConfirmSelect(
GlobalConst.TrackPushExchange,
ExchangeType.Topic,
$"TrackPush.{log.AppKey}",
data,
prop => { prop.Persistent = true; }
);
prop => { prop.Persistent = true; });
}
/// <summary>
/// 轨迹推送完成广播
/// </summary>
/// <param name="log"></param>
/// <returns></returns>
public void TrackPushCompletedPublicTopic(TrackPushRecordDto log)
{
var data = TrackPushRecordDtoToEvent(log);
_rabbitClient.PublishByConfirmSelect(GlobalConst.TrackPushExchange, $"TrackPush.{log.AppKey}",
Encoding.UTF8.GetBytes(data.ToJson()), prop => { prop.Persistent = true; });
PublishByConfirmSelect(
GlobalConst.TrackPushExchange,
ExchangeType.Topic,
$"TrackPush.{log.AppKey}",
Encoding.UTF8.GetBytes(data.ToJson()),
prop => { prop.Persistent = true; });
}
/// <summary>
/// 轨迹推送失败广播
/// </summary>
/// <param name="log"></param>
public void TrackPushFailedPubilcTopic(TrackPushRecordDto log)
{
//暂时使用推送完成模型，后续可能会出现有区分度的模型
TrackPushCompletedPublicTopic(log);
}
private void PublishByConfirmSelect(
string exchange,
string exchangeType,
string routingKey,
byte[] data,
Action<IBasicProperties>? options = null)
{
using var channelAccessor = channelPool.Acquire(routingKey);
var channel = channelAccessor.Channel;
channel.ExchangeDeclare(exchange, exchangeType, durable: true);
var props = channel.CreateBasicProperties();
options?.Invoke(props);
channel.ConfirmSelect();
channel.BasicPublish(exchange, routingKey, props, data);
channel.WaitForConfirmsOrDie();
}
private void PublishByConfirmSelect(
string exchange,
string exchangeType,
string routingKey,
IEnumerable<byte[]> datas,
Action<IBasicProperties>? options = null)
{
using var channelAccessor = channelPool.Acquire(routingKey);
var channel = channelAccessor.Channel;
channel.ExchangeDeclare(exchange, exchangeType, durable: true);
var props = channel.CreateBasicProperties();
options?.Invoke(props);
channel.ConfirmSelect();
foreach (var data in datas)
channel.BasicPublish(exchange, routingKey, props, data);
channel.WaitForConfirmsOrDie();
}
private static TrackPushCompletedEvent TrackPushRecordDtoToEvent(TrackPushRecordDto log)
{
return new TrackPushCompletedEvent

6
src/TrackCenter.Application/Services/ProvideServices/OrderService.Link.cs
View File

@ -1,5 +1,5 @@
using System.Data;
using Fake.EventBus.Distributed;
using Fake.Messaging;
using Fake.UnitOfWork;
using TrackCenter.Application.Dtos.ProvideServices;
using TrackCenter.Application.Events;
@ -13,7 +13,7 @@ namespace TrackCenter.Application.Services.ProvideServices;
/// </summary>
public partial class OrderService
{
private readonly IDistributedEventBus _eventBus = eventBus;
private readonly IMessagePublisher _messagePublisher = messagePublisher;
private readonly ITrackDetailsEsRepository _trackDetailsEsRepository = trackDetailsEsRepository;
/// <summary>
@ -130,7 +130,7 @@ public partial class OrderService
IsAdd = true // 存在则不新增
};
await _eventBus.PublishAsync(fillTrackEvent);
await _messagePublisher.PublishAsync(fillTrackEvent);
}
}

4
src/TrackCenter.Application/Services/ProvideServices/OrderService.cs
View File

@ -1,4 +1,4 @@
using Fake.EventBus.Distributed;
using Fake.Messaging;
using TrackCenter.Application.Dtos.ProvideServices;
using TrackCenter.Domain.Aggregates.OrderAggregate;
using TrackCenter.Domain.Aggregates.TrackDetailsAggregate;
@ -17,7 +17,7 @@ public partial class OrderService(
IOrderRepository orderRepository,
IOrderOperationRecordEsRepository orderOperationRecordEsRepository,
ITrackPullConfigRepository trackPullConfigRepository,
IDistributedEventBus eventBus,
IMessagePublisher messagePublisher,
ITrackDetailsEsRepository trackDetailsEsRepository,
IServiceProvider serviceProvider)
: ApplicationServiceBase

10
src/TrackCenter.Application/Services/ProvideServices/TrackPushService.cs
View File

@ -1,5 +1,5 @@
using System.Data;
using Fake.EventBus.Distributed;
using Fake.Messaging;
using Fake.UnitOfWork;
using TrackCenter.Application.Dtos.ProvideServices;
using TrackCenter.Application.Dtos.TrackOrder;
@ -20,20 +20,20 @@ namespace TrackCenter.Application.Services.ProvideServices;
[Route("/ProviderService")]
public class TrackPushService : ApplicationServiceBase
{
private readonly IDistributedEventBus _eventBus;
private readonly IMessagePublisher _messagePublisher;
private readonly IOrderRepository _orderRepository;
private readonly ITrackPushConfigRepository _trackPushConfigRepository;
private readonly ITrackPushRecordEsRepository _trackPushRecordEsRepository;
private readonly ITrackDetailsEsRepository _trackDetailsEsRepository;
public TrackPushService(IOrderRepository orderRepository
, IDistributedEventBus eventBus
, IMessagePublisher messagePublisher
, ITrackPushRecordEsRepository trackPushRecordEsRepository
, ITrackPushConfigRepository trackPushConfigRepository,
ITrackDetailsEsRepository trackDetailsEsRepository)
{
_orderRepository = orderRepository;
_eventBus = eventBus;
_messagePublisher = messagePublisher;
_trackPushRecordEsRepository = trackPushRecordEsRepository;
_trackPushConfigRepository = trackPushConfigRepository;
_trackDetailsEsRepository = trackDetailsEsRepository;
@ -155,7 +155,7 @@ public class TrackPushService : ApplicationServiceBase
}
};
await _eventBus.PublishAsync(trackPushEvent);
await _messagePublisher.PublishAsync(trackPushEvent);
pushedOrderNos.Add(order.OrderNo);
}

12
src/TrackCenter.Application/Services/ProvideServices/TrackService.cs
View File

@ -1,5 +1,5 @@
using System.Data;
using Fake.EventBus.Distributed;
using Fake.Messaging;
using Fake.UnitOfWork;
using InterfaceForward.Domain.Shared.Helpers;
using TrackCenter.Application.Dtos.ProvideServices;
@ -19,7 +19,7 @@ namespace TrackCenter.Application.Services.ProvideServices;
[Route("/ProviderService")]
public partial class TrackService(
IOrderRepository orderRepository,
IDistributedEventBus eventBus,
IMessagePublisher messagePublisher,
ITrackDetailsEsRepository trackDetailsEsRepository,
ITrackContentRepository trackContentRepository)
: ApplicationServiceBase
@ -90,7 +90,7 @@ public partial class TrackService(
}
// 删除轨迹(异步)
await eventBus.PublishAsync(new TrackDeleteEvent
await messagePublisher.PublishAsync(new TrackDeleteEvent
{
EsIds = esIds,
OrderIds = new[] { orderId },
@ -165,7 +165,7 @@ public partial class TrackService(
AppKey = input.AppKey,
OperationMan = CurrentUser.UserName
};
await eventBus.PublishAsync(@event);
await messagePublisher.PublishAsync(@event);
}
/// <summary>
@ -219,7 +219,7 @@ public partial class TrackService(
}
};
@event.OperationMan = CurrentUser.UserName;
await eventBus.PublishAsync(@event);
await messagePublisher.PublishAsync(@event);
}
/// <summary>
@ -292,7 +292,7 @@ public partial class TrackService(
ForcedPush = input.ForcedPush,
IsToMq = input.IsToMq,
};
await eventBus.PublishAsync(@event);
await messagePublisher.PublishAsync(@event);
}
}

10
src/TrackCenter.Application/Services/Warning/ReturnWarningService.cs
View File

@ -1,5 +1,5 @@
using System.Data;
using Fake.EventBus.Distributed;
using Fake.Messaging;
using Fake.Domain.Dtos;
using Fake.UnitOfWork;
using InterfaceForward.Domain.Shared.Helpers;
@ -27,7 +27,7 @@ public class ReturnWarningService : ApplicationServiceBase
private readonly IReturnWarningHandler _returnWarningHandler;
private readonly IWarningRuleRepository _warningRuleRepository;
private readonly IWarningDetailsEsRepository _warningDetailsEsRepository;
private readonly IDistributedEventBus _eventBus;
private readonly IMessagePublisher _messagePublisher;
private readonly TrackStatusNameDictCache _trackStatusNameDictCache;
private readonly AppNameDictCache _appNameDictCache;
@ -36,7 +36,7 @@ public class ReturnWarningService : ApplicationServiceBase
IReturnWarningHandler returnWarningHandler,
IWarningRuleRepository warningRuleRepository,
IWarningDetailsEsRepository warningDetailsEsRepository,
IDistributedEventBus eventBus,
IMessagePublisher messagePublisher,
TrackStatusNameDictCache trackStatusNameDictCache,
AppNameDictCache appNameDictCache
)
@ -45,7 +45,7 @@ public class ReturnWarningService : ApplicationServiceBase
_returnWarningHandler = returnWarningHandler;
_warningRuleRepository = warningRuleRepository;
_warningDetailsEsRepository = warningDetailsEsRepository;
_eventBus = eventBus;
_messagePublisher = messagePublisher;
_trackStatusNameDictCache = trackStatusNameDictCache;
_appNameDictCache = appNameDictCache;
}
@ -279,7 +279,7 @@ public class ReturnWarningService : ApplicationServiceBase
}
};
await _eventBus.PublishAsync(trackPushEvent);
await _messagePublisher.PublishAsync(trackPushEvent);
}
return $"即将进行{filtered.Count}个回传推送！";

3
src/TrackCenter.Application/TrackCenter.Application.csproj
View File

@ -9,8 +9,7 @@
<ItemGroup>
<PackageReference Include="Fake.AspNetCore" Version="$(FakePackageVersion)" />
<PackageReference Include="Fake.EventBus.RabbitMQ" Version="$(FakePackageVersion)" />
<PackageReference Include="Fake.EventBus.IntegrationEventLog" Version="$(FakePackageVersion)" />
<PackageReference Include="Fake.Messaging.RabbitMQ" Version="$(FakePackageVersion)" />
<PackageReference Include="Fake.ObjectMapping.AutoMapper" Version="$(FakePackageVersion)" />
<PackageReference Include="Fake.ObjectStorage.TencentCos" Version="$(FakePackageVersion)" />

67
src/TrackCenter.Application/TrackCenterApplicationModule.cs
View File

@ -1,28 +1,26 @@
using System.Diagnostics;
using System.Net;
using Fake;
using Fake.EventBus.RabbitMQ;
using Fake.Messaging;
using Fake.Messaging.RabbitMQ;
using Fake.Modularity;
using Fake.ObjectMapping.AutoMapper;
using Fake.ObjectStorage;
using Fake.RabbitMQ;
using FreeRedis;
using Hangfire;
using Hangfire.Redis.StackExchange;
using InterfaceForward.Domain.Shared.Helpers;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.Logging;
using RabbitMQ.Client;
using TrackCenter.Application.Kafka;
using TrackCenter.Application.HostServices;
using TrackCenter.Application.Kafka;
using TrackCenter.Application.Messaging;
using TrackCenter.Application.Options;
using TrackCenter.Application.Rabbit;
namespace TrackCenter.Application;
[DependsOn(typeof(FakeRabbitMqModule))]
[DependsOn(typeof(FakeObjectMappingAutoMapperModule))]
[DependsOn(typeof(FakeEventBusRabbitMqModule))]
[DependsOn(typeof(FakeMessagingRabbitMqModule))]
[DependsOn(typeof(FakeObjectStorageTencentCosModule))]
[DependsOn(typeof(TrackCenterDomainModule))]
public class TrackCenterApplicationModule : FakeModule
@ -36,50 +34,27 @@ public class TrackCenterApplicationModule : FakeModule
options.ScanProfiles<TrackCenterApplicationModule>();
});
// Messaging：独立队列；PrefetchCount=0 → ProcessorCount；接口通可按 TrackApp:TrackRefreshFetchCount 覆盖
context.Services.Configure<MessagingOptions>(configuration.GetSection(MessagingOptions.SectionName));
context.Services.Configure<RabbitMessageOptions>(o =>
{
configuration.GetSection(RabbitMessageOptions.SectionName).Bind(o);
var refreshFetch = configuration.GetValue<ushort>("TrackApp:TrackRefreshFetchCount");
if (refreshFetch > 0)
{
o.Handlers[nameof(TrackRefreshEventHandler)] =
new RabbitMessageHandlerOptions { PrefetchCount = refreshFetch };
o.Handlers[nameof(TrackPushCompletedEventHandler)] =
new RabbitMessageHandlerOptions { PrefetchCount = refreshFetch };
}
});
var redisOptions = configuration.GetSection("Redis").Get<ConnectionStringBuilder>();
if (redisOptions == null)
throw new FakeException("未找到Redis相关配置，请检查配置文件中是否包含名为Redis的节点");
RedisHelper.Init(redisOptions, showLogInfo: Debugger.IsAttached);
context.Services.AddHostedService<RabbitClient>();
context.Services.Configure<RabbitOptions>(options =>
{
options.Declarations = new List<Action<IDeclaration>>()
{
o => o.ExchangeDeclare(GlobalConst.TrackChangeExchange, ExchangeType.Topic, true),
o => o.ExchangeDeclare(GlobalConst.TrackPushExchange, ExchangeType.Topic, true)
};
options.Consumes = new List<ConsumeOptions>
{
new()
{
Queue = GlobalConst.TrackRefreshSubscribe,
FetchCount = configuration.GetValue<ushort>("TrackApp:TrackRefreshFetchCount"),
FailedRequeue = false,
Declaration = o =>
{
o.ExchangeDeclare(GlobalConst.InterfaceRelayExchange, ExchangeType.Direct, true);
o.QueueDeclare(GlobalConst.TrackRefreshSubscribe, true);
o.QueueBind(GlobalConst.TrackRefreshSubscribe, GlobalConst.InterfaceRelayExchange,
GlobalConst.TrackRefreshSubscribe);
}
},
new()
{
Queue = GlobalConst.TrackPushCompletedSubscribe,
FetchCount = configuration.GetValue<ushort>("TrackApp:TrackRefreshFetchCount"),
FailedRequeue = false,
Declaration = o =>
{
o.ExchangeDeclare(GlobalConst.InterfaceRelayExchange, ExchangeType.Direct, true);
o.QueueDeclare(GlobalConst.TrackPushCompletedSubscribe, true);
o.QueueBind(GlobalConst.TrackPushCompletedSubscribe, GlobalConst.InterfaceRelayExchange,
GlobalConst.TrackPushCompletedSubscribe);
}
}
};
});
context.Services.Configure<TrackAppOptions>(configuration.GetSection("TrackApp"));
context.Services.Configure<InterfaceRelayParamsConfigOptions>(
configuration.GetSection("InterfaceRelayParamsConfig"));