feat(IntegrationEventBus): Support does not specify local message table service

Author: zhenlei520

src/Dispatcher/Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF/DispatcherOptionsExtensions.cs

@@ -22,6 +22,8 @@ public static IDispatcherOptions UseEventLog<TDbContext>(
 
         options.Services.AddSingleton<EventLogProvider>();
 
+        options.Services.TryAddScoped<IIntegrationEventLogService, IntegrationEventLogService>();
+
         //Add local message table model mapping
         options.Services.TryAddEnumerable(new ServiceDescriptor(typeof(IModelCreatingProvider),
             typeof(IntegrationEventLogModelCreatingProvider), ServiceLifetime.Singleton));

src/Dispatcher/Masa.Contrib.Dispatcher.IntegrationEvents/DispatcherOptionsExtensions.cs

@@ -5,17 +5,25 @@ namespace Masa.Contrib.Dispatcher.IntegrationEvents;
 
 public static class DispatcherOptionsExtensions
 {
+    public static IDistributedDispatcherOptions UseIntegrationEventBus(
+        this IDistributedDispatcherOptions dispatcherOptions,
+        Action<DispatcherOptions>? optionAction = null)
+    {
+        ArgumentNullException.ThrowIfNull(dispatcherOptions.Services, nameof(dispatcherOptions.Services));
+
+        dispatcherOptions.Services.TryAddIntegrationEventBus(dispatcherOptions.Assemblies, option => optionAction?.Invoke(option));
+        return dispatcherOptions;
+    }
+
     public static IDistributedDispatcherOptions UseIntegrationEventBus<TIntegrationEventLogService>(
         this IDistributedDispatcherOptions dispatcherOptions,
         Action<DispatcherOptions>? optionAction = null)
         where TIntegrationEventLogService : class, IIntegrationEventLogService
     {
         ArgumentNullException.ThrowIfNull(dispatcherOptions.Services, nameof(dispatcherOptions.Services));
 
-        dispatcherOptions.Services.TryAddIntegrationEventBus<TIntegrationEventLogService>(dispatcherOptions.Assemblies, option =>
-        {
-            optionAction?.Invoke(option);
-        });
+        dispatcherOptions.Services.TryAddIntegrationEventBus<TIntegrationEventLogService>(dispatcherOptions.Assemblies,
+            option => optionAction?.Invoke(option));
         return dispatcherOptions;
     }
 }

src/Dispatcher/Masa.Contrib.Dispatcher.IntegrationEvents/IntegrationEventBus.cs

@@ -8,14 +8,14 @@ public class IntegrationEventBus : IIntegrationEventBus
     private readonly DispatcherOptions _dispatcherOptions;
     private readonly IPublisher _publisher;
     private readonly ILogger<IntegrationEventBus>? _logger;
-    private readonly IIntegrationEventLogService _eventLogService;
+    private readonly IIntegrationEventLogService? _eventLogService;
     private readonly IOptionsMonitor<AppConfig>? _appConfig;
     private readonly IEventBus? _eventBus;
     private readonly IUnitOfWork? _unitOfWork;
 
     public IntegrationEventBus(IOptions<DispatcherOptions> options,
         IPublisher publisher,
-        IIntegrationEventLogService eventLogService,
+        IIntegrationEventLogService? eventLogService = null,
         IOptionsMonitor<AppConfig>? appConfig = null,
         ILogger<IntegrationEventBus>? logger = null,
         IEventBus? eventBus = null,
@@ -59,7 +59,7 @@ private async Task PublishIntegrationAsync<TEvent>(TEvent @event)
             @event.UnitOfWork = _unitOfWork;
 
         var topicName = @event.Topic;
-        if (@event.UnitOfWork is { UseTransaction: true })
+        if (@event.UnitOfWork is { UseTransaction: true } && _eventLogService != null)
         {
             try
             {

src/Dispatcher/Masa.Contrib.Dispatcher.IntegrationEvents/Servers/DefaultHostedService.cs

@@ -16,7 +16,7 @@ public DefaultHostedService(IServiceProvider serviceProvider, IEnumerable<IProce
 
     public Task ExecuteAsync(CancellationToken stoppingToken)
     {
-        if (_serviceProvider.GetService<IUnitOfWorkManager>() == null)
+        if (_serviceProvider.GetService<IUnitOfWorkManager>() == null || _serviceProvider.GetService<IIntegrationEventLogService>() == null)
             return Task.CompletedTask;
 
         var processorTasks = _processors.Select(processor => new InfiniteLoopProcessor(_serviceProvider, processor))

src/Dispatcher/Masa.Contrib.Dispatcher.IntegrationEvents/ServiceCollectionExtensions.cs

@@ -5,6 +5,17 @@ namespace Microsoft.Extensions.DependencyInjection;
 
 public static class ServiceCollectionExtensions
 {
+    public static IServiceCollection AddIntegrationEventBus(
+        this IServiceCollection services,
+        Action<DispatcherOptions>? options = null)
+        => services.AddIntegrationEventBus(AppDomain.CurrentDomain.GetAssemblies(), options);
+
+    public static IServiceCollection AddIntegrationEventBus(
+        this IServiceCollection services,
+        Assembly[] assemblies,
+        Action<DispatcherOptions>? options = null)
+        => services.TryAddIntegrationEventBus(assemblies, options);
+
     public static IServiceCollection AddIntegrationEventBus<TIntegrationEventLogService>(
         this IServiceCollection services,
         Action<DispatcherOptions>? options = null)
@@ -23,6 +34,18 @@ internal static IServiceCollection TryAddIntegrationEventBus<TIntegrationEventLo
         Assembly[] assemblies,
         Action<DispatcherOptions>? options)
         where TIntegrationEventLogService : class, IIntegrationEventLogService
+    {
+        return services.TryAddIntegrationEventBus(assemblies, options, () =>
+        {
+            services.AddScoped<IIntegrationEventLogService, TIntegrationEventLogService>();
+        });
+    }
+
+    internal static IServiceCollection TryAddIntegrationEventBus(
+        this IServiceCollection services,
+        Assembly[] assemblies,
+        Action<DispatcherOptions>? options,
+        Action? action = null)
     {
         if (services.Any(service => service.ImplementationType == typeof(IntegrationEventBusProvider)))
             return services;
@@ -37,7 +60,7 @@ internal static IServiceCollection TryAddIntegrationEventBus<TIntegrationEventLo
 
         LocalQueueProcessor.SetLogger(services);
         services.AddScoped<IIntegrationEventBus, IntegrationEventBus>();
-        services.AddScoped<IIntegrationEventLogService, TIntegrationEventLogService>();
+        action?.Invoke();
         services.AddSingleton<IProcessor, RetryByDataProcessor>();
         services.AddSingleton<IProcessor, RetryByLocalQueueProcessor>();
         services.AddSingleton<IProcessor, DeletePublishedExpireEventProcessor>();

test/Masa.Contrib.Dispatcher.Tests/Application/Events/AddGoodsIntegrationEvent.cs

@@ -0,0 +1,17 @@
+// Copyright (c) MASA Stack All rights reserved.
+// Licensed under the MIT License. See LICENSE.txt in the project root for license information.
+
+namespace Masa.Contrib.Dispatcher.Tests.Application.Events;
+
+public record AddGoodsIntegrationEvent : IntegrationEvent
+{
+    public Guid Id { get; set; }
+
+    public string Name { get; set; }
+
+    public int Count { get; set; }
+
+    public decimal Price { get; set; }
+
+    public override string Topic { get; set; } = nameof(AddGoodsIntegrationEvent);
+}

test/Masa.Contrib.Dispatcher.Tests/TestDispatcher.cs

@@ -63,6 +63,22 @@ await dbContext.Set<User>().AddAsync(new User()
 
         Assert.IsTrue(RecordMiddleware<UserAgeQuery>.Time == 1);
         Assert.IsTrue(RecordMiddleware<CheckUserQuery>.Time == 0);
+    }
+
+    [TestMethod]
+    public async Task TestIntegrationEventAndNotUseEventLogServiceReturnNoError()
+    {
+        var services = new ServiceCollection();
+        services.AddIntegrationEventBus(option => option.UseTestPub().UseEventBus());
+        var serviceProvider = services.BuildServiceProvider();
 
+        var integrationEventBus = serviceProvider.GetRequiredService<IIntegrationEventBus>();
+        await integrationEventBus.PublishAsync(new AddGoodsIntegrationEvent()
+        {
+            Name = "Apple",
+            Count = 1,
+            Id = Guid.NewGuid(),
+            Price = 9.9m,
+        });
     }
 }