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

Merge pull request '1.03' (#4) from mq into test
Some checks failed
Build and Deploy / deploy (push) Has been cancelled
Details

Browse Source 
Reviewed-on: #4
This commit is contained in:
xiaolipro 2026-10-10 16:46:05 +08:00
commit 957ea62f70
24 changed files with 584 additions and 147 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
Directory.Build.props
View File

@ -1,6 +1,6 @@
<Project>
<PropertyGroup>
<!-- All Fake packages -->
<FakePackageVersion>8.0.5.18</FakePackageVersion>
<FakePackageVersion>8.0.5.19</FakePackageVersion>
</PropertyGroup>
</Project>

68
src/Oms.Application/Messaging/OmsIntegrationEventPublisher.cs Normal file
View File

@ -0,0 +1,68 @@
using Fake.EntityFrameworkCore;
using Fake.EventBus.Distributed;
using Fake.EventBus.IntegrationEventLog;
using Fake.UnitOfWork;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
namespace Oms.Application.Messaging;
/// <summary>
/// 可靠发布 IntegrationEvent：事务 UoW 内写入 Outbox（由 Fake.EventBus.IntegrationEventLog 投递）。
/// </summary>
internal static class OmsIntegrationEventPublisher
{
public static async Task PublishAsync(
IDistributedEventBus eventBus,
IUnitOfWorkManager unitOfWorkManager,
ILogger logger,
IntegrationEvent @event,
CancellationToken cancellationToken = default)
{
if (OmsOutboxUnitOfWork.IsAmbientTransactional(unitOfWorkManager))
{
await eventBus.PublishAsync(@event, cancellationToken);
logger.LogDebug(
"Outbox enqueued {EventName} Id={EventId} (ambient TX)",
@event.GetType().Name,
@event.Id);
return;
}
// 无环境事务时开启短事务只写 Outbox，并告警：业务落库与事件非原子。
logger.LogWarning(
"Publish {EventName} Id={EventId} without ambient transactional UoW; opening short outbox TX",
@event.GetType().Name,
@event.Id);
using var uow = unitOfWorkManager.Begin(OmsOutboxUnitOfWork.Transactional());
// 先打开 Outbox DbContext 以产生可征用的 DbTransaction
var dbProvider = uow.ServiceProvider.GetRequiredService<IEfDbContextProvider<IntegrationEventLogContext>>();
await dbProvider.GetDbContextAsync();
await eventBus.PublishAsync(@event, cancellationToken);
await uow.CompleteAsync(cancellationToken);
}
public static async Task<bool> TryPublishAsync(
IDistributedEventBus eventBus,
IUnitOfWorkManager unitOfWorkManager,
ILogger logger,
IntegrationEvent @event,
CancellationToken cancellationToken = default)
{
try
{
await PublishAsync(eventBus, unitOfWorkManager, logger, @event, cancellationToken);
return true;
}
catch (Exception ex)
{
logger.LogError(
ex,
"Outbox publish failed {EventName} Id={EventId}",
@event.GetType().Name,
@event.Id);
return false;
}
}
}

22
src/Oms.Application/Messaging/OmsOutboxUnitOfWork.cs Normal file
View File

@ -0,0 +1,22 @@
using System.Data;
using Fake.UnitOfWork;
namespace Oms.Application.Messaging;
/// <summary>
/// 集成事件 Outbox 所需的事务性工作单元约定。
/// </summary>
public static class OmsOutboxUnitOfWork
{
/// <summary>
/// 创单/后置事件发布使用的事务 UoW 属性（MySQL ReadCommitted）。
/// </summary>
public static UnitOfWorkAttribute Transactional() => new()
{
IsTransactional = true,
IsolationLevel = IsolationLevel.ReadCommitted
};
public static bool IsAmbientTransactional(IUnitOfWorkManager unitOfWorkManager) =>
unitOfWorkManager.Current?.Context.IsTransactional == true;
}

3
src/Oms.Application/Messaging/OrderBatchJobEventHandler.cs
View File

@ -29,7 +29,8 @@ public sealed class OrderBatchJobEventHandler(
return;
}
using var unitOfWork = unitOfWorkManager.Begin();
// ForecastBilling 成功后继续写轨迹/面单/自动触发 Outbox，须事务性 UoW
using var unitOfWork = unitOfWorkManager.Begin(OmsOutboxUnitOfWork.Transactional());
await processor.ProcessAsync(@event.ToMessage(), cancellationToken);
await unitOfWork.CompleteAsync(cancellationToken);
}

9
src/Oms.Application/Messaging/OrderBatchJobPublisher.cs
View File

