From 3ac4522d98eed5d0f344dca2e5d5a2ec771ff246 Mon Sep 17 00:00:00 2001 From: zhenlei520 Date: Sun, 27 Mar 2022 15:26:09 +0800 Subject: [PATCH 1/8] refactor(Dispatcher): Adjust the usage of EventBus and IntegrationEventBus, and change the state of the local message table to be processed by sql statement --- src/BuildingBlocks/MASA.BuildingBlocks | 2 +- .../Masa.Contrib.Ddd.Domain/DomainEventBus.cs | 4 +- .../Options/DispatcherOptions.cs | 4 +- .../ServiceCollectionExtensions.cs | 2 +- src/Ddd/Masa.Contrib.Ddd.Domain/_Imports.cs | 1 + .../EventBusBuilder.cs | 12 +++- .../Options/DispatcherOptions.cs | 13 +---- .../Masa.Contrib.Dispatcher.Events/README.md | 2 +- .../README.zh-CN.md | 2 +- .../ServiceCollectionExtensions.cs | 29 +++++----- .../UseMiddlewareExtensions.cs | 17 ------ .../DispatcherOptionsExtensions.cs | 20 +++---- .../DispatcherOptionsExtensions.cs | 5 -- ...ntegrationEventLogModelCreatingProvider.cs | 6 +- .../IntegrationEventLogService.cs | 55 +++++++++++++------ .../Internal/EfCommon.cs | 45 +++++++++++++++ .../Internal/Options/Tables.cs | 17 ++++++ .../_Imports.cs | 5 ++ 18 files changed, 156 insertions(+), 85 deletions(-) delete mode 100644 src/Dispatcher/Masa.Contrib.Dispatcher.Events/UseMiddlewareExtensions.cs create mode 100644 src/Dispatcher/Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF/Internal/EfCommon.cs create mode 100644 src/Dispatcher/Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF/Internal/Options/Tables.cs diff --git a/src/BuildingBlocks/MASA.BuildingBlocks b/src/BuildingBlocks/MASA.BuildingBlocks index 11c54401c..21b03761f 160000 --- a/src/BuildingBlocks/MASA.BuildingBlocks +++ b/src/BuildingBlocks/MASA.BuildingBlocks @@ -1 +1 @@ -Subproject commit 11c54401ca15e0c65de3f853e8739162faa22af4 +Subproject commit 21b03761fdcc4549031cecbda1bfbfd41fa4df01 diff --git a/src/Ddd/Masa.Contrib.Ddd.Domain/DomainEventBus.cs b/src/Ddd/Masa.Contrib.Ddd.Domain/DomainEventBus.cs index e734bd45a..9c3dbbf2c 100644 --- a/src/Ddd/Masa.Contrib.Ddd.Domain/DomainEventBus.cs +++ b/src/Ddd/Masa.Contrib.Ddd.Domain/DomainEventBus.cs @@ -2,8 +2,8 @@ namespace Masa.Contrib.Ddd.Domain; public class DomainEventBus : IDomainEventBus { - protected readonly IEventBus _eventBus; - protected readonly IIntegrationEventBus _integrationEventBus; + private readonly IEventBus _eventBus; + private readonly IIntegrationEventBus _integrationEventBus; private readonly IUnitOfWork _unitOfWork; private readonly DispatcherOptions _options; diff --git a/src/Ddd/Masa.Contrib.Ddd.Domain/Options/DispatcherOptions.cs b/src/Ddd/Masa.Contrib.Ddd.Domain/Options/DispatcherOptions.cs index 5077339b0..cb8661c45 100644 --- a/src/Ddd/Masa.Contrib.Ddd.Domain/Options/DispatcherOptions.cs +++ b/src/Ddd/Masa.Contrib.Ddd.Domain/Options/DispatcherOptions.cs @@ -1,6 +1,6 @@ -namespace Masa.Contrib.Ddd.Domain; +namespace Masa.Contrib.Ddd.Domain.Options; -public class DispatcherOptions : IDispatcherOptions +public class DispatcherOptions : IDistributedDispatcherOptions { public IServiceCollection Services { get; } diff --git a/src/Ddd/Masa.Contrib.Ddd.Domain/ServiceCollectionExtensions.cs b/src/Ddd/Masa.Contrib.Ddd.Domain/ServiceCollectionExtensions.cs index 98fddaec0..b2e69f887 100644 --- a/src/Ddd/Masa.Contrib.Ddd.Domain/ServiceCollectionExtensions.cs +++ b/src/Ddd/Masa.Contrib.Ddd.Domain/ServiceCollectionExtensions.cs @@ -19,7 +19,7 @@ public static IServiceCollection AddDomainEventBus( var dispatcherOptions = new DispatcherOptions(services, assemblies); options?.Invoke(dispatcherOptions); - services.AddSingleton(typeof(IOptions), _ => Options.Create(dispatcherOptions)); + services.AddSingleton(typeof(IOptions), _ => Microsoft.Extensions.Options.Options.Create(dispatcherOptions)); if (services.All(service => service.ServiceType != typeof(IEventBus))) throw new Exception("Please add EventBus first."); diff --git a/src/Ddd/Masa.Contrib.Ddd.Domain/_Imports.cs b/src/Ddd/Masa.Contrib.Ddd.Domain/_Imports.cs index a312bb7bf..428eec67b 100644 --- a/src/Ddd/Masa.Contrib.Ddd.Domain/_Imports.cs +++ b/src/Ddd/Masa.Contrib.Ddd.Domain/_Imports.cs @@ -6,6 +6,7 @@ global using Masa.BuildingBlocks.Dispatcher.Events; global using Masa.BuildingBlocks.Dispatcher.IntegrationEvents; global using Masa.Contrib.Ddd.Domain.Internal; +global using Masa.Contrib.Ddd.Domain.Options; global using Microsoft.Extensions.DependencyInjection; global using Microsoft.Extensions.DependencyInjection.Extensions; global using Microsoft.Extensions.Options; diff --git a/src/Dispatcher/Masa.Contrib.Dispatcher.Events/EventBusBuilder.cs b/src/Dispatcher/Masa.Contrib.Dispatcher.Events/EventBusBuilder.cs index d722e4f24..4c04428f4 100644 --- a/src/Dispatcher/Masa.Contrib.Dispatcher.Events/EventBusBuilder.cs +++ b/src/Dispatcher/Masa.Contrib.Dispatcher.Events/EventBusBuilder.cs @@ -1,8 +1,18 @@ namespace Masa.Contrib.Dispatcher.Events; -public class EventBusBuilder +public class EventBusBuilder: IEventBusBuilder { public IServiceCollection Services { get; } public EventBusBuilder(IServiceCollection services) => Services = services; + + public IEventBusBuilder UseMiddleware(Type middleware, ServiceLifetime middlewareLifetime = ServiceLifetime.Transient) + { + if (!typeof(IMiddleware<>).IsGenericInterfaceAssignableFrom(middleware)) + throw new ArgumentException($"{middleware.Name} doesn't implement IMiddleware<>"); + + var descriptor = new ServiceDescriptor(typeof(IMiddleware<>), middleware, middlewareLifetime); + Services.TryAddEnumerable(descriptor); + return this; + } } diff --git a/src/Dispatcher/Masa.Contrib.Dispatcher.Events/Options/DispatcherOptions.cs b/src/Dispatcher/Masa.Contrib.Dispatcher.Events/Options/DispatcherOptions.cs index bce2dbf61..beff7f071 100644 --- a/src/Dispatcher/Masa.Contrib.Dispatcher.Events/Options/DispatcherOptions.cs +++ b/src/Dispatcher/Masa.Contrib.Dispatcher.Events/Options/DispatcherOptions.cs @@ -1,10 +1,10 @@ namespace Masa.Contrib.Dispatcher.Events.Options; -public class DispatcherOptions : IDispatcherOptions +public class DispatcherOptions { - public IServiceCollection Services { get; } + private IServiceCollection Services { get; } - public Assembly[] Assemblies { get; } + private Assembly[] Assemblies { get; } private bool IsSupportUnitOfWork(Type eventType) => typeof(ITransaction).IsAssignableFrom(eventType) && !typeof(IDomainQuery<>).IsGenericInterfaceAssignableFrom(eventType); @@ -28,11 +28,4 @@ public DispatcherOptions(IServiceCollection services, Assembly[] assemblies) .ToList(); UnitOfWorkRelation = AllEventTypes.ToDictionary(type => type, IsSupportUnitOfWork); } - - public DispatcherOptions UseMiddleware(Type middleware, ServiceLifetime middlewareLifetime = ServiceLifetime.Scoped) - { - var descriptor = new ServiceDescriptor(typeof(IMiddleware<>), middleware, middlewareLifetime); - Services.TryAddEnumerable(descriptor); - return this; - } } diff --git a/src/Dispatcher/Masa.Contrib.Dispatcher.Events/README.md b/src/Dispatcher/Masa.Contrib.Dispatcher.Events/README.md index f66c8625f..9291a3ca4 100644 --- a/src/Dispatcher/Masa.Contrib.Dispatcher.Events/README.md +++ b/src/Dispatcher/Masa.Contrib.Dispatcher.Events/README.md @@ -166,7 +166,7 @@ public class LoggingMiddleware ```C# -builder.Services.AddEventBus(options => options.UseMiddleware(typeof(ValidatorMiddleware<>))); +builder.Services.AddEventBus(eventBusBuilder => eventBusBuilder.UseMiddleware(typeof(ValidatorMiddleware<>))); ``` 4. Support Transaction diff --git a/src/Dispatcher/Masa.Contrib.Dispatcher.Events/README.zh-CN.md b/src/Dispatcher/Masa.Contrib.Dispatcher.Events/README.zh-CN.md index 1575b3422..8613be4d8 100644 --- a/src/Dispatcher/Masa.Contrib.Dispatcher.Events/README.zh-CN.md +++ b/src/Dispatcher/Masa.Contrib.Dispatcher.Events/README.zh-CN.md @@ -165,7 +165,7 @@ public class LoggingMiddleware 2. 启用自定义Middleware ```C# -builder.Services.AddEventBus(options => options.UseMiddleware(typeof(ValidatorMiddleware<>))); +builder.Services.AddEventBus(eventBusBuilder => eventBusBuilder.UseMiddleware(typeof(ValidatorMiddleware<>))); ``` 4. 支持Transaction diff --git a/src/Dispatcher/Masa.Contrib.Dispatcher.Events/ServiceCollectionExtensions.cs b/src/Dispatcher/Masa.Contrib.Dispatcher.Events/ServiceCollectionExtensions.cs index 595aafdd4..d1a189f17 100644 --- a/src/Dispatcher/Masa.Contrib.Dispatcher.Events/ServiceCollectionExtensions.cs +++ b/src/Dispatcher/Masa.Contrib.Dispatcher.Events/ServiceCollectionExtensions.cs @@ -4,33 +4,33 @@ public static class ServiceCollectionExtensions { public static IServiceCollection AddEventBus( this IServiceCollection services, - Action? options = null) - => services.AddEventBus(AppDomain.CurrentDomain.GetAssemblies(), options); + Action? eventBusBuilder = null) + => services.AddEventBus(AppDomain.CurrentDomain.GetAssemblies(), eventBusBuilder); public static IServiceCollection AddEventBus( this IServiceCollection services, Assembly[] assemblies, - Action? options = null) - => services.AddEventBus(assemblies, ServiceLifetime.Scoped, options); + Action? eventBusBuilder = null) + => services.AddEventBus(assemblies, ServiceLifetime.Scoped, eventBusBuilder); public static IServiceCollection AddEventBus( this IServiceCollection services, Assembly[] assemblies, ServiceLifetime lifetime, - Action? options = null) + Action? eventBusBuilder = null) { if (services.Any(service => service.ImplementationType == typeof(EventBusProvider))) return services; services.AddSingleton(); + eventBusBuilder?.Invoke(new EventBusBuilder(services)); + DispatcherOptions dispatcherOptions = new DispatcherOptions(services, assemblies); - options?.Invoke(dispatcherOptions); services.AddSingleton(typeof(IOptions), _ => Microsoft.Extensions.Options.Options.Create(dispatcherOptions)); - - services.AddSingleton(new SagaDispatcher(services, dispatcherOptions.Assemblies).Build(lifetime)); - services.AddSingleton(new Internal.Dispatch.Dispatcher(services, dispatcherOptions.Assemblies).Build(lifetime)); + services.AddSingleton(new SagaDispatcher(services, assemblies).Build(lifetime)); + services.AddSingleton(new Internal.Dispatch.Dispatcher(services, assemblies).Build(lifetime)); services.TryAdd(typeof(IExecutionStrategy), typeof(ExecutionStrategy), ServiceLifetime.Singleton); services.AddTransient(typeof(IMiddleware<>), typeof(TransactionMiddleware<>)); services.AddScoped(typeof(IEventBus), typeof(EventBus)); @@ -41,20 +41,20 @@ public static IServiceCollection AddTestEventBus( this IServiceCollection services, Assembly[] assemblies, ServiceLifetime lifetime, - Action? options = null) + Action? eventBusBuilder = null) { if (services.Any(service => service.ImplementationType == typeof(EventBusProvider))) return services; services.AddSingleton(); - DispatcherOptions dispatcherOptions = new DispatcherOptions(services, assemblies); - options?.Invoke(dispatcherOptions); + eventBusBuilder?.Invoke(new EventBusBuilder(services)); + DispatcherOptions dispatcherOptions = new DispatcherOptions(services, assemblies); services.AddSingleton(typeof(IOptions), serviceProvider => Microsoft.Extensions.Options.Options.Create(dispatcherOptions)); - services.AddSingleton(new SagaDispatcher(services, dispatcherOptions.Assemblies, true).Build(lifetime)); - services.AddSingleton(new Internal.Dispatch.Dispatcher(services, dispatcherOptions.Assemblies).Build(lifetime)); + services.AddSingleton(new SagaDispatcher(services, assemblies, true).Build(lifetime)); + services.AddSingleton(new Internal.Dispatch.Dispatcher(services, assemblies).Build(lifetime)); services.TryAdd(typeof(IExecutionStrategy), typeof(ExecutionStrategy), ServiceLifetime.Singleton); services.AddTransient(typeof(IMiddleware<>), typeof(TransactionMiddleware<>)); services.AddScoped(typeof(IEventBus), typeof(EventBus)); @@ -64,6 +64,5 @@ public static IServiceCollection AddTestEventBus( private class EventBusProvider { - } } diff --git a/src/Dispatcher/Masa.Contrib.Dispatcher.Events/UseMiddlewareExtensions.cs b/src/Dispatcher/Masa.Contrib.Dispatcher.Events/UseMiddlewareExtensions.cs deleted file mode 100644 index e68898374..000000000 --- a/src/Dispatcher/Masa.Contrib.Dispatcher.Events/UseMiddlewareExtensions.cs +++ /dev/null @@ -1,17 +0,0 @@ -namespace Masa.Contrib.Dispatcher.Events; - -public static class UseMiddlewareExtensions -{ - public static EventBusBuilder UseMiddleware( - this EventBusBuilder eventBusBuilder, - Type middleware, - ServiceLifetime middlewareLifetime = ServiceLifetime.Scoped) - { - if (!typeof(IMiddleware<>).IsGenericInterfaceAssignableFrom(middleware)) - throw new ArgumentException($"{middleware.Name} doesn't implement IMiddleware<>"); - - var descriptor = new ServiceDescriptor(typeof(IMiddleware<>), middleware, middlewareLifetime); - eventBusBuilder.Services.TryAddEnumerable(descriptor); - return eventBusBuilder; - } -} diff --git a/src/Dispatcher/Masa.Contrib.Dispatcher.IntegrationEvents.Dapr/DispatcherOptionsExtensions.cs b/src/Dispatcher/Masa.Contrib.Dispatcher.IntegrationEvents.Dapr/DispatcherOptionsExtensions.cs index a559fbd49..6a8506ed8 100644 --- a/src/Dispatcher/Masa.Contrib.Dispatcher.IntegrationEvents.Dapr/DispatcherOptionsExtensions.cs +++ b/src/Dispatcher/Masa.Contrib.Dispatcher.IntegrationEvents.Dapr/DispatcherOptionsExtensions.cs @@ -4,26 +4,26 @@ public static class DispatcherOptionsExtensions { private const string DAPR_PUBSUB_NAME = "pubsub"; - public static IDispatcherOptions UseDaprEventBus( - this IDispatcherOptions options, - string daprPubsubName) + public static IDistributedDispatcherOptions UseDaprEventBus( + this IDistributedDispatcherOptions options, + string daprPubSubName) where TIntegrationEventLogService : class, IIntegrationEventLogService - => options.UseDaprEventBus(daprPubsubName, null); + => options.UseDaprEventBus(daprPubSubName, null); - public static IDispatcherOptions UseDaprEventBus( - this IDispatcherOptions options, - string daprPubsubName, + public static IDistributedDispatcherOptions UseDaprEventBus( + this IDistributedDispatcherOptions options, + string daprPubSubName, Action? builder) where TIntegrationEventLogService : class, IIntegrationEventLogService { return options.UseDaprEventBus(builder, dispatcherOptions => { - dispatcherOptions.PubSubName = daprPubsubName; + dispatcherOptions.PubSubName = daprPubSubName; }); } - public static IDispatcherOptions UseDaprEventBus( - this IDispatcherOptions options, + public static IDistributedDispatcherOptions UseDaprEventBus( + this IDistributedDispatcherOptions options, Action? builder = null, Action? action = null) where TIntegrationEventLogService : class, IIntegrationEventLogService diff --git a/src/Dispatcher/Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF/DispatcherOptionsExtensions.cs b/src/Dispatcher/Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF/DispatcherOptionsExtensions.cs index e7d443ff7..882893904 100644 --- a/src/Dispatcher/Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF/DispatcherOptionsExtensions.cs +++ b/src/Dispatcher/Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF/DispatcherOptionsExtensions.cs @@ -1,8 +1,3 @@ -using Masa.BuildingBlocks.Dispatcher.Events; -using Masa.Utils.Data.EntityFrameworkCore; -using Microsoft.Extensions.DependencyInjection; -using Microsoft.Extensions.DependencyInjection.Extensions; - namespace Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF; public static class DispatcherOptionsExtensions diff --git a/src/Dispatcher/Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF/IntegrationEventLogModelCreatingProvider.cs b/src/Dispatcher/Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF/IntegrationEventLogModelCreatingProvider.cs index 3268ff542..df39b1943 100644 --- a/src/Dispatcher/Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF/IntegrationEventLogModelCreatingProvider.cs +++ b/src/Dispatcher/Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF/IntegrationEventLogModelCreatingProvider.cs @@ -1,4 +1,4 @@ -namespace Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF; +namespace Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF; public class IntegrationEventLogModelCreatingProvider : IModelCreatingProvider { @@ -32,12 +32,14 @@ private void ConfigureEventLogEntry(EntityTypeBuilder build .IsRequired(); builder.Property(e => e.RowVersion) - .IsRowVersion(); + .HasMaxLength(36) + .IsRequired(); builder.Property(e => e.EventTypeName) .IsRequired(); builder.HasIndex(e => new { e.State, e.ModificationTime }, "index_state_modificationtime"); builder.HasIndex(e => new { e.State, e.TimesSent, e.ModificationTime }, "index_state_timessent_modificationtime"); + builder.HasIndex(e => new { e.EventId, e.RowVersion }, "index_eventid_version"); } } diff --git a/src/Dispatcher/Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF/IntegrationEventLogService.cs b/src/Dispatcher/Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF/IntegrationEventLogService.cs index f8785766c..76f111fc5 100644 --- a/src/Dispatcher/Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF/IntegrationEventLogService.cs +++ b/src/Dispatcher/Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF/IntegrationEventLogService.cs @@ -115,44 +115,65 @@ public async Task DeleteExpiresAsync(DateTime expiresAt, int batchCount = 1000, if (_eventLogContext.DbContext.ChangeTracker.QueryTrackingBehavior != QueryTrackingBehavior.TrackAll) { foreach (var log in eventLogs) - { _eventLogContext.DbContext.Entry(log).State = EntityState.Detached; - } } } private async Task UpdateEventStatus(Guid eventId, IntegrationEventStates status, Action? action = null) { - var eventLogEntry = _eventLogContext.EventLogs.FirstOrDefault(e => e.EventId == eventId); + var eventLogEntry = _eventLogContext.EventLogs.AsNoTracking().FirstOrDefault(e => e.EventId == eventId); if (eventLogEntry == null) throw new ArgumentException(nameof(eventId)); action?.Invoke(eventLogEntry); - eventLogEntry.State = status; eventLogEntry.ModificationTime = eventLogEntry.GetCurrentTime(); + if (status == IntegrationEventStates.InProgress) eventLogEntry.TimesSent++; - try - { - _eventLogContext.EventLogs.Update(eventLogEntry); - await _eventLogContext.DbContext.SaveChangesAsync(); - } - catch (DbUpdateConcurrencyException ex) - { - throw new UserFriendlyException(ex.Message); - } - - CheckAndDetached(eventLogEntry); + await UpdateAsync(eventLogEntry); } private void CheckAndDetached(IntegrationEventLog integrationEvent) { if (_eventLogContext.DbContext.ChangeTracker.QueryTrackingBehavior != QueryTrackingBehavior.TrackAll) - { _eventLogContext.DbContext.Entry(integrationEvent).State = EntityState.Detached; - } + } + + /// + /// By using SQL statements to handle high concurrency, reduce the dependence on the database + /// Turning on a transactional operation makes the operation atomic + /// Retrying the task operation in the background will not open the transaction, so that the task will only be executed once in high concurrency scenarios + /// + /// + private async Task UpdateAsync(IntegrationEventLog eventLogEntry) + { + string eventIdColumn = _eventLogContext.DbContext.GetPropertyName(nameof(IntegrationEventLog.EventId)); + string stateColumn = _eventLogContext.DbContext.GetPropertyName(nameof(IntegrationEventLog.State)); + string modificationTimeColumn = _eventLogContext.DbContext.GetPropertyName(nameof(IntegrationEventLog.ModificationTime)); + string timesSentColumn = _eventLogContext.DbContext.GetPropertyName(nameof(IntegrationEventLog.TimesSent)); + string rowVersionColumn = _eventLogContext.DbContext.GetPropertyName(nameof(IntegrationEventLog.RowVersion)); + string tableName = _eventLogContext.DbContext.GetTableName(); + var newVersion = Guid.NewGuid().ToString(); + + string updateSql = $"UPDATE { tableName } set [{ stateColumn }] = {{0}}, [{modificationTimeColumn }] = {{1}}, [{ timesSentColumn }] = {{2}}, [{ rowVersionColumn }] = {{3}} where [{ eventIdColumn }] = {{4}} and [{ rowVersionColumn }] = {{5}};"; + await ExecuteAsync(updateSql, new object[] + { + (int)eventLogEntry.State, + eventLogEntry.ModificationTime, + eventLogEntry.TimesSent, + newVersion, + eventLogEntry.EventId, + eventLogEntry.RowVersion + }); + } + + private async Task ExecuteAsync(string updateSql, params object[] parameters) + { + var effectRow = await _eventLogContext.DbContext.Database.ExecuteSqlRawAsync(updateSql, parameters); + if (effectRow == 0) + throw new UserFriendlyException("Concurrency conflict, update exception"); } } diff --git a/src/Dispatcher/Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF/Internal/EfCommon.cs b/src/Dispatcher/Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF/Internal/EfCommon.cs new file mode 100644 index 000000000..0d3467609 --- /dev/null +++ b/src/Dispatcher/Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF/Internal/EfCommon.cs @@ -0,0 +1,45 @@ +namespace Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF.Internal; + +internal static class EfCommon +{ + private static readonly ConcurrentDictionary TableDictionary = new(); + + public static string GetTableName(this DbContext dbContext) where TEntity : class + { + var type = typeof(TEntity); + if (TableDictionary.TryGetValue(type, out Tables? table)) + return table.Name; + + table = dbContext.GetTables(); + TableDictionary.TryAdd(type, table); + return table.Name; + } + + public static string GetPropertyName(this DbContext dbContext, string propertyName) where TEntity : class + { + var type = typeof(TEntity); + if (TableDictionary.TryGetValue(type, out Tables? table)) + return table.GetPropertyName(propertyName); + + table = dbContext.GetTables(); + TableDictionary.TryAdd(type, table); + return table.GetPropertyName(propertyName); + } + + private static string GetPropertyName(this Tables table, string propertyName) + => table.Properties.Where(p => p.Name == propertyName).Select(p => p.ColunName).FirstOrDefault()!; + + private static Tables GetTables(this DbContext dbContext) where TEntity : class + { + var entityType = GetEntityType(dbContext); + var tableName = entityType.GetTableName()!; + var schema = entityType.GetSchema()!; + List<(string Name, string ColunName)> properties = new List<(string Name, string ColunName)>(); + foreach (var property in entityType.GetProperties()) + properties.Add(new(property.Name, property.GetColumnName(StoreObjectIdentifier.Table(tableName, schema))!)); + return new Tables(tableName, schema, properties); + } + + private static IEntityType GetEntityType(this DbContext dbContext) where TEntity : class + => dbContext.Model.FindEntityType(typeof(TEntity))!; +} diff --git a/src/Dispatcher/Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF/Internal/Options/Tables.cs b/src/Dispatcher/Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF/Internal/Options/Tables.cs new file mode 100644 index 000000000..c38e12b89 --- /dev/null +++ b/src/Dispatcher/Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF/Internal/Options/Tables.cs @@ -0,0 +1,17 @@ +namespace Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF.Internal.Options; + +internal class Tables +{ + public string Name { get; } + + public string Schema { get; } + + public List<(string Name, string ColunName)> Properties { get; } + + public Tables(string name, string schema, List<(string Name, string ColunName)> properties) + { + Name = name; + Schema = schema; + Properties = properties; + } +} diff --git a/src/Dispatcher/Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF/_Imports.cs b/src/Dispatcher/Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF/_Imports.cs index d07767af4..b930aadc2 100644 --- a/src/Dispatcher/Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF/_Imports.cs +++ b/src/Dispatcher/Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF/_Imports.cs @@ -1,12 +1,17 @@ +global using Masa.BuildingBlocks.Dispatcher.Events; global using Masa.BuildingBlocks.Dispatcher.IntegrationEvents; global using Masa.BuildingBlocks.Dispatcher.IntegrationEvents.Logs; +global using Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF.Internal; +global using Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF.Internal.Options; global using Masa.Utils.Data.EntityFrameworkCore; global using Microsoft.EntityFrameworkCore; +global using Microsoft.EntityFrameworkCore.Metadata; global using Microsoft.EntityFrameworkCore.Metadata.Builders; global using Microsoft.Extensions.DependencyInjection; global using Microsoft.Extensions.DependencyInjection.Extensions; global using Microsoft.Extensions.Logging; global using System; +global using System.Collections.Concurrent; global using System.Collections.Generic; global using System.Data.Common; global using System.Linq; From ecdcc579d75d3fa8bdc91b06411c737a9c1d88ae Mon Sep 17 00:00:00 2001 From: zhenlei520 Date: Sun, 27 Mar 2022 15:27:17 +0800 Subject: [PATCH 2/8] test(dispatcher): Adjust the dispatcher unit test to adapt to the new writing method --- .../DomainEventBusTest.cs | 12 +- .../DomainIntegrationEventBusTest.cs | 4 +- .../AssemblyResolutionTests.cs | 79 +++-- .../BackgroundServiceTest.cs | 4 +- .../DispatcherOptionTest.cs | 11 +- .../IntegrationEventBusTest.cs | 15 +- .../Infrastructure/CustomDbContext.cs | 1 - .../IntegrationEventLogContextTest.cs | 14 +- .../IntegrationEventLogServiceTest.cs | 275 +++++++++++++++--- ...ntegrationEvents.EventLogs.EF.Tests.csproj | 1 - .../TestBase.cs | 32 +- .../_Imports.cs | 1 + 12 files changed, 312 insertions(+), 137 deletions(-) diff --git a/test/Masa.Contrib.Ddd.Domain.Tests/DomainEventBusTest.cs b/test/Masa.Contrib.Ddd.Domain.Tests/DomainEventBusTest.cs index 11a93eb17..4b8434c2b 100644 --- a/test/Masa.Contrib.Ddd.Domain.Tests/DomainEventBusTest.cs +++ b/test/Masa.Contrib.Ddd.Domain.Tests/DomainEventBusTest.cs @@ -8,7 +8,7 @@ public class DomainEventBusTest private Mock _eventBus = default!; private Mock _integrationEventBus = default!; private Mock _uoW = default!; - private IOptions _dispatcherOptions = default!; + private IOptions _dispatcherOptions = default!; [TestInitialize] public void Initialize() @@ -18,7 +18,7 @@ public void Initialize() _eventBus = new(); _integrationEventBus = new(); _uoW = new(); - _dispatcherOptions = Options.Create(new DispatcherOptions(new ServiceCollection(), _defaultAssemblies)); + _dispatcherOptions = Microsoft.Extensions.Options.Options.Create(new Options.DispatcherOptions(new ServiceCollection(), _defaultAssemblies)); } [TestMethod] @@ -99,7 +99,7 @@ public void TestAddMultDomainEventBusAsync() _services.AddDomainEventBus(new[] { typeof(DomainEventBusTest).Assembly }).AddDomainEventBus(); var serviceProvider = _services.BuildServiceProvider(); Assert.IsTrue(serviceProvider.GetServices().Count() == 1); - Assert.IsTrue(serviceProvider.GetServices>().Count() == 1); + Assert.IsTrue(serviceProvider.GetServices>().Count() == 1); } [TestMethod] @@ -201,7 +201,7 @@ public async Task TestPublishQueueAsync() var uoW = new Mock(); uoW.Setup(u => u.CommitAsync(default)).Verifiable(); - var options = Options.Create(new DispatcherOptions(_services, AppDomain.CurrentDomain.GetAssemblies())); + var options = Microsoft.Extensions.Options.Options.Create(new Options.DispatcherOptions(_services, AppDomain.CurrentDomain.GetAssemblies())); var domainEventBus = new DomainEventBus(_eventBus.Object, _integrationEventBus.Object, uoW.Object, options); @@ -230,7 +230,7 @@ public async Task TestPublishDomainQueryAsync() var uoW = new Mock(); uoW.Setup(u => u.CommitAsync(default)).Verifiable(); - var options = Options.Create(new DispatcherOptions(services, AppDomain.CurrentDomain.GetAssemblies())); + var options = Microsoft.Extensions.Options.Options.Create(new Options.DispatcherOptions(services, AppDomain.CurrentDomain.GetAssemblies())); var domainEventBus = new DomainEventBus(eventBus.Object, integrationEventBus.Object, uoW.Object, options); @@ -246,7 +246,7 @@ public async Task TestCommitAsync() var services = new ServiceCollection(); _uoW.Setup(uow => uow.CommitAsync(CancellationToken.None)).Verifiable(); - Mock> options = new(); + Mock> options = new(); var domainEventBus = new DomainEventBus(_eventBus.Object, _integrationEventBus.Object, _uoW.Object, options.Object); await domainEventBus.CommitAsync(CancellationToken.None); diff --git a/test/Masa.Contrib.Ddd.Domain.Tests/DomainIntegrationEventBusTest.cs b/test/Masa.Contrib.Ddd.Domain.Tests/DomainIntegrationEventBusTest.cs index ba0fba6a4..8774ad705 100644 --- a/test/Masa.Contrib.Ddd.Domain.Tests/DomainIntegrationEventBusTest.cs +++ b/test/Masa.Contrib.Ddd.Domain.Tests/DomainIntegrationEventBusTest.cs @@ -7,7 +7,7 @@ public class DomainIntegrationEventBus private IServiceCollection _services = default!; private Mock _integrationEventBus = default!; private Mock _uoW = default!; - private IOptions _dispatcherOptions = default!; + private IOptions _dispatcherOptions = default!; [TestInitialize] public void Initialize() @@ -17,7 +17,7 @@ public void Initialize() _integrationEventBus = new(); _integrationEventBus.Setup(eventBus => eventBus.PublishAsync(It.IsAny())).Verifiable(); _uoW = new(); - _dispatcherOptions = Options.Create(new DispatcherOptions(new ServiceCollection(), _defaultAssemblies)); + _dispatcherOptions = Microsoft.Extensions.Options.Options.Create(new Options.DispatcherOptions(new ServiceCollection(), _defaultAssemblies)); } [TestMethod] diff --git a/test/Masa.Contrib.Dispatcher.Events.Tests/AssemblyResolutionTests.cs b/test/Masa.Contrib.Dispatcher.Events.Tests/AssemblyResolutionTests.cs index 0586c0114..a5aafb380 100644 --- a/test/Masa.Contrib.Dispatcher.Events.Tests/AssemblyResolutionTests.cs +++ b/test/Masa.Contrib.Dispatcher.Events.Tests/AssemblyResolutionTests.cs @@ -7,30 +7,20 @@ public class AssemblyResolutionTests public void TestResolveEventBus() { var services = new ServiceCollection(); - services.AddLogging(loggingBuilder => loggingBuilder.AddConsole()); - services.AddTransient(typeof(IMiddleware<>), typeof(LoggingMiddleware<>)); - services.AddEventBus(); + services + .AddEventBus(eventBusBuilder => eventBusBuilder.UseMiddleware(typeof(LoggingMiddleware<>))) + .AddLogging(loggingBuilder => loggingBuilder.AddConsole()); var serviceProvider = services.BuildServiceProvider(); var eventBus = serviceProvider.GetService(); Assert.IsNotNull(eventBus, "Event bus injection failed"); Assert.IsNotNull(eventBus.GetAllEventTypes()); } - [TestMethod] - public void TestAddDefaultAssembly() - { - var services = new ServiceCollection(); - services.AddLogging(loggingBuilder => loggingBuilder.AddConsole()); - services.AddTransient(typeof(IMiddleware<>), typeof(LoggingMiddleware<>)); - services.AddTestEventBus(AppDomain.CurrentDomain.GetAssemblies(), ServiceLifetime.Scoped); - } - [TestMethod] public void TestAddNullAssembly() { var services = new ServiceCollection(); services.AddLogging(loggingBuilder => loggingBuilder.AddConsole()); - services.AddTransient(typeof(IMiddleware<>), typeof(LoggingMiddleware<>)); Assert.ThrowsException(() => { Assembly[] assemblies = null; @@ -43,7 +33,6 @@ public void TestAddEmptyAssembly() { var services = new ServiceCollection(); services.AddLogging(loggingBuilder => loggingBuilder.AddConsole()); - services.AddTransient(typeof(IMiddleware<>), typeof(LoggingMiddleware<>)); Assert.ThrowsException(() => { services.AddEventBus(Array.Empty()); @@ -55,43 +44,46 @@ public void TestEventBusByAddNullAssembly() { var services = new ServiceCollection(); services.AddLogging(loggingBuilder => loggingBuilder.AddConsole()); - services.AddTransient(typeof(IMiddleware<>), typeof(LoggingMiddleware<>)); Assert.ThrowsException(() => { services.AddTestEventBus(null, ServiceLifetime.Scoped); }); } - [TestMethod] - public void TestEventBusByAddEmptyAssembly() - { - var services = new ServiceCollection(); - services.AddLogging(loggingBuilder => loggingBuilder.AddConsole()); - services.AddTransient(typeof(IMiddleware<>), typeof(LoggingMiddleware<>)); - Assert.ThrowsException(() => - { - services.AddTestEventBus(Array.Empty(), ServiceLifetime.Scoped); - }); - } + // [TestMethod] + // public void TestEventBusByAddEmptyAssembly() + // { + // var services = new ServiceCollection(); + // services.AddLogging(loggingBuilder => loggingBuilder.AddConsole()); + // services.AddTransient(typeof(IMiddleware<>), typeof(LoggingMiddleware<>)); + // Assert.ThrowsException(() => + // { + // services.AddTestEventBus(Array.Empty(), ServiceLifetime.Scoped); + // }); + // } [TestMethod] public void TestEventBus() { var services = new ServiceCollection(); services.AddLogging(loggingBuilder => loggingBuilder.AddConsole()); - services.AddTransient(typeof(IMiddleware<>), typeof(LoggingMiddleware<>)); - services.AddTestEventBus(AppDomain.CurrentDomain.GetAssemblies(), ServiceLifetime.Scoped); + services.AddTestEventBus(AppDomain.CurrentDomain.GetAssemblies(), ServiceLifetime.Scoped,eventBusBuilder => eventBusBuilder.UseMiddleware(typeof(LoggingMiddleware<>))); + var serviceProvider = services.BuildServiceProvider(); + var eventBus = serviceProvider.GetService(); + Assert.IsNotNull(eventBus, "Event bus injection failed"); + Assert.IsNotNull(eventBus.GetAllEventTypes()); } [TestMethod] public void TestUseEventBus() { + var assemblies = AppDomain.CurrentDomain.GetAssemblies(); var services = new ServiceCollection(); services.AddLogging(loggingBuilder => loggingBuilder.AddConsole()); - services.AddTransient(typeof(IMiddleware<>), typeof(LoggingMiddleware<>)); - var options = new DispatcherOptions(services, AppDomain.CurrentDomain.GetAssemblies()); - options.UseEventBus(); - + Mock dispatcherOptions = new(); + dispatcherOptions.Setup(option => option.Assemblies).Returns(assemblies).Verifiable(); + dispatcherOptions.Setup(option => option.Services).Returns(services).Verifiable(); + dispatcherOptions.Object.UseEventBus(eventBuilder => eventBuilder.UseMiddleware(typeof(LoggingMiddleware<>))); var eventBus = services.BuildServiceProvider().GetService(); Assert.IsNotNull(eventBus); } @@ -102,23 +94,24 @@ public void TestAddMultEventBus() var assemblies = AppDomain.CurrentDomain.GetAssemblies(); var services = new ServiceCollection(); services.AddLogging(loggingBuilder => loggingBuilder.AddConsole()); - var options = new DispatcherOptions(services, assemblies); - options.UseEventBus().UseEventBus(); + Mock dispatcherOptions = new(); + dispatcherOptions.Setup(option => option.Assemblies).Returns(assemblies).Verifiable(); + dispatcherOptions.Setup(option => option.Services).Returns(services).Verifiable(); + dispatcherOptions.Object + .UseEventBus() + .UseEventBus(); Assert.IsTrue(services.BuildServiceProvider().GetServices().Count() == 1); - - var services2 = new ServiceCollection(); - services.AddLogging(loggingBuilder => loggingBuilder.AddConsole()); - services2.AddTestEventBus(assemblies, ServiceLifetime.Scoped) - .AddTestEventBus(assemblies, ServiceLifetime.Scoped); - var serviceProvider = services.BuildServiceProvider(); - Assert.IsTrue(serviceProvider.GetServices().Count() == 1); } [TestMethod] public void TestUseEventBusAndNullServices() { - var options = new DispatcherOptions(null!, AppDomain.CurrentDomain.GetAssemblies()); - Assert.ThrowsException(() => options.UseEventBus()); + var assemblies = AppDomain.CurrentDomain.GetAssemblies(); + IServiceCollection services = null!; + Mock dispatcherOptions = new(); + dispatcherOptions.Setup(option => option.Assemblies).Returns(assemblies).Verifiable(); + dispatcherOptions.Setup(option => option.Services).Returns(services).Verifiable(); + Assert.ThrowsException(() => dispatcherOptions.Object.UseEventBus()); } } diff --git a/test/Masa.Contrib.Dispatcher.IntegrationEvents.Dapr.Tests/BackgroundServiceTest.cs b/test/Masa.Contrib.Dispatcher.IntegrationEvents.Dapr.Tests/BackgroundServiceTest.cs index adb4b69c1..86b8d7edf 100644 --- a/test/Masa.Contrib.Dispatcher.IntegrationEvents.Dapr.Tests/BackgroundServiceTest.cs +++ b/test/Masa.Contrib.Dispatcher.IntegrationEvents.Dapr.Tests/BackgroundServiceTest.cs @@ -1,6 +1,4 @@ -using Microsoft.Extensions.Logging.Abstractions; - -namespace Masa.Contrib.Dispatcher.IntegrationEvents.Dapr.Tests; +namespace Masa.Contrib.Dispatcher.IntegrationEvents.Dapr.Tests; [TestClass] public class BackgroundServiceTest diff --git a/test/Masa.Contrib.Dispatcher.IntegrationEvents.Dapr.Tests/DispatcherOptionTest.cs b/test/Masa.Contrib.Dispatcher.IntegrationEvents.Dapr.Tests/DispatcherOptionTest.cs index ec4399909..6b55475c7 100644 --- a/test/Masa.Contrib.Dispatcher.IntegrationEvents.Dapr.Tests/DispatcherOptionTest.cs +++ b/test/Masa.Contrib.Dispatcher.IntegrationEvents.Dapr.Tests/DispatcherOptionTest.cs @@ -137,10 +137,17 @@ public void TestGetCurrentTime() [TestMethod] public void UseDaprEventBus() { - _options.UseDaprEventBus("pubsub2"); - var serviceProvider = _options.Services.BuildServiceProvider(); + Assembly[] assemblies = AppDomain.CurrentDomain.GetAssemblies(); + var services = new ServiceCollection(); + Mock distributedDispatcherOptions = new(); + distributedDispatcherOptions.Setup(option => option.Services).Returns(services).Verifiable(); + distributedDispatcherOptions.Setup(option => option.Assemblies).Returns(assemblies).Verifiable(); + distributedDispatcherOptions.Object.UseDaprEventBus("pubsub2"); + var serviceProvider = services.BuildServiceProvider(); + var integrationEventBus = serviceProvider.GetService(); Assert.IsNotNull(integrationEventBus); + var options = serviceProvider.GetRequiredService>(); Assert.IsTrue(options.Value.PubSubName == "pubsub2"); } diff --git a/test/Masa.Contrib.Dispatcher.IntegrationEvents.Dapr.Tests/IntegrationEventBusTest.cs b/test/Masa.Contrib.Dispatcher.IntegrationEvents.Dapr.Tests/IntegrationEventBusTest.cs index fb290dea6..0e19ec7d0 100644 --- a/test/Masa.Contrib.Dispatcher.IntegrationEvents.Dapr.Tests/IntegrationEventBusTest.cs +++ b/test/Masa.Contrib.Dispatcher.IntegrationEvents.Dapr.Tests/IntegrationEventBusTest.cs @@ -64,10 +64,14 @@ public void TestDispatcherOption() [TestMethod] public void TestAddMultDaprEventBus() { - _dispatcherOptions.Object.Value + var services = new ServiceCollection(); + Mock distributedDispatcherOptions = new(); + distributedDispatcherOptions.Setup(option => option.Services).Returns(services).Verifiable(); + distributedDispatcherOptions.Setup(option => option.Assemblies).Returns(AppDomain.CurrentDomain.GetAssemblies()).Verifiable(); + distributedDispatcherOptions.Object .UseDaprEventBus() .UseDaprEventBus(); - var serviceProvider = _dispatcherOptions.Object.Value.Services.BuildServiceProvider(); + var serviceProvider = services.BuildServiceProvider(); Assert.IsTrue(serviceProvider.GetServices().Count() == 1); } @@ -113,9 +117,12 @@ public void TestUseLogger() [TestMethod] public void TestAddDaprEventBusAndNullServicesAsync() { - _options.Setup(option => option.Services).Returns(() => null!); + IServiceCollection services = null; + Mock distributedDispatcherOptions = new(); + distributedDispatcherOptions.Setup(option => option.Services).Returns(services).Verifiable(); + distributedDispatcherOptions.Setup(option => option.Assemblies).Returns(AppDomain.CurrentDomain.GetAssemblies()).Verifiable(); Assert.ThrowsException(() => - _options.Object.UseDaprEventBus(), + distributedDispatcherOptions.Object.UseDaprEventBus(), $"Value cannot be null. (Parameter '{nameof(_options.Object.Services)}')"); } diff --git a/test/Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF.Tests/Infrastructure/CustomDbContext.cs b/test/Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF.Tests/Infrastructure/CustomDbContext.cs index 096c1c55c..e3652d174 100644 --- a/test/Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF.Tests/Infrastructure/CustomDbContext.cs +++ b/test/Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF.Tests/Infrastructure/CustomDbContext.cs @@ -6,6 +6,5 @@ internal class CustomDbContext : MasaDbContext public CustomDbContext(MasaDbContextOptions options) : base(options) { - } } diff --git a/test/Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF.Tests/IntegrationEventLogContextTest.cs b/test/Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF.Tests/IntegrationEventLogContextTest.cs index e96cf62a1..9b182ff14 100644 --- a/test/Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF.Tests/IntegrationEventLogContextTest.cs +++ b/test/Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF.Tests/IntegrationEventLogContextTest.cs @@ -6,7 +6,11 @@ public class IntegrationEventLogContextTest : TestBase [TestMethod] public void TestCreateDbContext() { - var serviceProvider = CreateDefaultProvider(option => option.UseEventLog()); + var services = new ServiceCollection(); + services.AddMasaDbContext(builder => builder.UseSqlite(ConnectionString)); + var distributedDispatcherOptions = CreateDispatcherOptions(services); + distributedDispatcherOptions.UseEventLog(); + var serviceProvider = services.BuildServiceProvider(); var customDbContext = serviceProvider.GetRequiredService(); var entity = customDbContext.Model.GetEntityTypes() @@ -42,10 +46,10 @@ public void TestAddDbContext() [TestMethod] public void TestUseEventLog() { - var dispatcherOptions = new DispatcherOptions(new ServiceCollection()); - dispatcherOptions.Services.AddDbContext(options => options.UseSqlite(Connection)); - dispatcherOptions.UseEventLog(); - var serviceProvider = dispatcherOptions.Services.BuildServiceProvider(); + var distributedDispatcherOptions = CreateDispatcherOptions(new ServiceCollection()); + distributedDispatcherOptions.Services.AddDbContext(options => options.UseSqlite(Connection)); + distributedDispatcherOptions.UseEventLog(); + var serviceProvider = distributedDispatcherOptions.Services.BuildServiceProvider(); Assert.ThrowsException(() => serviceProvider.GetService()); diff --git a/test/Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF.Tests/IntegrationEventLogServiceTest.cs b/test/Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF.Tests/IntegrationEventLogServiceTest.cs index e6a2d629d..333021e35 100644 --- a/test/Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF.Tests/IntegrationEventLogServiceTest.cs +++ b/test/Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF.Tests/IntegrationEventLogServiceTest.cs @@ -6,101 +6,288 @@ public class IntegrationEventLogServiceTest : TestBase [TestMethod] public async Task TestNullDbTransactionAsync() { + var services = new ServiceCollection(); + services.AddMasaDbContext(builder => builder.UseSqlite(ConnectionString)) + .AddScoped(); + IDispatcherOptions dispatcherOptions = CreateDispatcherOptions(services); + dispatcherOptions.UseEventLog(); + var serviceProvider = services.BuildServiceProvider(); + DbTransaction transaction = null!; var @event = new OrderPaymentSucceededIntegrationEvent() { OrderId = "1234567890123", PaymentTime = (long)(DateTime.UtcNow - new DateTime(1970, 1, 1, 0, 0, 0, DateTimeKind.Utc)).TotalSeconds }; - var serviceProvider = CreateDefaultProvider(option => option.UseEventLog()); + var eventLogService = serviceProvider.GetRequiredService(); await Assert.ThrowsExceptionAsync(async () => await eventLogService.SaveEventAsync(@event, transaction)); } [TestMethod] - public void TestMultUseEventLogService() + public void TestNullServices() { - var serviceProvider = CreateDefaultProvider(options => + var dispatcherOptions = CreateDispatcherOptions(null); + + Assert.ThrowsException(() => { - options.UseEventLog(); + dispatcherOptions.UseEventLog(); }); - Assert.IsTrue(serviceProvider.GetServices().Count() == 1); } [TestMethod] - public void TestNullServices() + public void TestAddMultEventLog() { - var options = new DispatcherOptions(null!); - Assert.ThrowsException(() => + var services = new ServiceCollection(); + IDispatcherOptions dispatcherOptions = CreateDispatcherOptions(services); + dispatcherOptions.UseEventLog().UseEventLog(); + Assert.IsTrue(services.Count(service => service.ImplementationType == typeof(IntegrationEventLogModelCreatingProvider)) == 1); + Assert.IsTrue(services.Count(service => service.ServiceType == typeof(IntegrationEventLogContext)) == 1); + } + + [TestMethod] + public async Task TestRetrieveEventLogsFailedToPublishAsync() + { + var dispatcherOptions = CreateDispatcherOptions(new ServiceCollection()); + dispatcherOptions.UseEventLog(); + dispatcherOptions.Services.AddMasaDbContext(option => option.DbContextOptionsBuilder.UseSqlite(Connection)); + dispatcherOptions.Services.AddScoped(); + var serviceProvider = dispatcherOptions.Services.BuildServiceProvider(); + await serviceProvider.GetRequiredService().Database.EnsureCreatedAsync(); + var logService = serviceProvider.GetRequiredService(); + var list = await logService.RetrieveEventLogsFailedToPublishAsync(); + Assert.IsTrue(!list.Any()); + } + + [TestMethod] + public async Task TestRetrieveEventLogsFailedToPublish2Async() + { + var response = await InitializeAsync(); + + #region Initialization data + + var logs = await response.CustomDbContext.Set().ToListAsync(); + response.CustomDbContext.Set().RemoveRange(logs); + + var @event = new OrderPaymentSucceededIntegrationEvent + { + OrderId = "1234567890123", + PaymentTime = (long)(DateTime.UtcNow - new DateTime(1970, 1, 1, 0, 0, 0, DateTimeKind.Utc)).TotalSeconds + }; + await response.CustomDbContext.Set().AddAsync(new IntegrationEventLog(@event, Guid.NewGuid()) { - options.UseEventLog(); + State = IntegrationEventStates.InProgress, + ModificationTime = DateTime.UtcNow.AddSeconds(-120), }); + await response.CustomDbContext.SaveChangesAsync(); + + #endregion + + var logService = response.ServiceProvider.GetRequiredService(); + var list = await logService.RetrieveEventLogsFailedToPublishAsync(); + Assert.IsTrue(list.Count() == 1); + + var eventLog = list.Select(log => log.Event).FirstOrDefault(); + Assert.IsTrue(eventLog.Equals(@event)); } [TestMethod] - public void TestUseCustomDbContextByNullServices() + public async Task TestSaveEventAsync() { - var options = new DispatcherOptions(null!); - Assert.IsNull(options.Services); - Assert.ThrowsException(() => options.UseEventLog()); + var response = await InitializeAsync(); + + Assert.IsTrue(await response.CustomDbContext.Set().CountAsync() == 0); + + await using (var transcation = await response.CustomDbContext.Database.BeginTransactionAsync()) + { + var logService = response.ServiceProvider.GetRequiredService(); + var @event = new OrderPaymentSucceededIntegrationEvent + { + OrderId = "1234567890123", + PaymentTime = (long)(DateTime.UtcNow - new DateTime(1970, 1, 1, 0, 0, 0, DateTimeKind.Utc)).TotalSeconds + }; + await logService.SaveEventAsync(@event, transcation.GetDbTransaction()); + await transcation.CommitAsync(); + } + + Assert.IsTrue(await response.CustomDbContext.Set().CountAsync() == 1); + + Assert.IsTrue(await response.CustomDbContext.Set() + .CountAsync(log => log.State == IntegrationEventStates.NotPublished) == 1); } [TestMethod] - public async Task TestCustomDbContextAsync() + public async Task TestSaveEventByExceptionAsync() { - var options = new DispatcherOptions(new ServiceCollection()); - options.Services.AddMasaDbContext(optionsBuilder - => optionsBuilder.UseSqlite(ConnectionString).DbContextOptionsBuilder.UseQueryTrackingBehavior(QueryTrackingBehavior.NoTracking)); + var response = await InitializeAsync(); - var integrationEventBus = new Mock(); - integrationEventBus.Setup(e => e.GetAllEventTypes()).Returns(() => - AppDomain.CurrentDomain.GetAssemblies().SelectMany(assembly => assembly.GetTypes()) - .Where(type => typeof(IIntegrationEvent).IsAssignableFrom(type))); - options.Services.AddScoped(_ => integrationEventBus.Object); + Assert.IsTrue(await response.CustomDbContext.Set().CountAsync() == 0); - options.Services.AddScoped(); + await Assert.ThrowsExceptionAsync(async () => + { + await using var transcation = await response.CustomDbContext.Database.BeginTransactionAsync(); + var logService = response.ServiceProvider.GetRequiredService(); + var @event = new OrderPaymentSucceededIntegrationEvent + { + OrderId = "1234567890123", + PaymentTime = (long)(DateTime.UtcNow - new DateTime(1970, 1, 1, 0, 0, 0, DateTimeKind.Utc)).TotalSeconds + }; + await logService.SaveEventAsync(@event, transcation.GetDbTransaction()); + throw new Exception("custom exception"); + }, "custom exception"); - options.UseEventLog(); + Assert.IsTrue(await response.CustomDbContext.Set().CountAsync() == 0); + } - var serviceProvider = options.Services.BuildServiceProvider(); - var eventLogService = serviceProvider.GetRequiredService(); + [TestMethod] + public async Task TestMarkEventAsInProgressAsync() + { + var response = await InitializeAsync(); + Assert.IsTrue(await response.CustomDbContext.Set().CountAsync() == 0); - var @event = new OrderPaymentSucceededIntegrationEvent() + await using var transcation = await response.CustomDbContext.Database.BeginTransactionAsync(); + var logService = response.ServiceProvider.GetRequiredService(); + var @event = new OrderPaymentSucceededIntegrationEvent { OrderId = "1234567890123", PaymentTime = (long)(DateTime.UtcNow - new DateTime(1970, 1, 1, 0, 0, 0, DateTimeKind.Utc)).TotalSeconds }; + await logService.SaveEventAsync(@event, transcation.GetDbTransaction()); - var dbContext = serviceProvider.GetRequiredService(); - await dbContext.Database.EnsureCreatedAsync(); + await logService.MarkEventAsInProgressAsync(@event.Id); + Assert.IsTrue(await response.CustomDbContext.Set() + .CountAsync(log => log.State == IntegrationEventStates.InProgress) == 1); + await transcation.CommitAsync(); + + Assert.IsTrue(await response.CustomDbContext.Set() + .CountAsync(log => log.State == IntegrationEventStates.InProgress) == 1); + Assert.IsTrue(await response.CustomDbContext.Set().CountAsync() == 1); } [TestMethod] - public async Task TestAddMultEventLog() + public async Task TestMarkEventAsInProgress2Async() { - var options = new DispatcherOptions(new ServiceCollection()); - options.Services.AddMasaDbContext(optionsBuilder => optionsBuilder.UseSqlite(ConnectionString)); + var response = await InitializeAsync(); + Assert.IsTrue(await response.CustomDbContext.Set().CountAsync() == 0); - var integrationEventBus = new Mock(); - integrationEventBus.Setup(e => e.GetAllEventTypes()).Returns(() => - AppDomain.CurrentDomain.GetAssemblies().SelectMany(assembly => assembly.GetTypes()) - .Where(type => typeof(IIntegrationEvent).IsAssignableFrom(type))); - options.Services.AddScoped(serviceProvider => integrationEventBus.Object); + await using var transcation = await response.CustomDbContext.Database.BeginTransactionAsync(); + var logService = response.ServiceProvider.GetRequiredService(); - options.Services.AddScoped(); - options.UseEventLog().UseEventLog(); + var @event = new OrderPaymentSucceededIntegrationEvent + { + OrderId = "1234567890123", + PaymentTime = (long)(DateTime.UtcNow - new DateTime(1970, 1, 1, 0, 0, 0, DateTimeKind.Utc)).TotalSeconds + }; + await response.CustomDbContext.Set().AddAsync(new IntegrationEventLog(@event, Guid.NewGuid()) + { + State = IntegrationEventStates.Published + }); - var serviceProvider = options.Services.BuildServiceProvider(); - var eventLogService = serviceProvider.GetRequiredService(); + await response.CustomDbContext.SaveChangesAsync(); - var @event = new OrderPaymentSucceededIntegrationEvent() + await Assert.ThrowsExceptionAsync(async () => await logService.MarkEventAsInProgressAsync(@event.Id)); + } + + [TestMethod] + public async Task TestMarkEventAsPublishedAsync() + { + var response = await InitializeAsync(); + Assert.IsTrue(await response.CustomDbContext.Set().CountAsync() == 0); + + await using var transcation = await response.CustomDbContext.Database.BeginTransactionAsync(); + var logService = response.ServiceProvider.GetRequiredService(); + var @event = new OrderPaymentSucceededIntegrationEvent + { + OrderId = "1234567890123", + PaymentTime = (long)(DateTime.UtcNow - new DateTime(1970, 1, 1, 0, 0, 0, DateTimeKind.Utc)).TotalSeconds + }; + await logService.SaveEventAsync(@event, transcation.GetDbTransaction()); + + await logService.MarkEventAsInProgressAsync(@event.Id); + + await logService.MarkEventAsPublishedAsync(@event.Id); + + await transcation.CommitAsync(); + + Assert.IsTrue(await response.CustomDbContext.Set() + .CountAsync(log => log.State == IntegrationEventStates.Published) == 1); + Assert.IsTrue(await response.CustomDbContext.Set().CountAsync() == 1); + } + + [TestMethod] + public async Task TestMarkEventAsPublished2Async() + { + var response = await InitializeAsync(); + Assert.IsTrue(await response.CustomDbContext.Set().CountAsync() == 0); + + await using var transcation = await response.CustomDbContext.Database.BeginTransactionAsync(); + var logService = response.ServiceProvider.GetRequiredService(); + var @event = new OrderPaymentSucceededIntegrationEvent + { + OrderId = "1234567890123", + PaymentTime = (long)(DateTime.UtcNow - new DateTime(1970, 1, 1, 0, 0, 0, DateTimeKind.Utc)).TotalSeconds + }; + await logService.SaveEventAsync(@event, transcation.GetDbTransaction()); + + await Assert.ThrowsExceptionAsync(async () => await logService.MarkEventAsPublishedAsync(@event.Id)); + } + + [TestMethod] + public async Task TestMarkEventAsFailedAsync() + { + var response = await InitializeAsync(); + Assert.IsTrue(await response.CustomDbContext.Set().CountAsync() == 0); + + await using var transcation = await response.CustomDbContext.Database.BeginTransactionAsync(); + var logService = response.ServiceProvider.GetRequiredService(); + var @event = new OrderPaymentSucceededIntegrationEvent { OrderId = "1234567890123", PaymentTime = (long)(DateTime.UtcNow - new DateTime(1970, 1, 1, 0, 0, 0, DateTimeKind.Utc)).TotalSeconds }; + await logService.SaveEventAsync(@event, transcation.GetDbTransaction()); + await logService.MarkEventAsInProgressAsync(@event.Id); + await logService.MarkEventAsFailedAsync(@event.Id); + await transcation.CommitAsync(); + + Assert.IsTrue(await response.CustomDbContext.Set().CountAsync(log => log.State == IntegrationEventStates.PublishedFailed) == 1); + } - var dbContext = serviceProvider.GetRequiredService(); - await dbContext.Database.EnsureCreatedAsync(); + [TestMethod] + public async Task TestMarkEventAsFailed2Async() + { + var response = await InitializeAsync(); + Assert.IsTrue(await response.CustomDbContext.Set().CountAsync() == 0); + + await using var transcation = await response.CustomDbContext.Database.BeginTransactionAsync(); + var logService = response.ServiceProvider.GetRequiredService(); + var @event = new OrderPaymentSucceededIntegrationEvent + { + OrderId = "1234567890123", + PaymentTime = (long)(DateTime.UtcNow - new DateTime(1970, 1, 1, 0, 0, 0, DateTimeKind.Utc)).TotalSeconds + }; + await logService.SaveEventAsync(@event, transcation.GetDbTransaction()); + await Assert.ThrowsExceptionAsync(async () => await logService.MarkEventAsFailedAsync(@event.Id)); + } + + private async Task<(CustomDbContext CustomDbContext, IServiceProvider ServiceProvider)> InitializeAsync() + { + var dispatcherOptions = CreateDispatcherOptions(new ServiceCollection()); + dispatcherOptions.UseEventLog(); + dispatcherOptions.Services.AddMasaDbContext(option => + option.DbContextOptionsBuilder.UseSqlite(Connection)); + dispatcherOptions.Services.AddScoped(); + Mock integrationEventBus = new(); + var types = AppDomain.CurrentDomain.GetAssemblies() + .SelectMany(assembly => assembly.GetTypes()) + .Where(type => type.IsClass && typeof(IEvent).IsAssignableFrom(type)) + .ToList(); + integrationEventBus.Setup(eventBus => eventBus.GetAllEventTypes()).Returns(types).Verifiable(); + dispatcherOptions.Services.AddScoped(_ => integrationEventBus.Object); + var serviceProvider = dispatcherOptions.Services.BuildServiceProvider(); + var customDbContext = serviceProvider.GetRequiredService(); + await customDbContext.Database.EnsureCreatedAsync(); + return new(customDbContext, serviceProvider); } } diff --git a/test/Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF.Tests/Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF.Tests.csproj b/test/Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF.Tests/Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF.Tests.csproj index f96a49f85..06704c6c0 100644 --- a/test/Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF.Tests/Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF.Tests.csproj +++ b/test/Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF.Tests/Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF.Tests.csproj @@ -14,7 +14,6 @@ runtime; build; native; contentfiles; analyzers; buildtransitive - diff --git a/test/Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF.Tests/TestBase.cs b/test/Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF.Tests/TestBase.cs index 6db9f6b81..defeee026 100644 --- a/test/Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF.Tests/TestBase.cs +++ b/test/Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF.Tests/TestBase.cs @@ -16,32 +16,12 @@ public void Dispose() Connection.Close(); } - protected IServiceProvider CreateDefaultProvider(Action? action = null) + protected IDispatcherOptions CreateDispatcherOptions(IServiceCollection services, Assembly[]? assemblies = null) { - var services = new ServiceCollection(); - services.AddScoped(); - var options = new DispatcherOptions(services); - services.AddMasaDbContext(builder => builder.UseSqlite(ConnectionString)); - action?.Invoke(options); - - var integrationEventBus = new Mock(); - integrationEventBus.Setup(e => e.GetAllEventTypes()).Returns(() - => AppDomain.CurrentDomain.GetAssemblies().SelectMany(assembly => assembly.GetTypes()) - .Where(type => typeof(IIntegrationEvent).IsAssignableFrom(type))); - services.AddScoped(serviceProvider => integrationEventBus.Object); - return services.BuildServiceProvider(); - } -} - -public class DispatcherOptions : IDispatcherOptions -{ - public IServiceCollection Services { get; } - - public Assembly[] Assemblies { get; } - - public DispatcherOptions(IServiceCollection services, Assembly[]? assemblies = null) - { - this.Services = services; - Assemblies = assemblies ?? AppDomain.CurrentDomain.GetAssemblies(); + Mock dispatcherOptions = new(); + dispatcherOptions.Setup(option => option.Services).Returns(services).Verifiable(); + dispatcherOptions.Setup(option => option.Assemblies).Returns(assemblies ?? AppDomain.CurrentDomain.GetAssemblies()) + .Verifiable(); + return dispatcherOptions.Object; } } diff --git a/test/Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF.Tests/_Imports.cs b/test/Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF.Tests/_Imports.cs index cfbaca7e3..0be1c9cb8 100644 --- a/test/Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF.Tests/_Imports.cs +++ b/test/Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF.Tests/_Imports.cs @@ -9,6 +9,7 @@ global using Masa.Utils.Data.EntityFrameworkCore.Sqlite; global using Microsoft.Data.Sqlite; global using Microsoft.EntityFrameworkCore; +global using Microsoft.EntityFrameworkCore.Storage; global using Microsoft.Extensions.DependencyInjection; global using Microsoft.VisualStudio.TestTools.UnitTesting; global using Moq; From b03f8bdb66e820fd9fbfefd1cdb45a8b3259a904 Mon Sep 17 00:00:00 2001 From: zhenlei520 Date: Sun, 27 Mar 2022 17:35:26 +0800 Subject: [PATCH 3/8] chore(Repository.EF): change _context To Context --- .../Repository.cs | 46 +++++++++---------- 1 file changed, 23 insertions(+), 23 deletions(-) diff --git a/src/Ddd/Masa.Contrib.Ddd.Domain.Repository.EF/Repository.cs b/src/Ddd/Masa.Contrib.Ddd.Domain.Repository.EF/Repository.cs index d849b0544..239910865 100644 --- a/src/Ddd/Masa.Contrib.Ddd.Domain.Repository.EF/Repository.cs +++ b/src/Ddd/Masa.Contrib.Ddd.Domain.Repository.EF/Repository.cs @@ -5,16 +5,16 @@ public class Repository : where TEntity : class, IEntity where TDbContext : DbContext { - protected readonly TDbContext _context; + protected readonly TDbContext Context; public Repository(TDbContext context, IUnitOfWork unitOfWork) { - _context = context; + Context = context; UnitOfWork = unitOfWork; } public override bool TransactionHasBegun - => _context.Database.CurrentTransaction != null; + => Context.Database.CurrentTransaction != null; public override DbTransaction Transaction { @@ -24,9 +24,9 @@ public override DbTransaction Transaction throw new NotSupportedException(nameof(Transaction)); if (TransactionHasBegun) - return _context.Database.CurrentTransaction!.GetDbTransaction(); + return Context.Database.CurrentTransaction!.GetDbTransaction(); - return _context.Database.BeginTransaction().GetDbTransaction(); + return Context.Database.BeginTransaction().GetDbTransaction(); } } @@ -53,7 +53,7 @@ public override async ValueTask AddAsync( TEntity entity, CancellationToken cancellationToken = default) { - var response = (await _context.AddAsync(entity, cancellationToken).AsTask()).Entity; + var response = (await Context.AddAsync(entity, cancellationToken).AsTask()).Entity; EntityState = EntityState.Changed; return response; } @@ -62,45 +62,45 @@ public override async Task AddRangeAsync( IEnumerable entities, CancellationToken cancellationToken = default) { - await _context.AddRangeAsync(entities, cancellationToken); + await Context.AddRangeAsync(entities, cancellationToken); EntityState = EntityState.Changed; } public override Task CommitAsync(CancellationToken cancellationToken = default) => UnitOfWork.CommitAsync(cancellationToken); - public override async ValueTask DisposeAsync() => await _context.DisposeAsync(); + public override async ValueTask DisposeAsync() => await Context.DisposeAsync(); - public override void Dispose() => _context.Dispose(); + public override void Dispose() => Context.Dispose(); public override Task FindAsync( IEnumerable> keyValues, CancellationToken cancellationToken = default) { Dictionary fields = new(keyValues); - return _context.Set().IgnoreQueryFilters().GetQueryable(fields).FirstOrDefaultAsync(cancellationToken); + return Context.Set().IgnoreQueryFilters().GetQueryable(fields).FirstOrDefaultAsync(cancellationToken); } public override Task FindAsync( Expression> predicate, CancellationToken cancellationToken = default) - => _context.Set().Where(predicate).FirstOrDefaultAsync(cancellationToken); + => Context.Set().Where(predicate).FirstOrDefaultAsync(cancellationToken); public override async Task GetCountAsync(CancellationToken cancellationToken = default) - => await _context.Set().LongCountAsync(cancellationToken); + => await Context.Set().LongCountAsync(cancellationToken); public override Task GetCountAsync( Expression> predicate, CancellationToken cancellationToken = default) - => _context.Set().LongCountAsync(predicate, cancellationToken); + => Context.Set().LongCountAsync(predicate, cancellationToken); public override async Task> GetListAsync(CancellationToken cancellationToken = default) - => await _context.Set().ToListAsync(cancellationToken); + => await Context.Set().ToListAsync(cancellationToken); public override async Task> GetListAsync( Expression> predicate, CancellationToken cancellationToken = default) - => await _context.Set().Where(predicate).ToListAsync(cancellationToken); + => await Context.Set().Where(predicate).ToListAsync(cancellationToken); /// /// @@ -118,7 +118,7 @@ public override Task> GetPaginatedListAsync( { sorting ??= new Dictionary(); - return _context.Set().OrderBy(sorting).Skip(skip).Take(take).ToListAsync(cancellationToken); + return Context.Set().OrderBy(sorting).Skip(skip).Take(take).ToListAsync(cancellationToken); } /// @@ -139,12 +139,12 @@ public override Task> GetPaginatedListAsync( { sorting ??= new Dictionary(); - return _context.Set().Where(predicate).OrderBy(sorting).Skip(skip).Take(take).ToListAsync(cancellationToken); + return Context.Set().Where(predicate).OrderBy(sorting).Skip(skip).Take(take).ToListAsync(cancellationToken); } public override Task RemoveAsync(TEntity entity, CancellationToken cancellationToken = default) { - _context.Set().Remove(entity); + Context.Set().Remove(entity); EntityState = EntityState.Changed; return Task.FromResult(entity); } @@ -153,12 +153,12 @@ public override async Task RemoveAsync(Expression> predicate { var entities = await GetListAsync(predicate, cancellationToken); EntityState = EntityState.Changed; - _context.Set().RemoveRange(entities); + Context.Set().RemoveRange(entities); } public override Task RemoveRangeAsync(IEnumerable entities, CancellationToken cancellationToken = default) { - _context.Set().RemoveRange(entities); + Context.Set().RemoveRange(entities); EntityState = EntityState.Changed; return Task.CompletedTask; } @@ -173,14 +173,14 @@ public override async Task SaveChangesAsync(CancellationToken cancellationToken public override Task UpdateAsync(TEntity entity, CancellationToken cancellationToken = default) { - _context.Set().Update(entity); + Context.Set().Update(entity); EntityState = EntityState.Changed; return Task.FromResult(entity); } public override Task UpdateRangeAsync(IEnumerable entities, CancellationToken cancellationToken = default) { - _context.Set().UpdateRange(entities); + Context.Set().UpdateRange(entities); EntityState = EntityState.Changed; return Task.CompletedTask; } @@ -213,5 +213,5 @@ public Repository(TDbContext context, IUnitOfWork unitOfWork) : base(context, un } public Task FindAsync(TKey id) - => _context.Set().FirstOrDefaultAsync(entity => entity.Id.Equals(id)); + => Context.Set().FirstOrDefaultAsync(entity => entity.Id.Equals(id)); } From 4605a168fe84a63bfb3a35b2ce8b7a8006b1b8e1 Mon Sep 17 00:00:00 2001 From: zhenlei520 Date: Sun, 27 Mar 2022 17:36:14 +0800 Subject: [PATCH 4/8] fix(UoW): Fix UnitOfWork release DbContext problem --- src/Data/Masa.Contrib.Data.UoW.EF/UnitOfWork.cs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/Data/Masa.Contrib.Data.UoW.EF/UnitOfWork.cs b/src/Data/Masa.Contrib.Data.UoW.EF/UnitOfWork.cs index 69b47d6db..1e97ee7b7 100644 --- a/src/Data/Masa.Contrib.Data.UoW.EF/UnitOfWork.cs +++ b/src/Data/Masa.Contrib.Data.UoW.EF/UnitOfWork.cs @@ -61,7 +61,7 @@ public async Task RollbackAsync(CancellationToken cancellationToken = default) await Context.Database.RollbackTransactionAsync(cancellationToken); } - public ValueTask DisposeAsync() => Context.DisposeAsync(); + public async ValueTask DisposeAsync() => await (_context?.DisposeAsync() ?? ValueTask.CompletedTask); - public void Dispose() => Context.Dispose(); + public void Dispose() => _context?.Dispose(); } From 7f55946e098de9497f510b6819241d652af00e3b Mon Sep 17 00:00:00 2001 From: zhenlei520 Date: Sun, 27 Mar 2022 17:37:11 +0800 Subject: [PATCH 5/8] refactor(IntegrationEvents.Dapr): adjust AddDaprEventBus methods parameter --- .../DispatcherOptionsExtensions.cs | 34 ++++++++----------- .../ServiceCollectionExtensions.cs | 9 ++--- 2 files changed, 19 insertions(+), 24 deletions(-) diff --git a/src/Dispatcher/Masa.Contrib.Dispatcher.IntegrationEvents.Dapr/DispatcherOptionsExtensions.cs b/src/Dispatcher/Masa.Contrib.Dispatcher.IntegrationEvents.Dapr/DispatcherOptionsExtensions.cs index 6a8506ed8..25261462f 100644 --- a/src/Dispatcher/Masa.Contrib.Dispatcher.IntegrationEvents.Dapr/DispatcherOptionsExtensions.cs +++ b/src/Dispatcher/Masa.Contrib.Dispatcher.IntegrationEvents.Dapr/DispatcherOptionsExtensions.cs @@ -5,36 +5,30 @@ public static class DispatcherOptionsExtensions private const string DAPR_PUBSUB_NAME = "pubsub"; public static IDistributedDispatcherOptions UseDaprEventBus( - this IDistributedDispatcherOptions options, + this IDistributedDispatcherOptions dispatcherOptions, string daprPubSubName) where TIntegrationEventLogService : class, IIntegrationEventLogService - => options.UseDaprEventBus(daprPubSubName, null); + => dispatcherOptions.UseDaprEventBus(option => option.PubSubName = daprPubSubName); public static IDistributedDispatcherOptions UseDaprEventBus( - this IDistributedDispatcherOptions options, - string daprPubSubName, - Action? builder) + this IDistributedDispatcherOptions dispatcherOptions, + Action? optionAction = null) where TIntegrationEventLogService : class, IIntegrationEventLogService - { - return options.UseDaprEventBus(builder, dispatcherOptions => - { - dispatcherOptions.PubSubName = daprPubSubName; - }); - } + => dispatcherOptions.UseDaprEventBus(optionAction, null); public static IDistributedDispatcherOptions UseDaprEventBus( - this IDistributedDispatcherOptions options, - Action? builder = null, - Action? action = null) + this IDistributedDispatcherOptions dispatcherOptions, + Action? optionAction, + Action? builder = null) where TIntegrationEventLogService : class, IIntegrationEventLogService { - ArgumentNullException.ThrowIfNull(options.Services, nameof(options.Services)); + ArgumentNullException.ThrowIfNull(dispatcherOptions.Services, nameof(dispatcherOptions.Services)); - options.Services.TryAddDaprEventBus(options.Assemblies, builder, dispatcherOptions => + dispatcherOptions.Services.TryAddDaprEventBus(dispatcherOptions.Assemblies, option => { - dispatcherOptions.PubSubName = DAPR_PUBSUB_NAME; - action?.Invoke(dispatcherOptions); - }); - return options; + option.PubSubName = DAPR_PUBSUB_NAME; + optionAction?.Invoke(option); + }, builder); + return dispatcherOptions; } } diff --git a/src/Dispatcher/Masa.Contrib.Dispatcher.IntegrationEvents.Dapr/ServiceCollectionExtensions.cs b/src/Dispatcher/Masa.Contrib.Dispatcher.IntegrationEvents.Dapr/ServiceCollectionExtensions.cs index 104907f94..eb585e51b 100644 --- a/src/Dispatcher/Masa.Contrib.Dispatcher.IntegrationEvents.Dapr/ServiceCollectionExtensions.cs +++ b/src/Dispatcher/Masa.Contrib.Dispatcher.IntegrationEvents.Dapr/ServiceCollectionExtensions.cs @@ -11,15 +11,16 @@ public static IServiceCollection AddDaprEventBus( public static IServiceCollection AddDaprEventBus( this IServiceCollection services, Assembly[] assemblies, - Action? options = null) + Action? options = null, + Action? builder = null) where TIntegrationEventLogService : class, IIntegrationEventLogService - => services.TryAddDaprEventBus(assemblies, null, options); + => services.TryAddDaprEventBus(assemblies, options, builder); internal static IServiceCollection TryAddDaprEventBus( this IServiceCollection services, Assembly[] assemblies, - Action? builder, - Action? options = null) + Action? options, + Action? builder = null) where TIntegrationEventLogService : class, IIntegrationEventLogService { if (services.Any(service => service.ImplementationType == typeof(IntegrationEventBusProvider))) From 22266a1f68218fbf865b4ef583a6ad1d25d795aa Mon Sep 17 00:00:00 2001 From: zhenlei520 Date: Sun, 27 Mar 2022 18:01:40 +0800 Subject: [PATCH 6/8] chore(IntegrationEvents.Dapr): adjust UseDaprEventBus methods parameter --- .../DispatcherOptionsExtensions.cs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/Dispatcher/Masa.Contrib.Dispatcher.IntegrationEvents.Dapr/DispatcherOptionsExtensions.cs b/src/Dispatcher/Masa.Contrib.Dispatcher.IntegrationEvents.Dapr/DispatcherOptionsExtensions.cs index 25261462f..9b8a2203e 100644 --- a/src/Dispatcher/Masa.Contrib.Dispatcher.IntegrationEvents.Dapr/DispatcherOptionsExtensions.cs +++ b/src/Dispatcher/Masa.Contrib.Dispatcher.IntegrationEvents.Dapr/DispatcherOptionsExtensions.cs @@ -19,7 +19,7 @@ public static IDistributedDispatcherOptions UseDaprEventBus( this IDistributedDispatcherOptions dispatcherOptions, Action? optionAction, - Action? builder = null) + Action? builder) where TIntegrationEventLogService : class, IIntegrationEventLogService { ArgumentNullException.ThrowIfNull(dispatcherOptions.Services, nameof(dispatcherOptions.Services)); From 326559dad23ead32aa6b1b822a3aeefcee3fc32f Mon Sep 17 00:00:00 2001 From: zhenlei520 Date: Sun, 27 Mar 2022 18:14:54 +0800 Subject: [PATCH 7/8] chore: Remove dead code --- .../AssemblyResolutionTests.cs | 12 ------------ 1 file changed, 12 deletions(-) diff --git a/test/Masa.Contrib.Dispatcher.Events.Tests/AssemblyResolutionTests.cs b/test/Masa.Contrib.Dispatcher.Events.Tests/AssemblyResolutionTests.cs index a5aafb380..cb70a8628 100644 --- a/test/Masa.Contrib.Dispatcher.Events.Tests/AssemblyResolutionTests.cs +++ b/test/Masa.Contrib.Dispatcher.Events.Tests/AssemblyResolutionTests.cs @@ -50,18 +50,6 @@ public void TestEventBusByAddNullAssembly() }); } - // [TestMethod] - // public void TestEventBusByAddEmptyAssembly() - // { - // var services = new ServiceCollection(); - // services.AddLogging(loggingBuilder => loggingBuilder.AddConsole()); - // services.AddTransient(typeof(IMiddleware<>), typeof(LoggingMiddleware<>)); - // Assert.ThrowsException(() => - // { - // services.AddTestEventBus(Array.Empty(), ServiceLifetime.Scoped); - // }); - // } - [TestMethod] public void TestEventBus() { From a4faac3f936293d5bd7bcf202ed6e1e20c9478b9 Mon Sep 17 00:00:00 2001 From: zhenlei520 Date: Mon, 28 Mar 2022 10:59:48 +0800 Subject: [PATCH 8/8] chore: Modify the EfCommon class name to DbContextExtensions --- .../Internal/{EfCommon.cs => DbContextExtensions.cs} | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) rename src/Dispatcher/Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF/Internal/{EfCommon.cs => DbContextExtensions.cs} (97%) diff --git a/src/Dispatcher/Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF/Internal/EfCommon.cs b/src/Dispatcher/Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF/Internal/DbContextExtensions.cs similarity index 97% rename from src/Dispatcher/Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF/Internal/EfCommon.cs rename to src/Dispatcher/Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF/Internal/DbContextExtensions.cs index 0d3467609..23e39a957 100644 --- a/src/Dispatcher/Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF/Internal/EfCommon.cs +++ b/src/Dispatcher/Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF/Internal/DbContextExtensions.cs @@ -1,6 +1,6 @@ namespace Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EF.Internal; -internal static class EfCommon +internal static class DbContextExtensions { private static readonly ConcurrentDictionary TableDictionary = new();