diff --git a/infrastructure.ps1 b/infrastructure.ps1 new file mode 100644 index 00000000..545a7ede --- /dev/null +++ b/infrastructure.ps1 @@ -0,0 +1 @@ +docker compose -f src/Infrastructure/docker-compose.yml up --force-recreate -V \ No newline at end of file diff --git a/src/Host.Plugin.Properties.xml b/src/Host.Plugin.Properties.xml index ffce6c5b..038b631a 100644 --- a/src/Host.Plugin.Properties.xml +++ b/src/Host.Plugin.Properties.xml @@ -4,7 +4,7 @@ - 2.3.5 + 2.4.0-rc1 \ No newline at end of file diff --git a/src/Infrastructure/README.md b/src/Infrastructure/README.md index ba4b9d51..caa4f7b5 100644 --- a/src/Infrastructure/README.md +++ b/src/Infrastructure/README.md @@ -8,12 +8,18 @@ No volumes have been configured so as to provide clean instances on each run. src/Infrastructure> docker compose up --force-recreate -V ``` -or +or Unix ``` > ./infrastructure.sh ``` +or Windows + +``` +> .\infrastructure.ps1 +``` + ## Configuration Personal instances of Azure resources are required as containers are not available for those services. @@ -26,7 +32,7 @@ Personal instances of Azure resources are required as containers are not availab ### Azure Event Hub -The transport/plug-in does not provide topology provisioning (just yet). +The transport/plug-in does not provide topology provisioning ([just yet](https://github.com/zarusz/SlimMessageBus/issues/111)). The below event hubs and consumer groups will need to be manually added to the Azure Event Hub instance for the integration tests to run. | Event Hub | Consumer Group | diff --git a/src/SlimMessageBus.Host.AzureServiceBus/ServiceBusMessageBus.cs b/src/SlimMessageBus.Host.AzureServiceBus/ServiceBusMessageBus.cs index 8ee49042..a9e76996 100644 --- a/src/SlimMessageBus.Host.AzureServiceBus/ServiceBusMessageBus.cs +++ b/src/SlimMessageBus.Host.AzureServiceBus/ServiceBusMessageBus.cs @@ -109,7 +109,7 @@ void AddConsumerFrom(TopicSubscriptionParams topicSubscription, IMessageProcesso responseConsumer: this, messagePayloadProvider: m => m.Body.ToArray()); - AddConsumerFrom(topicSubscription, messageProcessor, new[] { Settings.RequestResponse }); + AddConsumerFrom(topicSubscription, messageProcessor, [Settings.RequestResponse]); } } diff --git a/src/SlimMessageBus.Host.Interceptor/Bus/MessageBusLifecycleEventType.cs b/src/SlimMessageBus.Host.Interceptor/Bus/MessageBusLifecycleEventType.cs index 1b5b4baf..ac5d73c1 100644 --- a/src/SlimMessageBus.Host.Interceptor/Bus/MessageBusLifecycleEventType.cs +++ b/src/SlimMessageBus.Host.Interceptor/Bus/MessageBusLifecycleEventType.cs @@ -2,8 +2,13 @@ public enum MessageBusLifecycleEventType { + /// + /// Invoked when the master bus is created. + /// Can be used to initalize any resource before the messages are produced or consumed. + /// + Created, Starting, Started, Stopping, - Stopped, + Stopped } \ No newline at end of file diff --git a/src/SlimMessageBus.Host.Outbox/Interceptors/OutboxSendingTask.cs b/src/SlimMessageBus.Host.Outbox/Interceptors/OutboxSendingTask.cs index be05119a..7a5b4537 100644 --- a/src/SlimMessageBus.Host.Outbox/Interceptors/OutboxSendingTask.cs +++ b/src/SlimMessageBus.Host.Outbox/Interceptors/OutboxSendingTask.cs @@ -14,6 +14,12 @@ public class OutboxSendingTask( private CancellationTokenSource _loopCts; private Task _loopTask; + + private readonly object _migrateSchemaTaskLock = new(); + private Task _migrateSchemaTask; + private Task _startBusTask; + private Task _stopBusTask; + private int _busStartCount; private DateTime? _cleanupNextRun; @@ -75,35 +81,63 @@ protected async Task Stop() public Task OnBusLifecycle(MessageBusLifecycleEventType eventType, IMessageBus bus) { + if (eventType == MessageBusLifecycleEventType.Created) + { + return EnsureMigrateSchema(_serviceProvider, default); + } if (eventType == MessageBusLifecycleEventType.Started) { // The first started bus starts this outbox task if (Interlocked.Increment(ref _busStartCount) == 1) { - return Start(); + _startBusTask = Start(); } + return _startBusTask; } if (eventType == MessageBusLifecycleEventType.Stopping) { // The last stopped bus stops this outbox task if (Interlocked.Decrement(ref _busStartCount) == 0) { - return Stop(); + _stopBusTask = Stop(); } + return _stopBusTask; } return Task.CompletedTask; } + private static async Task MigrateSchema(IServiceProvider serviceProvider, CancellationToken cancellationToken) + { + try + { + var outboxMigrationService = serviceProvider.GetRequiredService(); + await outboxMigrationService.Migrate(cancellationToken); + } + catch (Exception e) + { + throw new MessageBusException("Outbox schema migration failed", e); + } + } + + private Task EnsureMigrateSchema(IServiceProvider serviceProvider, CancellationToken cancellationToken) + { + lock (_migrateSchemaTaskLock) + { + // We optimize to ever only run once the schema migration, regardless if it was triggered from 1) bus created lifecycle or 2) outbox sending loop. + return _migrateSchemaTask ??= MigrateSchema(serviceProvider, cancellationToken); + } + } + private async Task Run() { try { _logger.LogInformation("Outbox loop started"); var scope = _serviceProvider.CreateScope(); + try { - var outboxMigrationService = scope.ServiceProvider.GetRequiredService(); - await outboxMigrationService.Migrate(_loopCts.Token); + await EnsureMigrateSchema(scope.ServiceProvider, _loopCts.Token); var outboxRepository = scope.ServiceProvider.GetRequiredService(); @@ -187,17 +221,17 @@ private async Task SendMessages(IServiceProvider serviceProvider, IOutboxR for (var i = 0; i < outboxMessages.Count && !ct.IsCancellationRequested; i++) { var outboxMessage = outboxMessages[i]; - - var now = DateTime.UtcNow; - if (now.Add(_outboxSettings.LockExpirationBuffer) > outboxMessage.LockExpiresOn) - { - _logger.LogDebug("Stopping the outbox message processing after {MessageCount} (out of {BatchCount}) because the message lock was close to expiration {LockBuffer}", i, _outboxSettings.PollBatchSize, _outboxSettings.LockExpirationBuffer); - hasMore = false; - break; - } - - var bus = GetBus(compositeMessageBus, messageBusTarget, outboxMessage.BusName); - if (bus == null) + + var now = DateTime.UtcNow; + if (now.Add(_outboxSettings.LockExpirationBuffer) > outboxMessage.LockExpiresOn) + { + _logger.LogDebug("Stopping the outbox message processing after {MessageCount} (out of {BatchCount}) because the message lock was close to expiration {LockBuffer}", i, _outboxSettings.PollBatchSize, _outboxSettings.LockExpirationBuffer); + hasMore = false; + break; + } + + var bus = GetBus(compositeMessageBus, messageBusTarget, outboxMessage.BusName); + if (bus == null) { _logger.LogWarning("Not able to find matching bus provider for the outbox message with Id {MessageId} of type {MessageType} to path {Path} using {BusName} bus. The message will be skipped.", outboxMessage.Id, outboxMessage.MessageType.Name, outboxMessage.Path, outboxMessage.BusName); continue; diff --git a/src/SlimMessageBus.Host/MessageBusBase.cs b/src/SlimMessageBus.Host/MessageBusBase.cs index 1b4c22e8..a447b176 100644 --- a/src/SlimMessageBus.Host/MessageBusBase.cs +++ b/src/SlimMessageBus.Host/MessageBusBase.cs @@ -144,23 +144,27 @@ protected void OnBuildProvider() { ValidationService.AssertSettings(); - Build(); - - if (Settings.AutoStartConsumers) - { - // Fire and forget start - _ = Task.Run(async () => - { - try - { - await Start().ConfigureAwait(false); - } - catch (Exception e) - { - _logger.LogError(e, "Could not auto start consumers"); - } + Build(); + + // Notify the bus has been created - before any message can be produced + AddInit(OnBusLifecycle(MessageBusLifecycleEventType.Created)); + + // Auto start consumers if enabled + if (Settings.AutoStartConsumers) + { + // Fire and forget start + _ = Task.Run(async () => + { + try + { + await Start().ConfigureAwait(false); + } + catch (Exception e) + { + _logger.LogError(e, "Could not auto start consumers"); + } }); - } + } } protected virtual void Build() @@ -196,7 +200,7 @@ private Dictionary BuildProducerByBaseMessageType() private async Task OnBusLifecycle(MessageBusLifecycleEventType eventType) { - _lifecycleInterceptors ??= Settings.ServiceProvider?.GetService>(); + _lifecycleInterceptors ??= Settings.ServiceProvider?.GetServices(); if (_lifecycleInterceptors != null) { foreach (var i in _lifecycleInterceptors) diff --git a/src/SlimMessageBus.sln b/src/SlimMessageBus.sln index 1c2a380a..18b45c9d 100644 --- a/src/SlimMessageBus.sln +++ b/src/SlimMessageBus.sln @@ -241,11 +241,21 @@ EndProject Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Infrastructure", "Infrastructure", "{137BFD32-CD0A-47CA-8884-209CD49DEE8C}" ProjectSection(SolutionItems) = preProject Infrastructure\docker-compose.yml = Infrastructure\docker-compose.yml + ..\infrastructure.ps1 = ..\infrastructure.ps1 ..\infrastructure.sh = ..\infrastructure.sh Infrastructure\mosquitto.conf = Infrastructure\mosquitto.conf Infrastructure\README.md = Infrastructure\README.md EndProjectSection EndProject +Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Build", "Build", "{1A71BB05-58ED-4B27-B4A4-A03D9E608C1C}" + ProjectSection(SolutionItems) = preProject + ..\build\do_build.ps1 = ..\build\do_build.ps1 + ..\build\do_package.ps1 = ..\build\do_package.ps1 + ..\build\do_test.ps1 = ..\build\do_test.ps1 + ..\build\do_test_ci.ps1 = ..\build\do_test_ci.ps1 + ..\build\tasks.ps1 = ..\build\tasks.ps1 + EndProjectSection +EndProject Global GlobalSection(SolutionConfigurationPlatforms) = preSolution Debug|Any CPU = Debug|Any CPU diff --git a/src/Tests/SlimMessageBus.Host.AzureServiceBus.Test/ServiceBusMessageBusIt.cs b/src/Tests/SlimMessageBus.Host.AzureServiceBus.Test/ServiceBusMessageBusIt.cs index b1e2c281..9a192d9d 100644 --- a/src/Tests/SlimMessageBus.Host.AzureServiceBus.Test/ServiceBusMessageBusIt.cs +++ b/src/Tests/SlimMessageBus.Host.AzureServiceBus.Test/ServiceBusMessageBusIt.cs @@ -37,6 +37,8 @@ protected override void SetupServices(ServiceCollection services, IConfiguration } await next(); }; + cfg.PrefetchCount = 100; + cfg.MaxConcurrentSessions = 20; }); mbb.AddServicesFromAssemblyContaining(); mbb.AddJsonSerializer(); @@ -68,7 +70,6 @@ private static void MessageModifierWithSession(PingMessage message, ServiceBusMe [Fact] public async Task BasicPubSubOnTopic() { - var concurrency = 2; var subscribers = 2; var topic = "test-ping"; @@ -82,7 +83,7 @@ public async Task BasicPubSubOnTopic() .SubscriptionName($"subscriber-{i}") // ensure subscription exists on the ServiceBus topic .WithConsumer() .WithConsumer() - .Instances(concurrency)); + .Instances(20)); })); }); @@ -94,13 +95,12 @@ public async Task BasicPubSubOnTopic() } }; - await BasicPubSub(concurrency, subscribers, subscribers); + await BasicPubSub(subscribers); } [Fact] public async Task BasicPubSubOnQueue() { - var concurrency = 2; var queue = "test-ping-queue"; AddBusConfiguration(mbb => @@ -111,7 +111,7 @@ public async Task BasicPubSubOnQueue() .Queue(queue) .WithConsumer() .WithConsumer() - .Instances(concurrency)); + .Instances(20)); }); CleanTopology = async client => @@ -122,13 +122,12 @@ public async Task BasicPubSubOnQueue() } }; - await BasicPubSub(concurrency, 1, 1); + await BasicPubSub(1); } [Fact] public async Task BasicPubSubWithCustomConsumerOnQueue() { - var concurrency = 2; var queue = "test-ping-queue"; AddBusConfiguration(mbb => @@ -139,7 +138,7 @@ public async Task BasicPubSubWithCustomConsumerOnQueue() .Queue(queue) .WithConsumer(typeof(CustomPingConsumer), nameof(CustomPingConsumer.Handle)) .WithConsumer(typeof(CustomPingConsumer), typeof(PingDerivedMessage), nameof(CustomPingConsumer.Handle)) - .Instances(concurrency)); + .Instances(20)); }); CleanTopology = async client => @@ -150,7 +149,7 @@ public async Task BasicPubSubWithCustomConsumerOnQueue() } }; - await BasicPubSub(concurrency, 1, 1); + await BasicPubSub(1); } private static string GetMessageId(PingMessage message) => $"ID_{message.Counter}"; @@ -161,7 +160,7 @@ public class TestData public IReadOnlyCollection ConsumedMessages { get; set; } } - private async Task BasicPubSub(int concurrency, int subscribers, int expectedMessageCopies, Action additionalAssertion = null) + private async Task BasicPubSub(int expectedMessageCopies, Action additionalAssertion = null) { // arrange var testMetric = ServiceProvider.GetRequiredService(); @@ -186,7 +185,7 @@ private async Task BasicPubSub(int concurrency, int subscribers, int expectedMes } stopwatch.Stop(); - Logger.LogInformation("Published {0} messages in {1}", producedMessages.Count, stopwatch.Elapsed); + Logger.LogInformation("Published {Count} messages in {Elapsed}", producedMessages.Count, stopwatch.Elapsed); // consume stopwatch.Restart(); @@ -234,7 +233,7 @@ public async Task BasicReqRespOnTopic() .Handle(x => x.Topic(topic) .SubscriptionName("handler") .WithHandler() - .Instances(2)) + .Instances(20)) .ExpectRequestResponses(x => { x.ReplyToTopic("test-echo-resp"); @@ -259,7 +258,7 @@ public async Task BasicReqRespOnQueue() }) .Handle(x => x.Queue(queue) .WithHandler() - .Instances(2)) + .Instances(20)) .ExpectRequestResponses(x => { x.ReplyToQueue("test-echo-queue-resp"); @@ -293,7 +292,7 @@ private async Task BasicReqResp() await Task.WhenAll(responseTasks).ConfigureAwait(false); stopwatch.Stop(); - Logger.LogInformation("Published and received {0} messages in {1}", responses.Count, stopwatch.Elapsed); + Logger.LogInformation("Published and received {Count} messages in {Elapsed}", responses.Count, stopwatch.Elapsed); // assert @@ -305,7 +304,6 @@ private async Task BasicReqResp() [Fact] public async Task FIFOUsingSessionsOnQueue() { - var concurrency = 1; var queue = "test-session-queue"; AddBusConfiguration(mbb => @@ -316,10 +314,10 @@ public async Task FIFOUsingSessionsOnQueue() .Queue(queue) .WithConsumer() .WithConsumer() - .Instances(concurrency) + .Instances(1) .EnableSession(x => x.MaxConcurrentSessions(10).SessionIdleTimeout(TimeSpan.FromSeconds(5)))); }); - await BasicPubSub(concurrency, 1, 1, CheckMessagesWithinSameSessionAreInOrder); + await BasicPubSub(1, CheckMessagesWithinSameSessionAreInOrder); } private static void CheckMessagesWithinSameSessionAreInOrder(TestData testData) @@ -337,7 +335,6 @@ private static void CheckMessagesWithinSameSessionAreInOrder(TestData testData) [Fact] public async Task FIFOUsingSessionsOnTopic() { - var concurrency = 1; var queue = "test-session-topic"; AddBusConfiguration(mbb => @@ -347,12 +344,12 @@ public async Task FIFOUsingSessionsOnTopic() .Topic(queue) .WithConsumer() .WithConsumer() - .Instances(concurrency) + .Instances(1) .SubscriptionName($"subscriber") // ensure subscription exists on the ServiceBus topic .EnableSession(x => x.MaxConcurrentSessions(10).SessionIdleTimeout(TimeSpan.FromSeconds(5)))); }); - await BasicPubSub(concurrency, 1, 1, CheckMessagesWithinSameSessionAreInOrder); + await BasicPubSub(1, CheckMessagesWithinSameSessionAreInOrder); } } diff --git a/src/Tests/SlimMessageBus.Host.AzureServiceBus.Test/ServiceBusMessageBusTests.cs b/src/Tests/SlimMessageBus.Host.AzureServiceBus.Test/ServiceBusMessageBusTests.cs index b6d36696..cd52a4cb 100644 --- a/src/Tests/SlimMessageBus.Host.AzureServiceBus.Test/ServiceBusMessageBusTests.cs +++ b/src/Tests/SlimMessageBus.Host.AzureServiceBus.Test/ServiceBusMessageBusTests.cs @@ -6,6 +6,7 @@ using Azure.Messaging.ServiceBus; using SlimMessageBus.Host; +using SlimMessageBus.Host.Interceptor; using SlimMessageBus.Host.Serialization; public class ServiceBusMessageBusTests : IDisposable @@ -22,6 +23,7 @@ public ServiceBusMessageBusTests() serviceProviderMock.Setup(x => x.GetService(It.Is(t => t.IsGenericType && t.GetGenericTypeDefinition() == typeof(IEnumerable<>)))).Returns(Enumerable.Empty()); serviceProviderMock.Setup(x => x.GetService(typeof(IMessageSerializer))).Returns(new Mock().Object); serviceProviderMock.Setup(x => x.GetService(typeof(IMessageTypeResolver))).Returns(new AssemblyQualifiedNameMessageTypeResolver()); + serviceProviderMock.Setup(x => x.GetService(typeof(IEnumerable))).Returns(Array.Empty()); BusBuilder.WithDependencyResolver(serviceProviderMock.Object); diff --git a/src/Tests/SlimMessageBus.Host.Integration.Test/HybridTests.cs b/src/Tests/SlimMessageBus.Host.Integration.Test/HybridTests.cs index f13cde99..20d76ea8 100644 --- a/src/Tests/SlimMessageBus.Host.Integration.Test/HybridTests.cs +++ b/src/Tests/SlimMessageBus.Host.Integration.Test/HybridTests.cs @@ -78,11 +78,12 @@ private void SetupBus(MessageBusBuilder mbb, SerializerType serializerType) var topic = "integration-external-message"; mbb .Produce(x => x.DefaultTopic(topic)) - .Consume(x => x.Topic(topic)) + .Consume(x => x.Topic(topic).Instances(20)) .WithProviderServiceBus(cfg => { cfg.SubscriptionName("test"); cfg.ConnectionString = Secrets.Service.PopulateSecrets(_configuration["Azure:ServiceBus"]); + cfg.PrefetchCount = 100; }); }); } diff --git a/src/Tests/SlimMessageBus.Host.Outbox.DbContext.Test/BusType.cs b/src/Tests/SlimMessageBus.Host.Outbox.DbContext.Test/BusType.cs new file mode 100644 index 00000000..aaa7efbd --- /dev/null +++ b/src/Tests/SlimMessageBus.Host.Outbox.DbContext.Test/BusType.cs @@ -0,0 +1,7 @@ +namespace SlimMessageBus.Host.Outbox.DbContext.Test; + +public enum BusType +{ + AzureSB, + Kafka, +} diff --git a/src/Tests/SlimMessageBus.Host.Outbox.DbContext.Test/DatabaseFacadeExtenstions.cs b/src/Tests/SlimMessageBus.Host.Outbox.DbContext.Test/DatabaseFacadeExtenstions.cs new file mode 100644 index 00000000..92d2f921 --- /dev/null +++ b/src/Tests/SlimMessageBus.Host.Outbox.DbContext.Test/DatabaseFacadeExtenstions.cs @@ -0,0 +1,19 @@ +namespace SlimMessageBus.Host.Outbox.DbContext.Test; + +using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Infrastructure; + +public static class DatabaseFacadeExtenstions +{ + public static Task EraseTableIfExists(this DatabaseFacade db, string tableName) + { +#pragma warning disable EF1002 // Risk of vulnerability to SQL injection. + return db.ExecuteSqlRawAsync($""" + IF EXISTS (SELECT 1 FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA = 'dbo' AND TABLE_NAME = '{tableName}') + BEGIN + DELETE FROM dbo.{tableName}; + END + """); +#pragma warning restore EF1002 // Risk of vulnerability to SQL injection. + } +} diff --git a/src/Tests/SlimMessageBus.Host.Outbox.DbContext.Test/OutboxBenchmarkTests.cs b/src/Tests/SlimMessageBus.Host.Outbox.DbContext.Test/OutboxBenchmarkTests.cs new file mode 100644 index 00000000..97fb771f --- /dev/null +++ b/src/Tests/SlimMessageBus.Host.Outbox.DbContext.Test/OutboxBenchmarkTests.cs @@ -0,0 +1,139 @@ +namespace SlimMessageBus.Host.Outbox.DbContext.Test; + +/// +/// This test should help to understand the runtime performance and overhead of the outbox feature. +/// It will generate the time measurements for a given transport (Azure DB + Azure SQL instance) as the baseline, +/// and then measure times when outbox is enabled. +/// This should help asses the overhead of outbox and to baseline future outbox improvements. +/// +/// +[Trait("Category", "Integration")] // for benchmarks +public class OutboxBenchmarkTests(ITestOutputHelper testOutputHelper) : BaseIntegrationTest(testOutputHelper) +{ + private static readonly string OutboxTableName = "IntTest_Benchmark_Outbox"; + private static readonly string MigrationsTableName = "IntTest_Benchmark_Migrations"; + + private bool _useOutbox; + + protected override void SetupServices(ServiceCollection services, IConfigurationRoot configuration) + { + services.AddSlimMessageBus(mbb => + { + mbb.AddChildBus("ExternalBus", mbb => + { + var topic = "tests.outbox-benchmark/customer-events"; + mbb + .WithProviderServiceBus(cfg => + { + cfg.ConnectionString = Secrets.Service.PopulateSecrets(configuration["Azure:ServiceBus"]); + cfg.PrefetchCount = 100; // fetch 100 messages at a time + }) + .Produce(x => x.DefaultTopic(topic)) + .Consume(x => x + .Topic(topic) + .WithConsumer() + .Instances(20) // process messages in parallel + .SubscriptionName(nameof(OutboxBenchmarkTests))); // for AzureSB + + if (_useOutbox) + { + mbb + .UseOutbox(); // All outgoing messages from this bus will go out via an outbox + } + }); + mbb.AddServicesFromAssembly(Assembly.GetExecutingAssembly()); + mbb.AddJsonSerializer(); + mbb.AddOutboxUsingDbContext(opts => + { + opts.PollBatchSize = 100; + opts.PollIdleSleep = TimeSpan.FromSeconds(0.5); + opts.MessageCleanup.Interval = TimeSpan.FromSeconds(10); + opts.MessageCleanup.Age = TimeSpan.FromMinutes(1); + opts.SqlSettings.DatabaseTableName = OutboxTableName; + opts.SqlSettings.DatabaseMigrationsTableName = MigrationsTableName; + }); + mbb.AutoStartConsumersEnabled(false); + }); + + services.AddSingleton>(); + + // Entity Framework setup - application specific EF DbContext + services.AddDbContext(options => options.UseSqlServer(Secrets.Service.PopulateSecrets(Configuration.GetConnectionString("DefaultConnection")))); + } + + private async Task PerformDbOperation(Func action) + { + using var scope = ServiceProvider!.CreateScope(); + var context = scope.ServiceProvider.GetRequiredService(); + await action(context); + } + + [Theory] + [InlineData([true, 1000])] // compare with outbox + [InlineData([false, 1000])] // vs. without outbox + public async Task Given_EventPublisherAndConsumerUsingOutbox_When_BurstOfEventsIsSent_Then_EventsAreConsumedProperly(bool useOutbox, int messageCount) + { + // arrange + _useOutbox = useOutbox; + + await PerformDbOperation(async context => + { + // migrate db + await context.Database.MigrateAsync(); + + // clean outbox from previous test run + await context.Database.EraseTableIfExists(OutboxTableName); + await context.Database.EraseTableIfExists(MigrationsTableName); + }); + + var surnames = new[] { "Doe", "Smith", "Kowalsky" }; + var events = Enumerable.Range(0, messageCount).Select(x => new CustomerCreatedEvent(Guid.NewGuid(), $"John {x:000}", surnames[x % surnames.Length])).ToList(); + var store = ServiceProvider!.GetRequiredService>(); + + // act + + // publish the events in one shot (consumers are not started yet) + var publishTimer = Stopwatch.StartNew(); + + var publishTasks = events + .Select(async ev => + { + var unitOfWorkScope = ServiceProvider!.CreateScope(); + await using (unitOfWorkScope as IAsyncDisposable) + { + var bus = unitOfWorkScope.ServiceProvider.GetRequiredService(); + try + { + await bus.Publish(ev, headers: new Dictionary { ["CustomerId"] = ev.Id }); + } + catch (Exception ex) + { + Logger.LogInformation("Exception occurred while publishing event {Event}: {Message}", ev, ex.Message); + } + } + }) + .ToArray(); + + await Task.WhenAll(publishTasks); + + var publishTimerElapsed = publishTimer.Elapsed; + + // start consumers + await EnsureConsumersStarted(); + + // consume the events from outbox + var consumptionTimer = Stopwatch.StartNew(); + await store.WaitUntilArriving(newMessagesTimeout: 5, expectedCount: events.Count); + + // assert + + var consumeTimerElapsed = consumptionTimer.Elapsed; + + // Log the measured times + Logger.LogInformation("Message Publish took: {Elapsed}", publishTimerElapsed); + Logger.LogInformation("Message Consume took: {Elapsed}", consumeTimerElapsed); + + // Ensure the expected number of events was actually published to ASB and delivered via that channel. + store.Count.Should().Be(events.Count); + } +} diff --git a/src/Tests/SlimMessageBus.Host.Outbox.DbContext.Test/OutboxTests.cs b/src/Tests/SlimMessageBus.Host.Outbox.DbContext.Test/OutboxTests.cs index da53ba1a..6164536c 100644 --- a/src/Tests/SlimMessageBus.Host.Outbox.DbContext.Test/OutboxTests.cs +++ b/src/Tests/SlimMessageBus.Host.Outbox.DbContext.Test/OutboxTests.cs @@ -1,42 +1,13 @@ namespace SlimMessageBus.Host.Outbox.DbContext.Test; -using System.Reflection; - -using Confluent.Kafka; - -using Microsoft.EntityFrameworkCore; -using Microsoft.Extensions.Configuration; -using Microsoft.Extensions.DependencyInjection; -using Microsoft.Extensions.Logging; - -using SecretStore; - -using SlimMessageBus.Host; -using SlimMessageBus.Host.AzureServiceBus; -using SlimMessageBus.Host.Kafka; -using SlimMessageBus.Host.Memory; -using SlimMessageBus.Host.Outbox.DbContext.Test.DataAccess; -using SlimMessageBus.Host.Outbox.Sql; -using SlimMessageBus.Host.Serialization.SystemTextJson; -using SlimMessageBus.Host.Test.Common.IntegrationTest; - [Trait("Category", "Integration")] public class OutboxTests(ITestOutputHelper testOutputHelper) : BaseIntegrationTest(testOutputHelper) { private TransactionType _testParamTransactionType; private BusType _testParamBusType; - public enum TransactionType - { - SqlTransaction, - TarnsactionScope - } - - public enum BusType - { - AzureSB, - Kafka, - } + private static readonly string OutboxTableName = "IntTest_Outbox"; + private static readonly string MigrationsTableName = "IntTest_Migrations"; protected override void SetupServices(ServiceCollection services, IConfigurationRoot configuration) { @@ -98,7 +69,11 @@ protected override void SetupServices(ServiceCollection services, IConfiguration } if (_testParamBusType == BusType.AzureSB) { - mbb.WithProviderServiceBus(cfg => cfg.ConnectionString = Secrets.Service.PopulateSecrets(configuration["Azure:ServiceBus"])); + mbb.WithProviderServiceBus(cfg => + { + cfg.ConnectionString = Secrets.Service.PopulateSecrets(configuration["Azure:ServiceBus"]); + cfg.PrefetchCount = 100; + }); topic = "tests.outbox/customer-events"; } @@ -107,6 +82,7 @@ protected override void SetupServices(ServiceCollection services, IConfiguration .Consume(x => x .Topic(topic) .WithConsumer() + .Instances(20) // process messages in parallel .SubscriptionName(nameof(OutboxTests)) // for AzureSB .KafkaGroup("subscriber")) // for Kafka .UseOutbox(); // All outgoing messages from this bus will go out via an outbox @@ -119,7 +95,8 @@ protected override void SetupServices(ServiceCollection services, IConfiguration opts.PollIdleSleep = TimeSpan.FromSeconds(0.5); opts.MessageCleanup.Interval = TimeSpan.FromSeconds(10); opts.MessageCleanup.Age = TimeSpan.FromMinutes(1); - opts.SqlSettings.DatabaseTableName = "IntTest_Outbox"; + opts.SqlSettings.DatabaseTableName = OutboxTableName; + opts.SqlSettings.DatabaseMigrationsTableName = MigrationsTableName; }); }); @@ -139,10 +116,10 @@ private async Task PerformDbOperation(Func action) public const string InvalidLastname = "Exception"; [Theory] - [InlineData([TransactionType.SqlTransaction, BusType.AzureSB])] - [InlineData([TransactionType.TarnsactionScope, BusType.AzureSB])] - [InlineData([TransactionType.SqlTransaction, BusType.Kafka])] - public async Task Given_CommandHandlerInTransaction_When_ExceptionThrownDuringHandlingRaisedAtTheEnd_Then_TransactionIsRolledBack_And_NoDataSaved_And_NoEventRaised(TransactionType transactionType, BusType busType) + [InlineData([TransactionType.SqlTransaction, BusType.AzureSB, 100])] + [InlineData([TransactionType.TarnsactionScope, BusType.AzureSB, 100])] + [InlineData([TransactionType.SqlTransaction, BusType.Kafka, 100])] + public async Task Given_CommandHandlerInTransaction_When_ExceptionThrownDuringHandlingRaisedAtTheEnd_Then_TransactionIsRolledBack_And_NoDataSaved_And_NoEventRaised(TransactionType transactionType, BusType busType, int messageCount) { // arrange _testParamTransactionType = transactionType; @@ -153,14 +130,9 @@ await PerformDbOperation(async context => // migrate db await context.Database.MigrateAsync(); - //// clean outbox from previous test run - await context.Database.ExecuteSqlRawAsync( - """ - IF EXISTS (SELECT 1 FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA = 'dbo' AND TABLE_NAME = 'IntTest_Outbox') - BEGIN - DELETE FROM dbo.IntTest_Outbox; - END - """); + // clean outbox from previous test run + await context.Database.EraseTableIfExists(OutboxTableName); + await context.Database.EraseTableIfExists(MigrationsTableName); // clean the customers table var cust = await context.Customers.ToListAsync(); @@ -169,7 +141,7 @@ IF EXISTS (SELECT 1 FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA = 'dbo' AN }); var surnames = new[] { "Doe", "Smith", InvalidLastname }; - var commands = Enumerable.Range(0, 100).Select(x => new CreateCustomerCommand($"John {x:000}", surnames[x % surnames.Length])); + var commands = Enumerable.Range(0, messageCount).Select(x => new CreateCustomerCommand($"John {x:000}", surnames[x % surnames.Length])); var validCommands = commands.Where(x => !string.Equals(x.Lastname, InvalidLastname, StringComparison.InvariantCulture)).ToList(); var store = ServiceProvider!.GetRequiredService>(); @@ -180,23 +152,27 @@ IF EXISTS (SELECT 1 FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA = 'dbo' AN store.Clear(); // act - foreach (var cmd in commands) - { - var unitOfWorkScope = ServiceProvider!.CreateScope(); - await using (unitOfWorkScope as IAsyncDisposable) + var sendTasks = commands + .Select(async cmd => { - var bus = unitOfWorkScope.ServiceProvider.GetRequiredService(); - - try + var unitOfWorkScope = ServiceProvider!.CreateScope(); + await using (unitOfWorkScope as IAsyncDisposable) { - var res = await bus.Send(cmd); - } - catch (Exception ex) - { - Logger.LogInformation("Exception occurred while handling cmd {Command}: {Message}", cmd, ex.Message); + var bus = unitOfWorkScope.ServiceProvider.GetRequiredService(); + + try + { + var res = await bus.Send(cmd); + } + catch (Exception ex) + { + Logger.LogInformation("Exception occurred while handling cmd {Command}: {Message}", cmd, ex.Message); + } } - } - } + }) + .ToArray(); + + await Task.WhenAll(sendTasks); await store.WaitUntilArriving(newMessagesTimeout: 5, expectedCount: validCommands.Count); diff --git a/src/Tests/SlimMessageBus.Host.Outbox.DbContext.Test/TransactionType.cs b/src/Tests/SlimMessageBus.Host.Outbox.DbContext.Test/TransactionType.cs new file mode 100644 index 00000000..3bbe511e --- /dev/null +++ b/src/Tests/SlimMessageBus.Host.Outbox.DbContext.Test/TransactionType.cs @@ -0,0 +1,7 @@ +namespace SlimMessageBus.Host.Outbox.DbContext.Test; + +public enum TransactionType +{ + SqlTransaction, + TarnsactionScope +} diff --git a/src/Tests/SlimMessageBus.Host.Outbox.DbContext.Test/Usings.cs b/src/Tests/SlimMessageBus.Host.Outbox.DbContext.Test/Usings.cs index 58501a46..4728b0e1 100644 --- a/src/Tests/SlimMessageBus.Host.Outbox.DbContext.Test/Usings.cs +++ b/src/Tests/SlimMessageBus.Host.Outbox.DbContext.Test/Usings.cs @@ -1,3 +1,24 @@ +global using System.Diagnostics; +global using System.Reflection; + +global using Confluent.Kafka; + +global using FluentAssertions; + +global using Microsoft.EntityFrameworkCore; +global using Microsoft.Extensions.Configuration; +global using Microsoft.Extensions.DependencyInjection; +global using Microsoft.Extensions.Logging; + +global using SecretStore; + +global using SlimMessageBus.Host.AzureServiceBus; +global using SlimMessageBus.Host.Kafka; +global using SlimMessageBus.Host.Memory; +global using SlimMessageBus.Host.Outbox.DbContext.Test.DataAccess; +global using SlimMessageBus.Host.Outbox.Sql; +global using SlimMessageBus.Host.Serialization.SystemTextJson; +global using SlimMessageBus.Host.Test.Common.IntegrationTest; + global using Xunit; global using Xunit.Abstractions; -global using FluentAssertions; \ No newline at end of file diff --git a/src/Tests/SlimMessageBus.Host.Test.Common/IntegrationTest/BaseIntegrationTest.cs b/src/Tests/SlimMessageBus.Host.Test.Common/IntegrationTest/BaseIntegrationTest.cs index 00435556..70a335aa 100644 --- a/src/Tests/SlimMessageBus.Host.Test.Common/IntegrationTest/BaseIntegrationTest.cs +++ b/src/Tests/SlimMessageBus.Host.Test.Common/IntegrationTest/BaseIntegrationTest.cs @@ -83,7 +83,7 @@ protected async Task EnsureConsumersStarted() var consumerControl = ServiceProvider.GetRequiredService(); // ensure the consumers are warm - while (!consumerControl.IsStarted && timeout.ElapsedMilliseconds < 5000) await Task.Delay(200); + while (!consumerControl.IsStarted && timeout.ElapsedMilliseconds < 5000) await Task.Delay(100); } public Task InitializeAsync() diff --git a/src/Tests/SlimMessageBus.Host.Test.Common/MoqExtensions.cs b/src/Tests/SlimMessageBus.Host.Test.Common/MoqExtensions.cs new file mode 100644 index 00000000..85fc59b7 --- /dev/null +++ b/src/Tests/SlimMessageBus.Host.Test.Common/MoqExtensions.cs @@ -0,0 +1,31 @@ +namespace SlimMessageBus.Host.Test.Common; + +using System.Diagnostics; +using System.Linq.Expressions; + +public static class MoqExtensions +{ + public static async Task VerifyWithRetry(this Mock mock, TimeSpan timeout, Expression> expression, Times times) + where T : class + { + var stopwatch = Stopwatch.StartNew(); + do + { + try + { + mock.Verify(expression, times); + return; + } + catch (MockException) + { + if (stopwatch.Elapsed > timeout) + { + // when timed out rethrow the exception + throw; + } + // else keep on repeating + await Task.Delay(50); + } + } while (true); + } +} diff --git a/src/Tests/SlimMessageBus.Host.Test/Consumer/ConsumerInstanceMessageProcessorTest.cs b/src/Tests/SlimMessageBus.Host.Test/Consumer/ConsumerInstanceMessageProcessorTest.cs index 7b674f0c..50a6c09e 100644 --- a/src/Tests/SlimMessageBus.Host.Test/Consumer/ConsumerInstanceMessageProcessorTest.cs +++ b/src/Tests/SlimMessageBus.Host.Test/Consumer/ConsumerInstanceMessageProcessorTest.cs @@ -13,7 +13,7 @@ public ConsumerInstanceMessageProcessorTest() { _busMock = new MessageBusMock(); _messageProviderMock = new Mock>(); - _transportMessage = Array.Empty(); + _transportMessage = []; _topic = "topic"; var messageBusSettings = new MessageBusSettings(); _handlerSettings = new HandlerBuilder(messageBusSettings).Topic(_topic).WithHandler>().ConsumerSettings; diff --git a/src/Tests/SlimMessageBus.Host.Test/GlobalUsings.cs b/src/Tests/SlimMessageBus.Host.Test/GlobalUsings.cs index 6567bd04..a8f3ecbb 100644 --- a/src/Tests/SlimMessageBus.Host.Test/GlobalUsings.cs +++ b/src/Tests/SlimMessageBus.Host.Test/GlobalUsings.cs @@ -10,5 +10,6 @@ global using SlimMessageBus.Host.Interceptor; global using SlimMessageBus.Host.Serialization; +global using SlimMessageBus.Host.Serialization.Json; global using Xunit; diff --git a/src/Tests/SlimMessageBus.Host.Test/Hybrid/HybridMessageBusTest.cs b/src/Tests/SlimMessageBus.Host.Test/Hybrid/HybridMessageBusTest.cs index 5ddf3fe9..f6a18972 100644 --- a/src/Tests/SlimMessageBus.Host.Test/Hybrid/HybridMessageBusTest.cs +++ b/src/Tests/SlimMessageBus.Host.Test/Hybrid/HybridMessageBusTest.cs @@ -38,6 +38,7 @@ public HybridMessageBusTest() _serviceProviderMock.Setup(x => x.GetService(typeof(IMessageSerializer))).Returns(_messageSerializerMock.Object); _serviceProviderMock.Setup(x => x.GetService(typeof(IMessageTypeResolver))).Returns(new AssemblyQualifiedNameMessageTypeResolver()); _serviceProviderMock.Setup(x => x.GetService(typeof(ILoggerFactory))).Returns(_loggerFactoryMock.Object); + _serviceProviderMock.Setup(x => x.GetService(typeof(IEnumerable))).Returns(Array.Empty()); _loggerFactoryMock.Setup(x => x.CreateLogger(It.IsAny())).Returns(_loggerMock.Object); @@ -47,7 +48,7 @@ public HybridMessageBusTest() mbb.Produce(x => x.DefaultTopic("topic2")); mbb.WithProvider(mbs => { - _bus1Mock = new Mock(new[] { mbs }); + _bus1Mock = new Mock([mbs]); _bus1Mock.SetupGet(x => x.Settings).Returns(mbs); _bus1Mock.Setup(x => x.ProducePublish(It.IsAny(), It.IsAny(), It.IsAny>(), It.IsAny(), It.IsAny())).Returns(Task.CompletedTask); diff --git a/src/Tests/SlimMessageBus.Host.Test/MessageBusBaseTests.cs b/src/Tests/SlimMessageBus.Host.Test/MessageBusBaseTests.cs index f15dcac9..c6d660d8 100644 --- a/src/Tests/SlimMessageBus.Host.Test/MessageBusBaseTests.cs +++ b/src/Tests/SlimMessageBus.Host.Test/MessageBusBaseTests.cs @@ -1,23 +1,5 @@ namespace SlimMessageBus.Host.Test; - -using SlimMessageBus.Host; -using SlimMessageBus.Host.Interceptor; -using SlimMessageBus.Host.Serialization; -using SlimMessageBus.Host.Serialization.Json; - -public class RequestA : IRequest -{ - public string Id { get; set; } = Guid.NewGuid().ToString(); -} - -public class ResponseA -{ - public string Id { get; set; } -} - -public class RequestB : IRequest { } - -public class ResponseB { } +using SlimMessageBus.Host.Test.Common; public class MessageBusBaseTests : IDisposable { @@ -40,7 +22,7 @@ public MessageBusBaseTests() _timeZero = DateTimeOffset.Now; _timeNow = _timeZero; - _producedMessages = new List(); + _producedMessages = []; _serviceProviderMock = new Mock(); _serviceProviderMock.Setup(x => x.GetService(typeof(IMessageSerializer))).Returns(new JsonMessageSerializer()); @@ -108,6 +90,71 @@ public void When_Create_Given_ConfigurationThatDeclaresSameMessageTypeMoreThanOn .WithMessage("*was declared more than once*"); } + [Fact] + public async Task When_Create_Then_BusLifecycleCreatedIsSentToRegisteredInterceptors() + { + // arrange + var busLifecycleInterceptorMock = new Mock(); + + _serviceProviderMock + .Setup(x => x.GetService(typeof(IEnumerable))) + .Returns(new IMessageBusLifecycleInterceptor[] { busLifecycleInterceptorMock.Object }); + + // act + BusBuilder.Build(); + + // assert + await busLifecycleInterceptorMock + .VerifyWithRetry( + // give some time for the fire & forget task to complete + TimeSpan.FromSeconds(2), + x => x.OnBusLifecycle(MessageBusLifecycleEventType.Created, It.IsAny()), + Times.Once()); + } + + [Theory] + [InlineData(true)] + [InlineData(false)] + public async Task When_Produce_Given_LongRunningCreateInterceptor_Then_ProduceWaitsUntilInterceptorFinishes(bool isPublish) + { + // arrange + var longRunnitIntitTask = Task.Delay(4000); + + var busLifecycleInterceptorMock = new Mock(); + busLifecycleInterceptorMock + .Setup(x => x.OnBusLifecycle(MessageBusLifecycleEventType.Created, It.IsAny())) + .Returns(() => longRunnitIntitTask); + + _serviceProviderMock + .Setup(x => x.GetService(typeof(IEnumerable))) + .Returns(new IMessageBusLifecycleInterceptor[] { busLifecycleInterceptorMock.Object }); + + // setup responses for requests + Bus.OnReply = (type, topic, request) => + { + if (request is RequestA req) + { + return new ResponseA { Id = req.Id }; + } + return null; + }; + + BusBuilder.Build(); + + // act + if (isPublish) + { + await Bus.ProducePublish(new RequestA()); + } + else + { + await Bus.ProduceSend(new RequestA()); + } + + // assert + longRunnitIntitTask.IsCompletedSuccessfully.Should().BeTrue(); + } + [Fact] public async Task When_NoTimeoutProvided_Then_TakesDefaultTimeoutForRequestTypeAsync() { diff --git a/src/Tests/SlimMessageBus.Host.Test/MessageBusTested.cs b/src/Tests/SlimMessageBus.Host.Test/MessageBusTested.cs index b0f45386..b8654133 100644 --- a/src/Tests/SlimMessageBus.Host.Test/MessageBusTested.cs +++ b/src/Tests/SlimMessageBus.Host.Test/MessageBusTested.cs @@ -36,7 +36,9 @@ protected internal override Task OnStop() } protected override async Task ProduceToTransport(object message, string path, byte[] messagePayload, IDictionary messageHeaders, IMessageBusTarget targetBus, CancellationToken cancellationToken = default) - { + { + await EnsureInitFinished(); + var messageType = message.GetType(); OnProduced(messageType, path, message); diff --git a/src/Tests/SlimMessageBus.Host.Test/SampleMessages.cs b/src/Tests/SlimMessageBus.Host.Test/SampleMessages.cs index 8893349d..41808898 100644 --- a/src/Tests/SlimMessageBus.Host.Test/SampleMessages.cs +++ b/src/Tests/SlimMessageBus.Host.Test/SampleMessages.cs @@ -59,4 +59,18 @@ public interface ICustomConsumer Task HandleAMessageWithAContext(T message, IConsumerContext consumerContext, CancellationToken cancellationToken); Task MethodThatHasParamatersThatCannotBeSatisfied(T message, DateTimeOffset dateTimeOffset, CancellationToken cancellationToken); -} \ No newline at end of file +} + +public class RequestA : IRequest +{ + public string Id { get; set; } = Guid.NewGuid().ToString(); +} + +public class ResponseA +{ + public string Id { get; set; } +} + +public class RequestB : IRequest { } + +public class ResponseB { } \ No newline at end of file diff --git a/src/secrets.txt.sample b/src/secrets.txt.sample index f335f86e..00b53da1 100644 --- a/src/secrets.txt.sample +++ b/src/secrets.txt.sample @@ -13,10 +13,10 @@ kafka_username=user kafka_password=password mqtt_server=localhost +mqtt_secure=false mqtt_port=1883 mqtt_username= mqtt_password= -mqtt_secure=false sqlserver_connectionstring=Server=localhost;Initial Catalog=SlimMessageBus_Outbox;User ID=sa;Password=SuperSecretP@55word;TrustServerCertificate=true;MultipleActiveResultSets=true;