@ -1,14 +1,16 @@
using Fake.EventBus.Distributed;
using Fake.UnitOfWork;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
namespace Oms.Application.Messaging;
/// <summary>
/// 通过 Fake 分布式事件总线发布订单批量任务。
/// 通过 Fake 分布式事件总线（Outbox）发布订单批量任务。
/// </summary>
public sealed class OrderBatchJobPublisher(
IDistributedEventBus eventBus,
IUnitOfWorkManager unitOfWorkManager,
IOptions<OrderBatchRabbitMqOptions> options,
ILogger<OrderBatchJobPublisher> logger) : IOrderBatchJobPublisher
{
@ -22,10 +24,11 @@ public sealed class OrderBatchJobPublisher(
}
var @event = OrderBatchJobIntegrationEvent.FromMessage(message);
await eventBus.PublishAsync(@event, cancellationToken);
await OmsIntegrationEventPublisher.PublishAsync(
eventBus, unitOfWorkManager, logger, @event, cancellationToken);
logger.LogDebug(
"Published order batch event {EventId} messageId={MessageId} action={Action}",
"Outbox enqueued order batch event {EventId} messageId={MessageId} action={Action}",
@event.Id,
message.MessageId,
message.Action);

65
src/Oms.Application/Messaging/OrderSideEffects/OrderSideEffectEventPublisher.cs
View File

@ -1,16 +1,16 @@
using Fake.EventBus.Distributed;
using Microsoft.Extensions.DependencyInjection;
using Fake.UnitOfWork;
using Microsoft.Extensions.Logging;
using Oms.Application.Services.Orders.Issues;
namespace Oms.Application.Messaging.OrderSideEffects;
/// <summary>
/// 优先 Fake.EventBus.RabbitMQ；发布失败则同进程同步执行 Handler。
/// 订单旁路副作用事件：事务内写 Outbox，Commit 后投递 RabbitMQ。
/// </summary>
public sealed class OrderSideEffectEventPublisher(
IDistributedEventBus eventBus,
IServiceScopeFactory scopeFactory,
IUnitOfWorkManager unitOfWorkManager,
ILogger<OrderSideEffectEventPublisher> logger) : IOrderSideEffectEventPublisher
{
public async Task PublishLifecycleTrackAppendAsync(
@ -33,14 +33,8 @@ public sealed class OrderSideEffectEventPublisher(
Authorization = string.IsNullOrWhiteSpace(authorization) ? null : authorization.Trim()
};
if (await TryPublishAsync(@event, cancellationToken))
{
return;
}
await ExecuteInScopeAsync<OrderLifecycleTrackAppendEventHandler>(
h => h.HandleAsync(@event, cancellationToken),
cancellationToken);
await OmsIntegrationEventPublisher.PublishAsync(
eventBus, unitOfWorkManager, logger, @event, cancellationToken);
}
public async Task PublishAutoTriggerFireAsync(
@ -61,52 +55,7 @@ public sealed class OrderSideEffectEventPublisher(
OperatorUserId = operatorUserId
};
if (await TryPublishAsync(@event, cancellationToken))
{
return;
}
await ExecuteInScopeAsync<OrderAutoTriggerFireEventHandler>(
h => h.HandleAsync(@event, cancellationToken),
cancellationToken);
}
private async Task<bool> TryPublishAsync(IntegrationEvent @event, CancellationToken cancellationToken)
{
try
{
await eventBus.PublishAsync(@event, cancellationToken);
logger.LogDebug(
"Published side-effect event {EventName} Id={EventId}",
@event.GetType().Name,
@event.Id);
return true;
}
catch (Exception ex)
{
logger.LogWarning(
ex,
"Publish side-effect event {EventName} failed, fallback to sync. EventId={EventId}",
@event.GetType().Name,
@event.Id);
return false;
}
}
private async Task ExecuteInScopeAsync<THandler>(
Func<THandler, Task> action,
CancellationToken cancellationToken)
where THandler : notnull
{
try
{
using var scope = scopeFactory.CreateScope();
var handler = scope.ServiceProvider.GetRequiredService<THandler>();
await action(handler);
}
catch (Exception ex)
{
logger.LogWarning(ex, "Sync fallback for {Handler} failed", typeof(THandler).Name);
}
await OmsIntegrationEventPublisher.PublishAsync(
eventBus, unitOfWorkManager, logger, @event, cancellationToken);
}
}

79
src/Oms.Application/Messaging/ParcelCreate/OrderParcelCreateEventPublisher.cs
View File

@ -1,4 +1,5 @@
using Fake.EventBus.Distributed;
using Fake.UnitOfWork;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using Newtonsoft.Json;
@ -11,10 +12,12 @@ using Oms.Domain.OrdersAggregate.Entities;
namespace Oms.Application.Messaging.ParcelCreate;
/// <summary>
/// 创单后置事件发布器：优先走 Fake.EventBus.RabbitMQ，失败则同进程同步执行 Handler 逻辑。
/// 创单后置事件发布器：事务内写 Outbox（Fake.EventBus.IntegrationEventLog），Commit 后投递 RabbitMQ。
/// 揽收 Publish 失败时由调用方同步 ExecutePickupAttachAsync 回落。
/// </summary>
public sealed class OrderParcelCreateEventPublisher(
IDistributedEventBus eventBus,
IUnitOfWorkManager unitOfWorkManager,
IServiceScopeFactory scopeFactory,
ILogger<OrderParcelCreateEventPublisher> logger) : IOrderParcelCreateEventPublisher
{
@ -33,14 +36,8 @@ public sealed class OrderParcelCreateEventPublisher(
Authorization = request.Authorization
};
if (await TryPublishAsync(@event, cancellationToken))
{
return;
}
await ExecuteInScopeAsync<OrderParcelTrackUpsertEventHandler>(
h => h.HandleAsync(@event, cancellationToken),
cancellationToken);
await OmsIntegrationEventPublisher.PublishAsync(
eventBus, unitOfWorkManager, logger, @event, cancellationToken);
}
public async Task PublishLabelGenerateAsync(
@ -54,14 +51,8 @@ public sealed class OrderParcelCreateEventPublisher(
OperatorUserId = operatorUserId
};
if (await TryPublishAsync(@event, cancellationToken))
{
return;
}
await ExecuteInScopeAsync<OrderParcelLabelGenerateEventHandler>(
h => h.HandleAsync(@event, cancellationToken),
cancellationToken);
await OmsIntegrationEventPublisher.PublishAsync(
eventBus, unitOfWorkManager, logger, @event, cancellationToken);
}
public async Task PublishCustoms9610UpsertAsync(
@ -75,14 +66,8 @@ public sealed class OrderParcelCreateEventPublisher(
PlatformCode = platformCode
};
if (await TryPublishAsync(@event, cancellationToken))
{
return;
}
await ExecuteInScopeAsync<OrderParcelCustoms9610UpsertEventHandler>(
h => h.HandleAsync(@event, cancellationToken),
cancellationToken);
await OmsIntegrationEventPublisher.PublishAsync(
eventBus, unitOfWorkManager, logger, @event, cancellationToken);
}
public async Task<bool> PublishPickupAttachAsync(
@ -96,8 +81,8 @@ public sealed class OrderParcelCreateEventPublisher(
return false;
}
// 揽收回落由调用方 ExecutePickupAttachAsync 同步执行，避免重复建单且便于回写单号。
return await TryPublishAsync(@event, cancellationToken);
return await OmsIntegrationEventPublisher.TryPublishAsync(
eventBus, unitOfWorkManager, logger, @event, cancellationToken);
}
public async Task<PickupOrderOutput?> ExecutePickupAttachAsync(
@ -155,44 +140,4 @@ public sealed class OrderParcelCreateEventPublisher(
ExpectPickupTime = input.ExpectPickupTime.Value
};
}
private async Task<bool> TryPublishAsync(IntegrationEvent @event, CancellationToken cancellationToken)
{
try
{
await eventBus.PublishAsync(@event, cancellationToken);
logger.LogDebug(
"Published parcel-create event {EventName} Id={EventId}",
@event.GetType().Name,
@event.Id);
return true;
}
catch (Exception ex)
{
logger.LogWarning(
ex,
"Publish parcel-create event {EventName} failed, fallback to sync. EventId={EventId}",
@event.GetType().Name,
@event.Id);
return false;
}
}
private async Task ExecuteInScopeAsync<THandler>(
Func<THandler, Task> action,
CancellationToken cancellationToken)
where THandler : notnull
{
try
{
using var scope = scopeFactory.CreateScope();
var handler = scope.ServiceProvider.GetRequiredService<THandler>();
await action(handler);
}
catch (Exception ex)
{
// 同步回落失败也不抛回创单主流程
logger.LogWarning(ex, "Sync fallback for {Handler} failed", typeof(THandler).Name);
}
}
}

