From 053438ab21b48d2a1cb47cf6aba9d4e0304eb331 Mon Sep 17 00:00:00 2001 From: Leszek Pomianowski <13592821+pomianowski@users.noreply.github.com> Date: Fri, 21 Feb 2025 17:05:14 +0100 Subject: [PATCH] Add new background queue --- Directory.Build.props | 18 +- Directory.Packages.props | 9 +- LICENSE | 2 +- LICENSE.md | 2 +- .../ContainerBuilderExtensions.cs | 3 + .../ReflectionEventing.Autofac.csproj | 4 +- .../EventBusInstaller.cs | 2 + .../ReflectionEventing.Castle.Windsor.csproj | 4 +- .../Events/AsyncQueuedEvent.cs | 8 + .../MainWindow.xaml | 34 +++- .../ReflectionEventing.Demo.Wpf.csproj | 2 +- .../Services/BackgroundTickService.cs | 4 +- .../ViewModels/MainWindowViewModel.cs | 27 ++- .../QueueProcessorOptionsProvider.cs | 28 +++ .../DependencyInjectionQueueProcessor.cs | 160 +++++++++++++++++ .../GlobalUsings.cs | 8 + ...lectionEventing.DependencyInjection.csproj | 20 ++- .../ServiceCollectionExtensions.cs | 45 +++++ .../EventBusModule.cs | 3 + .../ReflectionEventing.Ninject.csproj | 8 +- .../ReflectionEventing.Unity.csproj | 4 +- .../UnityContainerExtensions.cs | 5 + src/ReflectionEventing/EventBus.cs | 45 ++++- .../EventBusBuilderOptions.cs | 18 ++ src/ReflectionEventing/EventBusExtensions.cs | 30 +++- src/ReflectionEventing/GlobalUsings.cs | 4 + src/ReflectionEventing/IEventBus.cs | 21 ++- src/ReflectionEventing/Queues/EventsQueue.cs | 41 +++++ src/ReflectionEventing/Queues/FailedEvent.cs | 13 ++ src/ReflectionEventing/Queues/IEventsQueue.cs | 40 +++++ .../ReflectionEventing.csproj | 14 +- ...eflectionEventing.Autofac.UnitTests.csproj | 4 +- .../EventBusInstallerTests.cs | 9 +- .../EventBusTests.cs | 13 +- ...onEventing.Castle.Windsor.UnitTests.csproj | 4 +- .../DependencyInjectionQueueProcessorTests.cs | 163 ++++++++++++++++++ .../GlobalUsings.cs | 2 + ...nting.DependencyInjection.UnitTests.csproj | 7 +- .../ServiceCollectionExtensionsTests.cs | 2 +- ...eflectionEventing.Ninject.UnitTests.csproj | 4 +- .../EventBusTests.cs | 14 +- .../HashedConsumerTypesProviderTests.cs | 17 +- .../ReflectionEventing.UnitTests.csproj | 4 +- .../ReflectionEventing.Unity.UnitTests.csproj | 6 +- 44 files changed, 794 insertions(+), 81 deletions(-) create mode 100644 src/ReflectionEventing.Demo.Wpf/Events/AsyncQueuedEvent.cs create mode 100644 src/ReflectionEventing.DependencyInjection/Configuration/QueueProcessorOptionsProvider.cs create mode 100644 src/ReflectionEventing.DependencyInjection/DependencyInjectionQueueProcessor.cs create mode 100644 src/ReflectionEventing/Queues/EventsQueue.cs create mode 100644 src/ReflectionEventing/Queues/FailedEvent.cs create mode 100644 src/ReflectionEventing/Queues/IEventsQueue.cs create mode 100644 tests/ReflectionEventing.DependencyInjection.UnitTests/DependencyInjectionQueueProcessorTests.cs diff --git a/Directory.Build.props b/Directory.Build.props index a3b2cd8..fc880f6 100644 --- a/Directory.Build.props +++ b/Directory.Build.props @@ -6,8 +6,8 @@ - 3.1.0 - 3.0.0 + 4.0.0 + 4.0.0 @@ -37,7 +37,7 @@ true - 12.0 + 13.0 enable diff --git a/src/ReflectionEventing.Unity/ReflectionEventing.Unity.csproj b/src/ReflectionEventing.Unity/ReflectionEventing.Unity.csproj index dbe03fc..7ff6966 100644 --- a/src/ReflectionEventing.Unity/ReflectionEventing.Unity.csproj +++ b/src/ReflectionEventing.Unity/ReflectionEventing.Unity.csproj @@ -2,7 +2,7 @@ ReflectionEventing.Unity - netstandard2.0;netstandard2.1;net462;net6.0;net8.0 + net9.0;net8.0;net6.0;netstandard2.0;net462;net472 true true true @@ -10,7 +10,7 @@ Unity container extensions with ReflectionEventing, which promotes better Inversion of Control (IoC), reducing coupling and enhancing the modularity and flexibility of your applications. - + true true Speed diff --git a/src/ReflectionEventing.Unity/UnityContainerExtensions.cs b/src/ReflectionEventing.Unity/UnityContainerExtensions.cs index b205a3b..3ca58e0 100644 --- a/src/ReflectionEventing.Unity/UnityContainerExtensions.cs +++ b/src/ReflectionEventing.Unity/UnityContainerExtensions.cs @@ -3,6 +3,7 @@ // Copyright (C) Leszek Pomianowski and ReflectionEventing Contributors. // All Rights Reserved. +using ReflectionEventing.Queues; using Unity; using Unity.Lifetime; @@ -37,6 +38,10 @@ Action configure new ContainerControlledLifetimeManager() ); + _ = container.RegisterType( + new ContainerControlledLifetimeManager() + ); + _ = container.RegisterType( new HierarchicalLifetimeManager() ); diff --git a/src/ReflectionEventing/EventBus.cs b/src/ReflectionEventing/EventBus.cs index a12dc17..5c62f7d 100644 --- a/src/ReflectionEventing/EventBus.cs +++ b/src/ReflectionEventing/EventBus.cs @@ -3,6 +3,8 @@ // Copyright (C) Leszek Pomianowski and ReflectionEventing Contributors. // All Rights Reserved. +using ReflectionEventing.Queues; + namespace ReflectionEventing; /// @@ -13,13 +15,31 @@ namespace ReflectionEventing; /// public class EventBus( IConsumerProvider consumerProviders, - IConsumerTypesProvider consumerTypesProvider + IConsumerTypesProvider consumerTypesProvider, + IEventsQueue queue ) : IEventBus { + private static readonly ActivitySource ActivitySource = new("ReflectionEventing.EventBus"); + + private static readonly Meter Meter = new("ReflectionEventing.EventBus"); + + private static readonly Counter SentCounter = Meter.CreateCounter("bus.sent"); + + private static readonly Counter PublishedCounter = Meter.CreateCounter( + "bus.published" + ); + /// - public async Task PublishAsync(TEvent eventItem, CancellationToken cancellationToken) + public async Task SendAsync( + TEvent eventItem, + CancellationToken cancellationToken = default + ) where TEvent : class { + using Activity? activity = ActivitySource.StartActivity(ActivityKind.Producer); + + activity?.AddTag("co.lepo.reflection.eventing.message", typeof(TEvent).Name); + if (eventItem is null) { throw new ArgumentNullException(nameof(eventItem)); @@ -38,5 +58,26 @@ public async Task PublishAsync(TEvent eventItem, CancellationToken cance } await Task.WhenAll(tasks).ConfigureAwait(false); + + SentCounter.Add(1, new KeyValuePair("message_type", eventType.Name)); + } + + /// + public async Task PublishAsync( + TEvent eventItem, + CancellationToken cancellationToken = default + ) + where TEvent : class + { + using Activity? activity = ActivitySource.StartActivity(ActivityKind.Producer); + + activity?.AddTag("co.lepo.reflection.eventing.message", typeof(TEvent).Name); + + await queue.EnqueueAsync(eventItem, cancellationToken); + + PublishedCounter.Add( + 1, + new KeyValuePair("message_type", typeof(TEvent).Name) + ); } } diff --git a/src/ReflectionEventing/EventBusBuilderOptions.cs b/src/ReflectionEventing/EventBusBuilderOptions.cs index 58aa90f..65d47dd 100644 --- a/src/ReflectionEventing/EventBusBuilderOptions.cs +++ b/src/ReflectionEventing/EventBusBuilderOptions.cs @@ -18,4 +18,22 @@ public class EventBusBuilderOptions /// The default value is false. /// public bool UseEventPolymorphism { get; set; } = false; + + /// + /// Gets or sets the rate at which the event queue is processed. + /// The default value is 20ms. + /// + /// + /// Adjust this value to control how frequently the event queue is processed. + /// + public TimeSpan QueueTickRate { get; set; } = TimeSpan.FromMilliseconds(20); // NOTE: There are 10,000 ticks in a millisecond. + + /// + /// Gets or sets the rate at which the error queue is processed when default queue consumption fails. + /// The default value is 20ms. + /// + /// + /// Adjust this value to control how frequently the error queue is processed. + /// + public TimeSpan ErrorTickRate { get; set; } = TimeSpan.FromMilliseconds(20); // NOTE: There are 10,000 ticks in a millisecond. } diff --git a/src/ReflectionEventing/EventBusExtensions.cs b/src/ReflectionEventing/EventBusExtensions.cs index a2678c4..0dedc07 100644 --- a/src/ReflectionEventing/EventBusExtensions.cs +++ b/src/ReflectionEventing/EventBusExtensions.cs @@ -10,13 +10,41 @@ namespace ReflectionEventing; /// public static class EventBusExtensions { + /// + /// Sends the specified event synchronously. + /// + /// The type of the event to publish. + /// The event bus to extend. + /// The event to publish. + [Obsolete($"May cause deadlock on UI threads, use {nameof(IEventBus.SendAsync)} instead.")] + public static void Send(this IEventBus eventBus, TEvent eventItem) + where TEvent : class + { + using CancellationTokenSource cancellationSource = new(); + + Task.Run( + () => + { + // ReSharper disable once AccessToDisposedClosure + eventBus + .SendAsync(eventItem, cancellationSource.Token) + .ConfigureAwait(false) + .GetAwaiter() + .GetResult(); + }, + cancellationSource.Token + ) + .GetAwaiter() + .GetResult(); + } + /// /// Publishes the specified event synchronously. /// /// The type of the event to publish. /// The event bus to extend. /// The event to publish. - [Obsolete($"May cause deadlock on UI threads, use {nameof(IEventBus.PublishAsync)} instead.")] + [Obsolete($"May cause deadlock on UI threads, use {nameof(IEventBus.SendAsync)} instead.")] public static void Publish(this IEventBus eventBus, TEvent eventItem) where TEvent : class { diff --git a/src/ReflectionEventing/GlobalUsings.cs b/src/ReflectionEventing/GlobalUsings.cs index b59a06c..f1a166a 100644 --- a/src/ReflectionEventing/GlobalUsings.cs +++ b/src/ReflectionEventing/GlobalUsings.cs @@ -4,9 +4,13 @@ // All Rights Reserved. global using System; +global using System.Collections.Concurrent; global using System.Collections.Generic; +global using System.Diagnostics; global using System.Diagnostics.CodeAnalysis; +global using System.Diagnostics.Metrics; global using System.Linq; global using System.Reflection; global using System.Threading; +global using System.Threading.Channels; global using System.Threading.Tasks; diff --git a/src/ReflectionEventing/IEventBus.cs b/src/ReflectionEventing/IEventBus.cs index 2dea8e3..5b55841 100644 --- a/src/ReflectionEventing/IEventBus.cs +++ b/src/ReflectionEventing/IEventBus.cs @@ -11,16 +11,29 @@ namespace ReflectionEventing; public interface IEventBus { /// - /// Publishes the specified event asynchronously. + /// Sends the specified event asynchronously within the current scope, waiting for its execution. /// - /// The type of the event to publish. - /// The event to publish. + /// The type of the event to send. + /// The event to send. /// A cancellation token that can be used to cancel the operation. /// A task that represents the asynchronous operation. /// /// This method gets the consumers for the specified event type from the consumer provider and then uses the service provider to get the required service for each consumer. /// Each consumer is then used to consume the event asynchronously. /// - Task PublishAsync(TEvent eventItem, CancellationToken cancellationToken) + Task SendAsync(TEvent eventItem, CancellationToken cancellationToken = default) + where TEvent : class; + + /// + /// Adds the specified event to the queue.Another scope will take over execution as configured. + /// + /// The type of the event to publish. + /// The event to publish. + /// A cancellation token that can be used to cancel the operation. + /// A task that represents the asynchronous operation. + /// + /// The method only adds the event to the execution queue, it does not wait for its successful execution. + /// + Task PublishAsync(TEvent eventItem, CancellationToken cancellationToken = default) where TEvent : class; } diff --git a/src/ReflectionEventing/Queues/EventsQueue.cs b/src/ReflectionEventing/Queues/EventsQueue.cs new file mode 100644 index 0000000..a511fb6 --- /dev/null +++ b/src/ReflectionEventing/Queues/EventsQueue.cs @@ -0,0 +1,41 @@ +// This Source Code Form is subject to the terms of the MIT License. +// If a copy of the MIT was not distributed with this file, You can obtain one at https://opensource.org/licenses/MIT. +// Copyright (C) Leszek Pomianowski and ReflectionEventing Contributors. +// All Rights Reserved. + +namespace ReflectionEventing.Queues; + +public class EventsQueue : IEventsQueue +{ + private readonly Channel events = Channel.CreateUnbounded(); + + private readonly ConcurrentQueue errorQueue = new(); + + /// + public virtual async Task EnqueueAsync( + TEvent @event, + CancellationToken cancellationToken = default + ) + where TEvent : class + { + await events.Writer.WriteAsync(@event, cancellationToken); + } + + /// + public ChannelReader GetReader() + { + return events.Reader; + } + + /// + public void EnqueueError(object @event, Exception exception) + { + errorQueue.Enqueue(new FailedEvent { Data = @event, Exception = exception }); + } + + /// + public IEnumerable GetErrorQueue() + { + return errorQueue; + } +} diff --git a/src/ReflectionEventing/Queues/FailedEvent.cs b/src/ReflectionEventing/Queues/FailedEvent.cs new file mode 100644 index 0000000..64ce27e --- /dev/null +++ b/src/ReflectionEventing/Queues/FailedEvent.cs @@ -0,0 +1,13 @@ +// This Source Code Form is subject to the terms of the MIT License. +// If a copy of the MIT was not distributed with this file, You can obtain one at https://opensource.org/licenses/MIT. +// Copyright (C) Leszek Pomianowski and ReflectionEventing Contributors. +// All Rights Reserved. + +namespace ReflectionEventing.Queues; + +public sealed record FailedEvent +{ + public required object Data { get; init; } + + public required Exception Exception { get; init; } +} diff --git a/src/ReflectionEventing/Queues/IEventsQueue.cs b/src/ReflectionEventing/Queues/IEventsQueue.cs new file mode 100644 index 0000000..382f726 --- /dev/null +++ b/src/ReflectionEventing/Queues/IEventsQueue.cs @@ -0,0 +1,40 @@ +// This Source Code Form is subject to the terms of the MIT License. +// If a copy of the MIT was not distributed with this file, You can obtain one at https://opensource.org/licenses/MIT. +// Copyright (C) Leszek Pomianowski and ReflectionEventing Contributors. +// All Rights Reserved. + +namespace ReflectionEventing.Queues; + +/// +/// Defines a contract for an event queue that supports asynchronous operations for appending and retrieving events. +/// +public interface IEventsQueue +{ + /// + /// Appends an event to the queue asynchronously. + /// + /// The event to append to the queue. + /// A token to monitor for cancellation requests. + /// A task that represents the asynchronous append operation. + Task EnqueueAsync(TEvent @event, CancellationToken cancellationToken = default) + where TEvent : class; + + /// + /// Appends an event that failed processing to the error queue. + /// + /// The event that failed processing. + /// The exception that occurred during processing. + void EnqueueError(object @event, Exception exception); + + /// + /// Gets a reader for the event queue. + /// + /// A that can be used to read events from the queue. + ChannelReader GetReader(); + + /// + /// Gets the events that failed processing from the error queue. + /// + /// An of events that failed processing. + IEnumerable GetErrorQueue(); +} diff --git a/src/ReflectionEventing/ReflectionEventing.csproj b/src/ReflectionEventing/ReflectionEventing.csproj index 3a3b1be..8efcfe7 100644 --- a/src/ReflectionEventing/ReflectionEventing.csproj +++ b/src/ReflectionEventing/ReflectionEventing.csproj @@ -1,20 +1,25 @@ - + ReflectionEventing - netstandard2.0;netstandard2.1;net462;net6.0;net8.0 + net9.0;net8.0;net6.0;netstandard2.0;net462;net472 true true true By leveraging the power of Dependency Injection (DI) and eventing, ReflectionEventing promotes better Inversion of Control (IoC), reducing coupling and enhancing the modularity and flexibility of your applications. - + true true Speed + + + + + @@ -26,9 +31,12 @@ System.Diagnostics.CodeAnalysis.NotNullWhenAttribute; System.Diagnostics.CodeAnalysis.RequiresUnreferencedCodeAttribute; System.Diagnostics.CodeAnalysis.UnconditionalSuppressMessageAttribute; + System.Diagnostics.CodeAnalysis.SetsRequiredMembersAttribute; System.Runtime.CompilerServices.CallerArgumentExpressionAttribute; System.Runtime.CompilerServices.IsExternalInit; System.Runtime.CompilerServices.SkipLocalsInitAttribute; + System.Runtime.CompilerServices.RequiredMemberAttribute; + System.Runtime.CompilerServices.CompilerFeatureRequiredAttribute; diff --git a/tests/ReflectionEventing.Autofac.UnitTests/ReflectionEventing.Autofac.UnitTests.csproj b/tests/ReflectionEventing.Autofac.UnitTests/ReflectionEventing.Autofac.UnitTests.csproj index 9366fbc..51c7a6e 100644 --- a/tests/ReflectionEventing.Autofac.UnitTests/ReflectionEventing.Autofac.UnitTests.csproj +++ b/tests/ReflectionEventing.Autofac.UnitTests/ReflectionEventing.Autofac.UnitTests.csproj @@ -1,13 +1,13 @@ - net8.0 + net9.0 false true - + diff --git a/tests/ReflectionEventing.Castle.Windsor.UnitTests/EventBusInstallerTests.cs b/tests/ReflectionEventing.Castle.Windsor.UnitTests/EventBusInstallerTests.cs index 6ff0f77..b0dfb74 100644 --- a/tests/ReflectionEventing.Castle.Windsor.UnitTests/EventBusInstallerTests.cs +++ b/tests/ReflectionEventing.Castle.Windsor.UnitTests/EventBusInstallerTests.cs @@ -18,11 +18,10 @@ public void Install_RegistersServicesAndAddsConsumer() _ = container.Register(Component.For().LifestyleScoped()); - EventBusInstaller installer = - new(builder => - { - _ = builder.AddConsumer(); - }); + EventBusInstaller installer = new(builder => + { + _ = builder.AddConsumer(); + }); installer.Install(container, null!); diff --git a/tests/ReflectionEventing.Castle.Windsor.UnitTests/EventBusTests.cs b/tests/ReflectionEventing.Castle.Windsor.UnitTests/EventBusTests.cs index 8ad3c10..089d3d5 100644 --- a/tests/ReflectionEventing.Castle.Windsor.UnitTests/EventBusTests.cs +++ b/tests/ReflectionEventing.Castle.Windsor.UnitTests/EventBusTests.cs @@ -18,22 +18,21 @@ public EventBusTests() _container = new WindsorContainer(); _ = _container.Register(Component.For().LifestyleScoped()); - EventBusInstaller installer = - new(builder => - { - _ = builder.AddConsumer(); - }); + EventBusInstaller installer = new(builder => + { + _ = builder.AddConsumer(); + }); installer.Install(_container, null!); } [Fact] - public async Task PublishAsync_ShouldCallConsumeAsyncOnAllConsumers() + public async Task SendAsync_ShouldCallConsumeAsyncOnAllConsumers() { using IDisposable scope = _container.BeginScope(); IEventBus eventBus = _container.Resolve(); - await eventBus.PublishAsync(new TestEvent(), CancellationToken.None); + await eventBus.SendAsync(new TestEvent(), CancellationToken.None); _ = _container.Resolve().ReceivedEvents.Should().Be(1); } diff --git a/tests/ReflectionEventing.Castle.Windsor.UnitTests/ReflectionEventing.Castle.Windsor.UnitTests.csproj b/tests/ReflectionEventing.Castle.Windsor.UnitTests/ReflectionEventing.Castle.Windsor.UnitTests.csproj index 4134d20..554dc11 100644 --- a/tests/ReflectionEventing.Castle.Windsor.UnitTests/ReflectionEventing.Castle.Windsor.UnitTests.csproj +++ b/tests/ReflectionEventing.Castle.Windsor.UnitTests/ReflectionEventing.Castle.Windsor.UnitTests.csproj @@ -1,13 +1,13 @@ - net8.0 + net9.0 false true - + diff --git a/tests/ReflectionEventing.DependencyInjection.UnitTests/DependencyInjectionQueueProcessorTests.cs b/tests/ReflectionEventing.DependencyInjection.UnitTests/DependencyInjectionQueueProcessorTests.cs new file mode 100644 index 0000000..29b782e --- /dev/null +++ b/tests/ReflectionEventing.DependencyInjection.UnitTests/DependencyInjectionQueueProcessorTests.cs @@ -0,0 +1,163 @@ +// This Source Code Form is subject to the terms of the MIT License. +// If a copy of the MIT was not distributed with this file, You can obtain one at https://opensource.org/licenses/MIT. +// Copyright (C) Leszek Pomianowski and ReflectionEventing Contributors. +// All Rights Reserved. + +namespace ReflectionEventing.DependencyInjection.UnitTests; + +public sealed class DependencyInjectionQueueProcessorTests +{ + [Fact] + public async Task PublishAsync_ShouldNotSendEventsToWrongCustomers() + { + using IHost host = Host.CreateDefaultBuilder() + .ConfigureServices( + (_, services) => + { + services.AddEventBus(builder => + { + builder.Options.QueueTickRate = TimeSpan.FromTicks(10_000); + builder.Options.ErrorTickRate = TimeSpan.FromTicks(10_000); + builder.Options.UseEventPolymorphism = true; + + builder.AddSingletonConsumer(); + }); + } + ) + .Build(); + + await host.StartAsync(); + + IEventBus bus = host.Services.GetRequiredService(); + + await bus.PublishAsync(new VoidEvent()); + + await Task.Delay(TimeSpan.FromSeconds(1)); + + TestConsumer testConsumer = host.Services.GetRequiredService(); + + testConsumer.TestEventConsumed.Should().BeFalse(); + testConsumer.OtherEventConsumed.Should().BeFalse(); + testConsumer.AsyncQueuedEventConsumed.Should().BeFalse(); + + await host.StopAsync(); + } + + [Fact] + public async Task PublishAsync_ShouldSendOnlyOneEvent() + { + using IHost host = Host.CreateDefaultBuilder() + .ConfigureServices( + (_, services) => + { + services.AddEventBus(builder => + { + builder.Options.QueueTickRate = TimeSpan.FromTicks(10_000); + builder.Options.ErrorTickRate = TimeSpan.FromTicks(10_000); + builder.Options.UseEventPolymorphism = true; + + builder.AddSingletonConsumer(); + }); + } + ) + .Build(); + + await host.StartAsync(); + + IEventBus bus = host.Services.GetRequiredService(); + + await bus.PublishAsync(new OtherEvent()); + + await Task.Delay(TimeSpan.FromSeconds(1)); + + TestConsumer testConsumer = host.Services.GetRequiredService(); + + testConsumer.TestEventConsumed.Should().BeFalse(); + testConsumer.OtherEventConsumed.Should().BeTrue(); + testConsumer.AsyncQueuedEventConsumed.Should().BeFalse(); + + await host.StopAsync(); + } + + [Fact] + public async Task PublishAsync_ShouldProperlyAddEventsToQueue() + { + using IHost host = Host.CreateDefaultBuilder() + .ConfigureServices( + (_, services) => + { + services.AddEventBus(builder => + { + builder.Options.QueueTickRate = TimeSpan.FromTicks(10_000); + builder.Options.ErrorTickRate = TimeSpan.FromTicks(10_000); + builder.Options.UseEventPolymorphism = true; + + builder.AddSingletonConsumer(); + }); + } + ) + .Build(); + + await host.StartAsync(); + + IEventBus bus = host.Services.GetRequiredService(); + + await bus.PublishAsync(new TestEvent()); + await bus.PublishAsync(new OtherEvent()); + await bus.PublishAsync(new AsyncQueuedEvent()); + + await Task.Delay(TimeSpan.FromSeconds(1)); + + TestConsumer testConsumer = host.Services.GetRequiredService(); + + testConsumer.TestEventConsumed.Should().BeTrue(); + testConsumer.OtherEventConsumed.Should().BeTrue(); + testConsumer.AsyncQueuedEventConsumed.Should().BeTrue(); + + await host.StopAsync(); + } + + public record TestEvent; + + public record OtherEvent; + + public record AsyncQueuedEvent; + + public record VoidEvent; + + public class TestConsumer + : IConsumer, + IConsumer, + IConsumer + { + public bool TestEventConsumed { get; private set; } + public bool OtherEventConsumed { get; private set; } + public bool AsyncQueuedEventConsumed { get; private set; } + + public Task ConsumeAsync(TestEvent payload, CancellationToken cancellationToken) + { + TestEventConsumed = true; + return Task.CompletedTask; + } + + public Task ConsumeAsync(OtherEvent payload, CancellationToken cancellationToken) + { + OtherEventConsumed = true; + return Task.CompletedTask; + } + + public Task ConsumeAsync(AsyncQueuedEvent payload, CancellationToken cancellationToken) + { + AsyncQueuedEventConsumed = true; + return Task.CompletedTask; + } + } + + public class FailingConsumer : IConsumer + { + public Task ConsumeAsync(TestEvent payload, CancellationToken cancellationToken) + { + throw new Exception("Consumer failed."); + } + } +} diff --git a/tests/ReflectionEventing.DependencyInjection.UnitTests/GlobalUsings.cs b/tests/ReflectionEventing.DependencyInjection.UnitTests/GlobalUsings.cs index e49f51b..fc3403c 100644 --- a/tests/ReflectionEventing.DependencyInjection.UnitTests/GlobalUsings.cs +++ b/tests/ReflectionEventing.DependencyInjection.UnitTests/GlobalUsings.cs @@ -9,5 +9,7 @@ global using System.Threading; global using System.Threading.Tasks; global using FluentAssertions; +global using Microsoft.Extensions.DependencyInjection; +global using Microsoft.Extensions.Hosting; global using NSubstitute; global using Xunit; diff --git a/tests/ReflectionEventing.DependencyInjection.UnitTests/ReflectionEventing.DependencyInjection.UnitTests.csproj b/tests/ReflectionEventing.DependencyInjection.UnitTests/ReflectionEventing.DependencyInjection.UnitTests.csproj index 84dbe6b..b37d719 100644 --- a/tests/ReflectionEventing.DependencyInjection.UnitTests/ReflectionEventing.DependencyInjection.UnitTests.csproj +++ b/tests/ReflectionEventing.DependencyInjection.UnitTests/ReflectionEventing.DependencyInjection.UnitTests.csproj @@ -1,14 +1,15 @@ - net8.0 + net9.0 false true - - + + + diff --git a/tests/ReflectionEventing.DependencyInjection.UnitTests/ServiceCollectionExtensionsTests.cs b/tests/ReflectionEventing.DependencyInjection.UnitTests/ServiceCollectionExtensionsTests.cs index d57f7a4..1124676 100644 --- a/tests/ReflectionEventing.DependencyInjection.UnitTests/ServiceCollectionExtensionsTests.cs +++ b/tests/ReflectionEventing.DependencyInjection.UnitTests/ServiceCollectionExtensionsTests.cs @@ -12,7 +12,7 @@ public sealed class ServiceCollectionExtensionsTests [Fact] public void AddEventBus_RegistersServicesAndAddsConsumer() { - ServiceCollection services = new ServiceCollection(); + ServiceCollection services = new(); _ = services.AddScoped(); _ = services.AddEventBus(builder => diff --git a/tests/ReflectionEventing.Ninject.UnitTests/ReflectionEventing.Ninject.UnitTests.csproj b/tests/ReflectionEventing.Ninject.UnitTests/ReflectionEventing.Ninject.UnitTests.csproj index 77b1c39..44c6815 100644 --- a/tests/ReflectionEventing.Ninject.UnitTests/ReflectionEventing.Ninject.UnitTests.csproj +++ b/tests/ReflectionEventing.Ninject.UnitTests/ReflectionEventing.Ninject.UnitTests.csproj @@ -1,13 +1,13 @@ - net8.0 + net9.0 false true - + diff --git a/tests/ReflectionEventing.UnitTests/EventBusTests.cs b/tests/ReflectionEventing.UnitTests/EventBusTests.cs index f73b84b..6158a1c 100644 --- a/tests/ReflectionEventing.UnitTests/EventBusTests.cs +++ b/tests/ReflectionEventing.UnitTests/EventBusTests.cs @@ -3,23 +3,27 @@ // Copyright (C) Leszek Pomianowski and ReflectionEventing Contributors. // All Rights Reserved. +using ReflectionEventing.Queues; + namespace ReflectionEventing.UnitTests; public sealed class EventBusTests { private readonly IConsumerProvider _consumerProvider; private readonly IConsumerTypesProvider _consumerTypesProvider; + private readonly IEventsQueue _eventsQueue; private readonly EventBus _eventBus; public EventBusTests() { _consumerProvider = Substitute.For(); _consumerTypesProvider = Substitute.For(); - _eventBus = new EventBus(_consumerProvider, _consumerTypesProvider); + _eventsQueue = Substitute.For(); + _eventBus = new EventBus(_consumerProvider, _consumerTypesProvider, _eventsQueue); } [Fact] - public async Task PublishAsync_ShouldCallConsumeAsyncOnAllConsumers() + public async Task SendAsync_ShouldCallConsumeAsyncOnAllConsumers() { TestEvent testEvent = new(); Type consumerType = typeof(IConsumer); @@ -28,14 +32,14 @@ public async Task PublishAsync_ShouldCallConsumeAsyncOnAllConsumers() _ = _consumerTypesProvider.GetConsumerTypes().Returns([consumerType]); _ = _consumerProvider.GetConsumers(consumerType).Returns([consumer]); - await _eventBus.PublishAsync(testEvent, CancellationToken.None); + await _eventBus.SendAsync(testEvent, CancellationToken.None); await consumer.Received().ConsumeAsync(testEvent, Arg.Any()); } [Fact] #pragma warning disable CS0618 // Type or member is obsolete - public async Task Publish_ShouldCallPublishAsync() + public async Task Send_ShouldCallSendAsync() { TestEvent testEvent = new(); Type consumerType = typeof(IConsumer); @@ -44,7 +48,7 @@ public async Task Publish_ShouldCallPublishAsync() _ = _consumerTypesProvider.GetConsumerTypes().Returns([consumerType]); _ = _consumerProvider.GetConsumers(consumerType).Returns([consumer]); - _eventBus.Publish(testEvent); + _eventBus.Send(testEvent); await consumer.Received().ConsumeAsync(testEvent, Arg.Any()); } diff --git a/tests/ReflectionEventing.UnitTests/HashedConsumerTypesProviderTests.cs b/tests/ReflectionEventing.UnitTests/HashedConsumerTypesProviderTests.cs index ad0fee6..528bb3a 100644 --- a/tests/ReflectionEventing.UnitTests/HashedConsumerTypesProviderTests.cs +++ b/tests/ReflectionEventing.UnitTests/HashedConsumerTypesProviderTests.cs @@ -10,15 +10,14 @@ public sealed class HashedConsumerTypesProviderTests [Fact] public void GetConsumerTypes_ShouldReturnCollectionOfConsumers() { - HashedConsumerTypesProvider testEvent = - new( - new Dictionary> - { - { typeof(PrimarySampleConsumer), [typeof(PrimaryTestEvent)] }, - { typeof(SecondarySampleConsumer), [typeof(PrimaryTestEvent)] }, - { typeof(TertiarySampleConsumer), [typeof(SecondaryTestEvent)] } - } - ); + HashedConsumerTypesProvider testEvent = new( + new Dictionary> + { + { typeof(PrimarySampleConsumer), [typeof(PrimaryTestEvent)] }, + { typeof(SecondarySampleConsumer), [typeof(PrimaryTestEvent)] }, + { typeof(TertiarySampleConsumer), [typeof(SecondaryTestEvent)] }, + } + ); IEnumerable consumers = testEvent.GetConsumerTypes(); diff --git a/tests/ReflectionEventing.UnitTests/ReflectionEventing.UnitTests.csproj b/tests/ReflectionEventing.UnitTests/ReflectionEventing.UnitTests.csproj index fe99c83..36672a0 100644 --- a/tests/ReflectionEventing.UnitTests/ReflectionEventing.UnitTests.csproj +++ b/tests/ReflectionEventing.UnitTests/ReflectionEventing.UnitTests.csproj @@ -1,13 +1,13 @@ - net8.0 + net9.0 false true - + diff --git a/tests/ReflectionEventing.Unity.UnitTests/ReflectionEventing.Unity.UnitTests.csproj b/tests/ReflectionEventing.Unity.UnitTests/ReflectionEventing.Unity.UnitTests.csproj index e18d1e1..9d8d2de 100644 --- a/tests/ReflectionEventing.Unity.UnitTests/ReflectionEventing.Unity.UnitTests.csproj +++ b/tests/ReflectionEventing.Unity.UnitTests/ReflectionEventing.Unity.UnitTests.csproj @@ -1,13 +1,13 @@ - + - net8.0 + net9.0 false true - +