1
src/Oms.Application/Oms.Application.csproj
View File

@ -18,6 +18,7 @@
</ItemGroup>
<ItemGroup>
<PackageReference Include="Fake.EventBus.IntegrationEventLog" Version="$(FakePackageVersion)" />
<PackageReference Include="Fake.EventBus.RabbitMQ" Version="$(FakePackageVersion)" />
<PackageReference Include="Fake.ObjectMapping.AutoMapper" Version="$(FakePackageVersion)" />
<PackageReference Include="Fake.ObjectStorage" Version="$(FakePackageVersion)" />

2
src/Oms.Application/Rbac/UserProfileRepository.cs
View File

@ -187,6 +187,8 @@ public class UserProfileRepository(FakeRbacDbContext dbContext) : IScopedDepende
if (int.TryParse(match.Groups[1].Value, out var number))
{
max = Math.Max(max, number);
}
}

7
src/Oms.Application/Services/Operations/InboundRecordService.cs
View File

@ -1,4 +1,6 @@
using System.Data;
using Fake.Application;
using Fake.UnitOfWork;
using Microsoft.AspNetCore.Mvc;
using Oms.Application.Dtos.Operations;
using Oms.Application.Dtos.Shared;
@ -85,6 +87,7 @@ public class InboundRecordService : ApplicationService
/// <param name="input">入参：入库记录创建参数。</param>
/// <returns>入库记录结果。</returns>
[HttpPost("/operation/inbound/create")]
[UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)]
public async Task<InboundRecordOutput> CreateAsync(CreateInboundRecordInput input)
{
if (string.IsNullOrWhiteSpace(input.WaybillNo))
@ -141,8 +144,8 @@ public class InboundRecordService : ApplicationService
var savedEntity = await InboundRecordRepository.InsertAsync(entity);
// 自动触发改异步，避免拉长扫码 RT
AutoTriggerRuntimeService.EnqueueFire(order.Id, AutoTriggerTiming.Inbound, operatorId);
// 自动触发走 Outbox，避免拉长扫码 RT
await AutoTriggerRuntimeService.EnqueueFireAsync(order.Id, AutoTriggerTiming.Inbound, operatorId);
var refreshed = await OrderRepository.GetByWaybillNoAsync(input.WaybillNo.Trim());
var output = ObjectMapper.Map<InboundRecord, InboundRecordOutput>(savedEntity);

7
src/Oms.Application/Services/Operations/OutboundRecordService.cs
View File

@ -1,4 +1,6 @@
using System.Data;
using Fake.Application;
using Fake.UnitOfWork;
using Microsoft.AspNetCore.Mvc;
using Oms.Application.Dtos.Operations;
using Oms.Application.Dtos.Shared;
@ -99,6 +101,7 @@ public class OutboundRecordService : ApplicationService
/// <param name="input">入参：出库记录创建参数。</param>
/// <returns>出库记录结果。</returns>
[HttpPost("/operation/outbound/create")]
[UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)]
public async Task<OutboundRecordOutput> CreateAsync(CreateOutboundRecordInput input)
{
if (string.IsNullOrWhiteSpace(input.WaybillNo))
@ -161,8 +164,8 @@ public class OutboundRecordService : ApplicationService
var savedEntity = await OutboundRecordRepository.InsertAsync(entity);
// 自动触发改走 Fake.EventBus，避免拉长出库扫码 RT
AutoTriggerRuntimeService.EnqueueFire(order.Id, AutoTriggerTiming.Outbound, operatorId);
// 自动触发走 Outbox，避免拉长出库扫码 RT
await AutoTriggerRuntimeService.EnqueueFireAsync(order.Id, AutoTriggerTiming.Outbound, operatorId);
return ObjectMapper.Map<OutboundRecord, OutboundRecordOutput>(savedEntity);
}

15
src/Oms.Application/Services/Orders/Issues/AutoTriggerRuntimeService.cs
View File

@ -22,17 +22,12 @@ public class AutoTriggerRuntimeService(
private const string InterceptOperation = "拦截";
/// <summary>
/// 后台异步触发（Fake.EventBus.RabbitMQ），避免拉长入库/出库/创单请求。
/// 投递自动触发事件（须在事务 UoW 内调用，写入 Outbox）。
/// </summary>
public void EnqueueFire(int orderId, AutoTriggerTiming timing, int operatorUserId)
{
if (orderId <= 0)
{
return;
}
_ = sideEffectEventPublisher.PublishAutoTriggerFireAsync(orderId, timing, operatorUserId);
}
public Task EnqueueFireAsync(int orderId, AutoTriggerTiming timing, int operatorUserId) =>
orderId <= 0
? Task.CompletedTask
: sideEffectEventPublisher.PublishAutoTriggerFireAsync(orderId, timing, operatorUserId);
public async Task FireAsync(Order order, AutoTriggerTiming timing, int operatorUserId)
{

2
src/Oms.Application/Services/Orders/OrderService.ParcelBilling.cs
View File

@ -2800,7 +2800,7 @@ public partial class OrderService
appendInitialTrack: true,
operatorUserId);
AutoTriggerRuntimeService.EnqueueFire(
await AutoTriggerRuntimeService.EnqueueFireAsync(
order.Id,
AutoTriggerTiming.OrderCreate,
operatorUserId);

6
src/Oms.Application/Services/Orders/OrderService.cs
View File

@ -1,7 +1,9 @@
using System.Data;
using System.Globalization;
using Fake.Application;
using Fake.Domain.Exceptions;
using Fake.ObjectStorage;
using Fake.UnitOfWork;
using Microsoft.AspNetCore.Hosting;
using Microsoft.AspNetCore.Mvc;
using MiniExcelLibs;
@ -265,6 +267,7 @@ public partial class OrderService : ApplicationService
/// <param name="input">入参：订单操作参数。</param>
/// <returns>无返回体，仅完成订单预报处理。</returns>
[HttpPost("/order/pre-declare")]
[UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)]
public async Task PreDeclareAsync(OrderActionInput input)
{
var order = await GetOrderOrThrowAsync(input.Id);
@ -605,6 +608,7 @@ public partial class OrderService : ApplicationService
/// <param name="input">入参：订单包裹创建参数。</param>
/// <returns>订单包裹创建结果。</returns>
[HttpPost("/order/parcel/create")]
[UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)]
public async Task<OrderParcelCreateOutput> CreateParcelAsync(OrderParcelCreateInput input)
{
var inbound = string.IsNullOrWhiteSpace(input.InboundMethod)
@ -636,6 +640,7 @@ public partial class OrderService : ApplicationService
/// <summary>
/// 客户 OpenAPI 创单：同一客户相同 clientNo 视为幂等，返回已存在订单，不新建。
/// </summary>
[UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)]
public async Task<OrderParcelCreateOutput> CreateParcelForCustomerOpenApiAsync(OrderParcelCreateInput input)
{
NormalizeParcelCreateInput(input);
@ -734,6 +739,7 @@ public partial class OrderService : ApplicationService
/// <param name="input">入参：批量创建小包请求集合，每条记录沿用单票创建结构。</param>
/// <returns>批量处理结果，包含批次号、成功失败数量与逐条执行明细。</returns>
[HttpPost("/order/parcel/batch-create")]
[UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)]
public async Task<OrderParcelBatchCreateOutput> BatchCreateParcelAsync(OrderParcelBatchCreateInput input)
{
if (input == null || input.Count == 0)

23
src/Oms.Tests/Messaging/Fakes/RecordingPhysicalDistributedEventBus.cs Normal file
View File

@ -0,0 +1,23 @@
using System.Collections.Concurrent;
using Fake.EventBus;
using Fake.EventBus.Distributed;
namespace Oms.Tests.Messaging.Fakes;
public sealed class RecordingPhysicalDistributedEventBus : IPhysicalDistributedEventBus
{
private readonly ConcurrentQueue<IntegrationEvent> _published = new();
public IReadOnlyList<IntegrationEvent> Published => _published.ToArray();
public void Clear() => _published.Clear();
public Task PublishAsync(IntegrationEvent @event, CancellationToken cancellationToken = default)
{
_published.Enqueue(@event);
return Task.CompletedTask;
}
public Task PublishAsync(Event @event, CancellationToken cancellationToken = default)
=> PublishAsync((IntegrationEvent)@event, cancellationToken);
}

51
src/Oms.Tests/Messaging/OmsOutboxTestModule.cs Normal file
View File

@ -0,0 +1,51 @@
using Fake.Autofac;
using Fake.EntityFrameworkCore;
using Fake.EventBus.Distributed;
using Fake.EventBus.IntegrationEventLog;
using Fake.Modularity;
using Fake.SyncEx;
using Microsoft.Data.Sqlite;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.DependencyInjection;
using Oms.Tests.Messaging.Fakes;
namespace Oms.Tests.Messaging;
[DependsOn(typeof(FakeAutofacModule))]
[DependsOn(typeof(FakeEntityFrameworkCoreModule))]
[DependsOn(typeof(FakeEventBusIntegrationEventLogModule))]
public class OmsOutboxTestModule : FakeModule
{
public override void ConfigureServices(ServiceConfigurationContext context)
{
context.Services.AddSingleton<RecordingPhysicalDistributedEventBus>();
context.Services.AddSingleton<IPhysicalDistributedEventBus>(sp =>
sp.GetRequiredService<RecordingPhysicalDistributedEventBus>());
context.Services.Configure<OutboxPublisherOptions>(o =>
{
o.StartupDelay = TimeSpan.FromHours(1);
o.ScanInterval = TimeSpan.FromHours(1);
});
context.Services.Configure<InboxCleanupOptions>(o =>
{
o.StartupDelay = TimeSpan.FromHours(1);
o.CleanupInterval = TimeSpan.FromHours(1);
});
var connection = new SqliteConnection("Filename=:memory:");
connection.Open();
context.Services.AddDbContext<IntegrationEventLogContext>(builder =>
{
builder.UseSqlite(connection);
}, ServiceLifetime.Scoped, ServiceLifetime.Singleton);
}
public override void PreConfigureApplication(ApplicationConfigureContext context)
{
using var scope = context.ServiceProvider.CreateScope();
var ctx = scope.ServiceProvider.GetRequiredService<IntegrationEventLogContext>();
SyncContext.Run(async () => await ctx.Database.EnsureCreatedAsync());
}
}

90
src/Oms.Tests/Messaging/OrderBatchJobOutboxPublisherTests.cs Normal file
View File

@ -0,0 +1,90 @@
using Fake;
using Fake.EntityFrameworkCore;
using Fake.EventBus.Distributed;
using Fake.EventBus.IntegrationEventLog;
using Fake.Testing;
using Fake.UnitOfWork;
using FluentAssertions;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging.Abstractions;
using Microsoft.Extensions.Options;
using Oms.Application.Messaging;
using Oms.Domain.OrdersAggregate.Consts;
using Oms.Tests.Messaging.Fakes;
namespace Oms.Tests.Messaging;
public class OrderBatchJobOutboxPublisherTests : ApplicationTestWithTools<OmsOutboxTestModule>
{
private readonly IUnitOfWorkManager _uowManager;
private readonly RecordingPhysicalDistributedEventBus _physicalBus;
private readonly IEfDbContextProvider<IntegrationEventLogContext> _dbProvider;
private readonly OrderBatchJobPublisher _publisher;
protected override void SetApplicationCreationOptions(FakeApplicationCreationOptions options)
{
options.UseAutofac();
}
public OrderBatchJobOutboxPublisherTests()
{
_uowManager = ServiceProvider.GetRequiredService<IUnitOfWorkManager>();
_physicalBus = ServiceProvider.GetRequiredService<RecordingPhysicalDistributedEventBus>();
_dbProvider = ServiceProvider.GetRequiredService<IEfDbContextProvider<IntegrationEventLogContext>>();
_publisher = new OrderBatchJobPublisher(
ServiceProvider.GetRequiredService<IDistributedEventBus>(),
_uowManager,
Options.Create(new OrderBatchRabbitMqOptions { Enabled = true }),
NullLogger<OrderBatchJobPublisher>.Instance);
}
[Fact]
public async Task ForecastBilling任务消息_事务Commit后投递()
{
_physicalBus.Clear();
var message = new OrderBatchJobMessage
{
JobId = 1,
JobItemId = 2,
Action = OrderBatchActionCodes.ForecastBilling,
OrderId = 55,
MessageId = "job:1:55:forecast-billing",
OperatorUserId = 9,
PayloadJson = "{\"CouponInstanceId\":1}"
};
using (var uow = _uowManager.Begin(OmsOutboxUnitOfWork.Transactional()))
{
await _dbProvider.GetDbContextAsync();
await _publisher.PublishAsync(message);
_physicalBus.Published.Should().BeEmpty();
await uow.CompleteAsync();
}
var evt = _physicalBus.Published.Should().ContainSingle()
.Which.Should().BeOfType<OrderBatchJobIntegrationEvent>().Subject;
evt.Action.Should().Be(OrderBatchActionCodes.ForecastBilling);
evt.OrderId.Should().Be(55);
evt.MessageId.Should().Be(message.MessageId);
}
[Fact]
public async Task 队列未启用时_Publish应抛错()
{
var disabled = new OrderBatchJobPublisher(
ServiceProvider.GetRequiredService<IDistributedEventBus>(),
_uowManager,
Options.Create(new OrderBatchRabbitMqOptions { Enabled = false }),
NullLogger<OrderBatchJobPublisher>.Instance);
var act = () => disabled.PublishAsync(new OrderBatchJobMessage
{
Action = OrderBatchActionCodes.ForecastBilling,
OrderId = 1,
MessageId = "x"
});
await act.Should().ThrowAsync<InvalidOperationException>()
.WithMessage("*disabled*");
}
}

131
src/Oms.Tests/Messaging/OrderParcelCreateOutboxPublisherTests.cs Normal file
View File

@ -0,0 +1,131 @@
using Fake;
using Fake.EntityFrameworkCore;
using Fake.EventBus.Distributed;
using Fake.EventBus.IntegrationEventLog;
using Fake.Testing;
using Fake.UnitOfWork;
using FluentAssertions;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging.Abstractions;
using Oms.Application.Integrations.SmallBagTrack;
using Oms.Application.Messaging;
using Oms.Application.Messaging.ParcelCreate;
using Oms.Domain.OrdersAggregate.Entities;
using Oms.Tests.Messaging.Fakes;
namespace Oms.Tests.Messaging;
public class OrderParcelCreateOutboxPublisherTests : ApplicationTestWithTools<OmsOutboxTestModule>
{
private readonly IUnitOfWorkManager _uowManager;
private readonly IDistributedEventBus _distributedEventBus;
private readonly RecordingPhysicalDistributedEventBus _physicalBus;
private readonly IOutboxEventLogService _outbox;
private readonly IEfDbContextProvider<IntegrationEventLogContext> _dbProvider;
private readonly OrderParcelCreateEventPublisher _publisher;
protected override void SetApplicationCreationOptions(FakeApplicationCreationOptions options)
{
options.UseAutofac();
}
public OrderParcelCreateOutboxPublisherTests()
{
_uowManager = ServiceProvider.GetRequiredService<IUnitOfWorkManager>();
_distributedEventBus = ServiceProvider.GetRequiredService<IDistributedEventBus>();
_physicalBus = ServiceProvider.GetRequiredService<RecordingPhysicalDistributedEventBus>();
_outbox = ServiceProvider.GetRequiredService<IOutboxEventLogService>();
_dbProvider = ServiceProvider.GetRequiredService<IEfDbContextProvider<IntegrationEventLogContext>>();
_publisher = new OrderParcelCreateEventPublisher(
_distributedEventBus,
_uowManager,
ServiceProvider.GetRequiredService<IServiceScopeFactory>(),
NullLogger<OrderParcelCreateEventPublisher>.Instance);
}
[Fact]
public void IDistributedEventBus_应为Outbox装饰器()
{
_distributedEventBus.Should().BeOfType<OutboxDistributedEventBus>();
}
[Fact]
public async Task 事务Uow内发布9610_应在Complete后投递物理总线()
{
_physicalBus.Clear();
var order = new Order { Id = 101, ServiceNo = "SN101" };
using (var uow = _uowManager.Begin(OmsOutboxUnitOfWork.Transactional()))
{
await _dbProvider.GetDbContextAsync();
await _publisher.PublishCustoms9610UpsertAsync(order, "lazada");
_physicalBus.Published.Should().BeEmpty();
await uow.CompleteAsync();
}
_physicalBus.Published.Should().ContainSingle();
var evt = _physicalBus.Published[0].Should().BeOfType<OrderParcelCustoms9610UpsertIntegrationEvent>().Subject;
evt.OrderId.Should().Be(101);
evt.PlatformCode.Should().Be("lazada");
var pending = (await _outbox.RetrieveEventLogsPendingToPublishAsync(50)).ToList();
pending.Should().NotContain(e => e.EventId == evt.Id);
}
[Fact]
public async Task 事务Uow内发布轨迹与面单_Commit后应各投递一次()
{
_physicalBus.Clear();
var order = new Order { Id = 202, ServiceNo = "SN202" };
var request = new SmallBagTrackUpsertRequest
{
OrderNo = "SN202",
AppendInitialTrackOnSuccess = true
};
using (var uow = _uowManager.Begin(OmsOutboxUnitOfWork.Transactional()))
{
await _dbProvider.GetDbContextAsync();
await _publisher.PublishTrackUpsertAsync(order, request, customerId: 9);
await _publisher.PublishLabelGenerateAsync(order, operatorUserId: 7);
_physicalBus.Published.Should().BeEmpty();
await uow.CompleteAsync();
}
_physicalBus.Published.Should().HaveCount(2);
_physicalBus.Published.OfType<OrderParcelTrackUpsertIntegrationEvent>().Should().ContainSingle(e =>
e.OrderId == 202 && e.CustomerId == 9 && e.AppendInitialTrackOnSuccess);
_physicalBus.Published.OfType<OrderParcelLabelGenerateIntegrationEvent>().Should().ContainSingle(e =>
e.OrderId == 202 && e.OperatorUserId == 7);
}
[Fact]
public async Task 回滚后不应投递()
{
_physicalBus.Clear();
var order = new Order { Id = 303, ServiceNo = "SN303" };
using (var uow = _uowManager.Begin(OmsOutboxUnitOfWork.Transactional()))
{
await _dbProvider.GetDbContextAsync();
await _publisher.PublishCustoms9610UpsertAsync(order);
await uow.RollbackAsync();
}
_physicalBus.Published.Should().BeEmpty();
}
[Fact]
public async Task 无环境事务时_短事务仍能写入并投递()
{
_physicalBus.Clear();
_uowManager.Current.Should().BeNull();
var order = new Order { Id = 404, ServiceNo = "SN404" };
await _publisher.PublishCustoms9610UpsertAsync(order, "shopee");
_physicalBus.Published.Should().ContainSingle()
.Which.Should().BeOfType<OrderParcelCustoms9610UpsertIntegrationEvent>()
.Which.OrderId.Should().Be(404);
}
}

62
src/Oms.Tests/Messaging/OrderSideEffectOutboxPublisherTests.cs Normal file
View File

@ -0,0 +1,62 @@
using Fake;
using Fake.EntityFrameworkCore;
using Fake.EventBus.Distributed;
using Fake.EventBus.IntegrationEventLog;
using Fake.Testing;
using Fake.UnitOfWork;
using FluentAssertions;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging.Abstractions;
using Oms.Application.Messaging;
using Oms.Application.Messaging.OrderSideEffects;
using Oms.Application.Services.Orders.Issues;
using Oms.Tests.Messaging.Fakes;
namespace Oms.Tests.Messaging;
public class OrderSideEffectOutboxPublisherTests : ApplicationTestWithTools<OmsOutboxTestModule>
{
private readonly IUnitOfWorkManager _uowManager;
private readonly IDistributedEventBus _distributedEventBus;
private readonly RecordingPhysicalDistributedEventBus _physicalBus;
private readonly IEfDbContextProvider<IntegrationEventLogContext> _dbProvider;
private readonly OrderSideEffectEventPublisher _publisher;
protected override void SetApplicationCreationOptions(FakeApplicationCreationOptions options)
{
options.UseAutofac();
}
public OrderSideEffectOutboxPublisherTests()
{
_uowManager = ServiceProvider.GetRequiredService<IUnitOfWorkManager>();
_distributedEventBus = ServiceProvider.GetRequiredService<IDistributedEventBus>();
_physicalBus = ServiceProvider.GetRequiredService<RecordingPhysicalDistributedEventBus>();
_dbProvider = ServiceProvider.GetRequiredService<IEfDbContextProvider<IntegrationEventLogContext>>();
_publisher = new OrderSideEffectEventPublisher(
_distributedEventBus,
_uowManager,
NullLogger<OrderSideEffectEventPublisher>.Instance);
}
[Fact]
public async Task 事务内发布自动触发与生命周期轨迹_Commit后投递()
{
_physicalBus.Clear();
using (var uow = _uowManager.Begin(OmsOutboxUnitOfWork.Transactional()))
{
await _dbProvider.GetDbContextAsync();
await _publisher.PublishAutoTriggerFireAsync(11, AutoTriggerTiming.OrderCreate, 3);
await _publisher.PublishLifecycleTrackAppendAsync(11, "SN11", "A0301");
_physicalBus.Published.Should().BeEmpty();
await uow.CompleteAsync();
}
_physicalBus.Published.Should().HaveCount(2);
_physicalBus.Published.OfType<OrderAutoTriggerFireIntegrationEvent>().Should().ContainSingle(e =>
e.OrderId == 11 && e.Timing == AutoTriggerTiming.OrderCreate.ToString() && e.OperatorUserId == 3);
_physicalBus.Published.OfType<OrderLifecycleTrackAppendIntegrationEvent>().Should().ContainSingle(e =>
e.OrderId == 11 && e.OrderNo == "SN11");
}
}

5
src/Oms.Tests/Oms.Tests.csproj
View File

@ -13,9 +13,14 @@
<PrivateAssets>all</PrivateAssets>
<IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
</PackageReference>
<PackageReference Include="Fake.Autofac" Version="$(FakePackageVersion)" />
<PackageReference Include="Fake.EventBus.IntegrationEventLog" Version="$(FakePackageVersion)" />
<PackageReference Include="Fake.Testing" Version="$(FakePackageVersion)" />
<PackageReference Include="FluentAssertions" Version="8.7.1" />
<PackageReference Include="Microsoft.EntityFrameworkCore.Sqlite" Version="8.0.11" />
<PackageReference Include="Microsoft.NET.Test.Sdk" Version="17.12.0" />
<PackageReference Include="Moq" Version="4.20.72" />
<PackageReference Include="Shouldly" Version="4.3.0" />
<PackageReference Include="xunit" Version="2.9.2" />
<PackageReference Include="xunit.runner.visualstudio" Version="2.8.2">
<PrivateAssets>all</PrivateAssets>

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

@ -15,12 +15,13 @@
<ItemGroup>
<PackageReference Include="Fake.Autofac" Version="$(FakePackageVersion)" />
<PackageReference Include="Fake.Consul" Version="$(FakePackageVersion)" />
<PackageReference Include="Fake.EventBus.IntegrationEventLog" Version="$(FakePackageVersion)" />
<PackageReference Include="Fake.FileManagement.Application" Version="$(FakePackageVersion)" />
<PackageReference Include="Fake.FileManagement.Infrastructure" Version="$(FakePackageVersion)" />
<PackageReference Include="Fake.ObjectStorage.TencentCos" Version="$(FakePackageVersion)" />
<PackageReference Include="Fake.Rbac.Application" Version="$(FakePackageVersion)" />
<PackageReference Include="Fake.Serilog" Version="$(FakePackageVersion)" />
<PackageReference Include="Serilog.AspNetCore" Version="8.0.3"/>
<PackageReference Include="Serilog.AspNetCore" Version="8.0.3" />
</ItemGroup>
<ItemGroup>

22
src/SimpleAdmin.Api/SimpleAdminApiModule.cs
View File

@ -7,6 +7,7 @@ using Fake.AspNetCore.Auditing;
using Fake.AspNetCore.ExceptionHandling;
using Fake.AspNetCore.Newtonsoft;
using Fake.AspNetCore.Swagger;
using Fake.EventBus.IntegrationEventLog;
using Fake.FeiShu;
using Fake.FileManagement.Application;
using Fake.FileManagement.Infrastructure;
@ -37,6 +38,8 @@ namespace SimpleAdmin.Api;
typeof(FakeFileManagementApplicationModule),
typeof(FakeFileManagementInfrastructureModule),
typeof(FakeObjectStorageTencentCosModule),
// 在 RabbitMQ 物理总线之后 Replace IDistributedEventBus → Outbox
typeof(FakeEventBusIntegrationEventLogModule),
typeof(FakeAspNetCoreModule),
typeof(FakeSerilogModule)
)]
@ -80,6 +83,20 @@ public class SimpleAdminApiModule : FakeModule
services.AddOmsWeChat(configuration);
services.AddSingleton<IRbacAuthCache, RbacAuthCache>();
// Outbox / Inbox 与业务同库（EF），通过 DbTransaction 征用 SqlSugar 事务
services.AddDbContext<IntegrationEventLogContext>(options =>
{
var connectionString = configuration.GetConnectionString("IntegrationEventLog")
?? configuration.GetConnectionString("Default")
?? configuration["ConnectionStrings:Default"]
?? string.Empty;
var serverVersion = ServerVersion.AutoDetect(connectionString);
options.UseMySql(connectionString, serverVersion);
#if DEBUG
options.EnableSensitiveDataLogging();
#endif
});
//Json格式化全局处理
context.Services.AddFakeNewtonsoft();
@ -188,11 +205,14 @@ public class SimpleAdminApiModule : FakeModule
// 端点映射
app.MapControllers();
// FileManagement 与 RBAC 同库并行（EF）；应用 Pending migrations 建 stored_file
// FileManagement / Outbox 与业务同库并行（EF）
using (var scope = app.Services.CreateScope())
{
var fileDb = scope.ServiceProvider.GetRequiredService<FileManagementDbContext>();
fileDb.Database.Migrate();
var outboxDb = scope.ServiceProvider.GetRequiredService<IntegrationEventLogContext>();
IntegrationEventLogSchema.EnsureTablesAsync(outboxDb).GetAwaiter().GetResult();
}
app.Services.GetRequiredService<IFeiShuNotificationService>().Send("Application started");

48
src/SimpleAdmin.Api/Storage/IntegrationEventLogSchema.cs Normal file
View File

@ -0,0 +1,48 @@
using Fake.EventBus.IntegrationEventLog;
using Microsoft.EntityFrameworkCore;
namespace SimpleAdmin.Api.Storage;
/// <summary>
/// Outbox/Inbox 与业务同库时 EnsureCreated 无效（库已存在），用 IF NOT EXISTS 建表。
/// </summary>
internal static class IntegrationEventLogSchema
{
public static async Task EnsureTablesAsync(
IntegrationEventLogContext db,
CancellationToken cancellationToken = default)
{
await db.Database.ExecuteSqlRawAsync(
"""
CREATE TABLE IF NOT EXISTS `OutboxEventLog` (
`EventId` char(36) CHARACTER SET ascii COLLATE ascii_general_ci NOT NULL,
`EventTypeName` varchar(512) NOT NULL,
`State` int NOT NULL,
`TimesSent` int NOT NULL,
`CreationTime` datetime(6) NOT NULL,
`Content` longtext NOT NULL,
`TransactionId` varchar(64) NOT NULL,
`LockExpiresAt` datetime(6) NULL,
PRIMARY KEY (`EventId`),
KEY `IX_OutboxEventLog_State_CreationTime` (`State`, `CreationTime`),
KEY `IX_OutboxEventLog_TransactionId` (`TransactionId`)
) CHARACTER SET utf8mb4;
""",
cancellationToken);
await db.Database.ExecuteSqlRawAsync(
"""
CREATE TABLE IF NOT EXISTS `InboxEventLog` (
`EventId` char(36) CHARACTER SET ascii COLLATE ascii_general_ci NOT NULL,
`EventTypeName` varchar(512) NOT NULL,
`State` int NOT NULL,
`ProcessedTime` datetime(6) NOT NULL,
`Content` longtext NOT NULL,
`ErrorMessage` longtext NULL,
PRIMARY KEY (`EventId`),
KEY `IX_InboxEventLog_ProcessedTime` (`ProcessedTime`)
) CHARACTER SET utf8mb4;
""",
cancellationToken);
}
}

8
src/SimpleAdmin.Api/appsettings.Development.json
View File

@ -146,5 +146,13 @@
"RabbitMq": {
"Enabled": false
}
},
"IntegrationEventLog": {
"OutboxPublisher": {
"StartupDelay": "00:00:05",
"ScanInterval": "00:00:10",
"BatchSize": 100,
"MaxDegreeOfParallelism": 8
}
}
}