diff --git a/.github/workflows/publish.yml b/.github/workflows/publish.yml index d986fe3..37bba2f 100644 --- a/.github/workflows/publish.yml +++ b/.github/workflows/publish.yml @@ -68,6 +68,13 @@ jobs: --configuration Release --output ./nupkgs + - name: Pack ChannelMediator.InMemory + run: > + dotnet pack src/ChannelMediator.InMemory/ChannelMediator.InMemory.csproj + --no-build + --configuration Release + --output ./nupkgs + - name: Pack ChannelMediator.ApiGenerators.Abstraction run: > dotnet pack src/ChannelMediator.ApiGenerators.Abstraction/ChannelMediator.ApiGenerators.Abstraction.csproj diff --git a/ChannelMediator.slnx b/ChannelMediator.slnx index 047ab3c..9e27a4b 100644 --- a/ChannelMediator.slnx +++ b/ChannelMediator.slnx @@ -5,6 +5,7 @@ + @@ -35,7 +36,6 @@ - diff --git a/README.md b/README.md index 3c4e26b..2113554 100644 --- a/README.md +++ b/README.md @@ -1,9 +1,10 @@ ο»Ώ# πŸš€ ChannelMediator [![NuGet ChannelMediator](https://img.shields.io/nuget/v/ChannelMediator?label=ChannelMediator&logo=nuget)](https://www.nuget.org/packages/ChannelMediator/) -[![NuGet ChannelMediator.Contracts](https://img.shields.io/nuget/v/ChannelMediator.Contracts?label=ChannelMediator.Contracts&logo=nuget)](https://www.nuget.org/packages/ChannelMediator.Contracts/) -[![NuGet ChannelMediator.AzureBus](https://img.shields.io/nuget/v/ChannelMediator.AzureBus?label=ChannelMediator.AzureBus&logo=nuget)](https://www.nuget.org/packages/ChannelMediator.AzureBus/) -[![NuGet ChannelMediator.RabbitMQ](https://img.shields.io/nuget/v/ChannelMediator.RabbitMQ?label=ChannelMediator.RabbitMQ&logo=nuget)](https://www.nuget.org/packages/ChannelMediator.RabbitMQ/) +[![NuGet ChannelMediator.Contracts](https://img.shields.io/nuget/v/ChannelMediator.Contracts?label=ChannelMediator.Contracts&logo=nuget)](https://www.nuget.org/packages/ChannelMediator.Contracts/) +[![NuGet ChannelMediator.AzureBus](https://img.shields.io/nuget/v/ChannelMediator.AzureBus?label=ChannelMediator.AzureBus&logo=nuget)](https://www.nuget.org/packages/ChannelMediator.AzureBus/) +[![NuGet ChannelMediator.InMemory](https://img.shields.io/nuget/v/ChannelMediator.InMemory?label=ChannelMediator.InMemory&logo=nuget)](https://www.nuget.org/packages/ChannelMediator.InMemory/) +[![NuGet ChannelMediator.RabbitMQ](https://img.shields.io/nuget/v/ChannelMediator.RabbitMQ?label=ChannelMediator.RabbitMQ&logo=nuget)](https://www.nuget.org/packages/ChannelMediator.RabbitMQ/) [![NuGet ChannelMediator.ApiGenerators.Abstraction](https://img.shields.io/nuget/v/ChannelMediator.ApiGenerators.Abstraction?label=ChannelMediator.ApiGenerators.Abstraction&logo=nuget)](https://www.nuget.org/packages/ChannelMediator.ApiGenerators.Abstraction/) [![NuGet ChannelMediator.MinimalApiGenerator](https://img.shields.io/nuget/v/ChannelMediator.MinimalApiGenerator?label=ChannelMediator.MinimalApiGenerator&logo=nuget)](https://www.nuget.org/packages/ChannelMediator.MinimalApiGenerator/) [![NuGet ChannelMediator.ApiClientGenerator](https://img.shields.io/nuget/v/ChannelMediator.ApiClientGenerator?label=ChannelMediator.ApiClientGenerator&logo=nuget)](https://www.nuget.org/packages/ChannelMediator.ApiClientGenerator/) @@ -24,9 +25,10 @@ Compatible with **.NET 8**, **.NET 9**, and **.NET 10**. - βœ… **Pipeline Behaviors** - Global AND specific - βœ… **Streaming** - `IAsyncEnumerable` with `IStreamRequest` and stream pipeline behaviors - βœ… **Parallel Notifications** - Sequential or parallel broadcasting -- βœ… **High Performance** - Channel-based with modern optimizations -- βœ… **Azure Service Bus** - Distributed messaging with queues and topics -- βœ… **RabbitMQ** - Self-hosted distributed messaging with exchanges and queues +- βœ… **High Performance** - Channel-based with modern optimizations +- βœ… **In-Memory Pub/Sub** - Fire-and-forget `Notify` and `EnqueueRequest` without an external broker +- βœ… **Azure Service Bus** - Distributed messaging with queues and topics +- βœ… **RabbitMQ** - Self-hosted distributed messaging with exchanges and queues - βœ… **Minimal API Generator** - Source-generated endpoint mapping from request attributes - βœ… **API Client Generator** - Source-generated `HttpClient` handlers for consuming generated APIs - βœ… **gRPC Generator** - Source-generated code-first gRPC services via `protobuf-net.Grpc` @@ -253,10 +255,11 @@ services.AddScoped, | `Publish(TNotification, CancellationToken)` | `Task` | Publishes a notification to multiple handlers | | `CreateStream(IStreamRequest, CancellationToken)` | `IAsyncEnumerable` | Creates an async stream from a streaming handler | -## πŸ“š Documentation - -- [🚌 Azure Service Bus Integration](./AZURE_SERVICE_BUS.md) -- [πŸ‡ RabbitMQ Integration](./RABBITMQ.md) +## πŸ“š Documentation + +- [🧠 In-Memory Integration](#-in-memory-integration) +- [🚌 Azure Service Bus Integration](./AZURE_SERVICE_BUS.md) +- [πŸ‡ RabbitMQ Integration](./RABBITMQ.md) - [⚑ Minimal API & Client Generators](./GENERATORS.md) - [πŸ”„ MediatR Compatibility](./MEDIATR_COMPATIBILITY.md) - [🎭 Pipeline Behaviors](./PIPELINE_BEHAVIORS.md) @@ -334,7 +337,7 @@ See [⚑ Generators documentation](./GENERATORS.md) for the full reference. πŸ‘‰ **[Full documentation β†’](./GENERATORS.md)** -## πŸ—οΈ Architecture +## πŸ—οΈ Architecture ``` Client @@ -350,11 +353,36 @@ Pipeline Behaviors (chain) β”œβ”€ Global Behavior 2 β”œβ”€ Specific Behavior 1 └─ Request Handler (business logic) -``` - -## 🚌 Azure Service Bus Integration - -In a microservice architecture, a single process cannot handle all requests. You need to **distribute workloads** across multiple consumer instances and **decouple services** through asynchronous messaging. +``` + +## 🧠 In-Memory Integration + +`ChannelMediator.InMemory` provides the same `Notify` and `EnqueueRequest` extension methods as the broker integrations, but dispatches everything inside the current process. + +This is useful when you want the producer call to return immediately while the actual handler execution continues on a background thread, without requiring Azure Service Bus or RabbitMQ. + +```csharp +using ChannelMediator.InMemory; + +services.AddChannelMediator(config => +{ + config.UseChannelMediatorInMemory(); +}, Assembly.GetExecutingAssembly()); + +var mediator = provider.GetRequiredService(); + +// Scheduled on a background thread, returns immediately +await mediator.Notify(new ProductAddedNotification("SKU-001", 5)); + +// Scheduled on a background thread, returns immediately +await mediator.EnqueueRequest(new MyRequest("process-order-42")); +``` + +Both methods use fire-and-forget scheduling: the returned task completes once the in-memory work has been scheduled, not when the handler has completed. + +## 🚌 Azure Service Bus Integration + +In a microservice architecture, a single process cannot handle all requests. You need to **distribute workloads** across multiple consumer instances and **decouple services** through asynchronous messaging. `ChannelMediator.AzureBus` extends the mediator with two extension methods that transparently route messages through **Azure Service Bus**: diff --git a/src/ChannelMediator.InMemory/ChannelMediator.InMemory.csproj b/src/ChannelMediator.InMemory/ChannelMediator.InMemory.csproj new file mode 100644 index 0000000..73d0697 --- /dev/null +++ b/src/ChannelMediator.InMemory/ChannelMediator.InMemory.csproj @@ -0,0 +1,61 @@ + + + + net10.0;net9.0;net8.0 + enable + enable + true + true + snupkg + true + 1.1.14.0 + Appliman + https://github.com/appliman/channelmediator + https://github.com/appliman/channelmediator + git + true + ChannelMediator;InMemory;Messaging;Queues;Topics + In-memory messaging integration for ChannelMediator notifications and queued requests. + README.md + + + + + + + + + + + + + + + + + + + + None + lib\net8.0 + true + + + + + + None + lib\net9.0 + true + + + + + + None + lib\net10.0 + true + + + + diff --git a/src/ChannelMediator.InMemory/GlobalInitializerHostedService.cs b/src/ChannelMediator.InMemory/GlobalInitializerHostedService.cs new file mode 100644 index 0000000..56a35dc --- /dev/null +++ b/src/ChannelMediator.InMemory/GlobalInitializerHostedService.cs @@ -0,0 +1,27 @@ +using Microsoft.Extensions.Hosting; + +namespace ChannelMediator.InMemory; + +/// +/// Hosted service that initializes the global publisher for the memory mediator extensions. +/// +internal sealed class GlobalInitializerHostedService : IHostedService +{ + private readonly IMemoryPublisher _globalPublisher; + + public GlobalInitializerHostedService(IMemoryPublisher globalPublisher) + { + _globalPublisher = globalPublisher ?? throw new ArgumentNullException(nameof(globalPublisher)); + } + + public Task StartAsync(CancellationToken cancellationToken) + { + MediatorExtensions.SetGlobalPublisher(_globalPublisher); + return Task.CompletedTask; + } + + public Task StopAsync(CancellationToken cancellationToken) + { + return Task.CompletedTask; + } +} diff --git a/src/ChannelMediator.InMemory/GlobalUsings.cs b/src/ChannelMediator.InMemory/GlobalUsings.cs new file mode 100644 index 0000000..dcb1cfc --- /dev/null +++ b/src/ChannelMediator.InMemory/GlobalUsings.cs @@ -0,0 +1 @@ +global using ChannelMediator; diff --git a/src/ChannelMediator.InMemory/IMemoryPublisher.cs b/src/ChannelMediator.InMemory/IMemoryPublisher.cs new file mode 100644 index 0000000..aea5fba --- /dev/null +++ b/src/ChannelMediator.InMemory/IMemoryPublisher.cs @@ -0,0 +1,27 @@ +namespace ChannelMediator.InMemory; + +/// +/// Interface for publishing notifications and enqueuing requests in memory. +/// +internal interface IMemoryPublisher +{ + /// + /// Publishes the specified notification through the local mediator. + /// + /// The notification type. + /// The notification to publish. + /// Cancellation token. + /// A task representing the asynchronous operation. + Task Notify(T notification, CancellationToken cancellationToken = default) + where T : INotification; + + /// + /// Dispatches the specified request through the local mediator. + /// + /// The request type. + /// The request to enqueue. + /// Cancellation token. + /// A task representing the asynchronous operation. + Task EnqueueRequest(R request, CancellationToken cancellationToken = default) + where R : IRequest; +} diff --git a/src/ChannelMediator.InMemory/InMemoryOptions.cs b/src/ChannelMediator.InMemory/InMemoryOptions.cs new file mode 100644 index 0000000..4db482f --- /dev/null +++ b/src/ChannelMediator.InMemory/InMemoryOptions.cs @@ -0,0 +1,18 @@ +using Microsoft.Extensions.DependencyInjection; + +namespace ChannelMediator.InMemory; + +/// +/// Configuration options for ChannelMediator in-memory publishing. +/// +public sealed class InMemoryOptions +{ + internal InMemoryOptions() + { + } + + /// + /// Gets or sets the collection of service descriptors for dependency injection. + /// + public IServiceCollection Services { get; set; } = default!; +} diff --git a/src/ChannelMediator.InMemory/MediatorExtensions.cs b/src/ChannelMediator.InMemory/MediatorExtensions.cs new file mode 100644 index 0000000..388a2a8 --- /dev/null +++ b/src/ChannelMediator.InMemory/MediatorExtensions.cs @@ -0,0 +1,69 @@ +namespace ChannelMediator.InMemory; + +/// +/// Extension methods for IMediator to support in-memory publishing. +/// +public static class MediatorExtensions +{ + private static IMemoryPublisher? _globalPublisher; + private static readonly object Lock = new(); + + /// + /// Sets the global publisher instance. This is called internally during service configuration. + /// + /// The global publisher instance. + internal static void SetGlobalPublisher(IMemoryPublisher globalPublisher) + { + lock (Lock) + { + _globalPublisher = globalPublisher; + } + } + + /// + /// Publishes a notification in memory through the configured mediator. + /// + /// The notification type. + /// The mediator instance. + /// The notification to publish. + /// Cancellation token. + /// A completed task once the background work has been scheduled. + public static Task Notify( + this IMediator mediator, + TNotification notification, + CancellationToken cancellationToken = default) + where TNotification : INotification + { + ArgumentNullException.ThrowIfNull(mediator); + ArgumentNullException.ThrowIfNull(notification); + + var publisher = _globalPublisher + ?? throw new InvalidOperationException( + "GlobalPublisher is not configured. Ensure UseChannelMediatorInMemory() has been called during service configuration."); + + _ = Task.Run(() => publisher.Notify(notification, cancellationToken), CancellationToken.None); + return Task.CompletedTask; + } + + /// + /// Enqueues a request in memory through the configured mediator. + /// + /// The request type. + /// The mediator instance. + /// The request to enqueue. + /// Cancellation token. + /// A completed task once the background work has been scheduled. + public static Task EnqueueRequest(this IMediator mediator, R request, CancellationToken cancellationToken = default) + where R : IRequest + { + ArgumentNullException.ThrowIfNull(mediator); + ArgumentNullException.ThrowIfNull(request); + + var publisher = _globalPublisher + ?? throw new InvalidOperationException( + "GlobalPublisher is not configured. Ensure UseChannelMediatorInMemory() has been called during service configuration."); + + _ = Task.Run(() => publisher.EnqueueRequest(request, cancellationToken), CancellationToken.None); + return Task.CompletedTask; + } +} diff --git a/src/ChannelMediator.InMemory/MemoryPublisher.cs b/src/ChannelMediator.InMemory/MemoryPublisher.cs new file mode 100644 index 0000000..bf9710c --- /dev/null +++ b/src/ChannelMediator.InMemory/MemoryPublisher.cs @@ -0,0 +1,26 @@ +using Microsoft.Extensions.Logging; + +namespace ChannelMediator.InMemory; + +internal sealed class MemoryPublisher( + IMediator mediator, + ILogger logger) : IMemoryPublisher +{ + public async Task Notify(TNotification notification, CancellationToken cancellationToken = default) + where TNotification : INotification + { + ArgumentNullException.ThrowIfNull(notification); + + await mediator.Publish(notification, cancellationToken); + logger.LogDebug("Memory publisher processed notification {NotificationType}.", typeof(TNotification).Name); + } + + public async Task EnqueueRequest(R request, CancellationToken cancellationToken = default) + where R : IRequest + { + ArgumentNullException.ThrowIfNull(request); + + await mediator.Send(request, cancellationToken); + logger.LogDebug("Memory publisher processed request {RequestType}.", request.GetType().Name); + } +} diff --git a/src/ChannelMediator.InMemory/ServiceCollectionExtensions.cs b/src/ChannelMediator.InMemory/ServiceCollectionExtensions.cs new file mode 100644 index 0000000..f13bad5 --- /dev/null +++ b/src/ChannelMediator.InMemory/ServiceCollectionExtensions.cs @@ -0,0 +1,48 @@ +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.DependencyInjection.Extensions; + +namespace ChannelMediator.InMemory; + +/// +/// Provides dependency injection extensions for enabling ChannelMediator in-memory publishing. +/// +public static class ServiceCollectionExtensions +{ + /// + /// Adds in-memory publishing to ChannelMediator. + /// + /// The ChannelMediator configuration being extended. + /// Optional action to configure in-memory options. + /// The in-memory options for chaining. + public static InMemoryOptions UseChannelMediatorInMemory( + this ChannelMediatorConfiguration configuration, + Action? configure = null) + { + ArgumentNullException.ThrowIfNull(configuration); + + configuration.Services.TryAddEnumerable( + ServiceDescriptor.Singleton()); + + InMemoryOptions options = default!; + + var optionsDescriptor = configuration.Services.FirstOrDefault(descriptor => descriptor.ServiceType == typeof(InMemoryOptions)); + if (optionsDescriptor is null) + { + options = new InMemoryOptions + { + Services = configuration.Services + }; + configuration.Services.TryAddSingleton(options); + } + else + { + options = (InMemoryOptions)optionsDescriptor.ImplementationInstance!; + } + + configure?.Invoke(options); + + configuration.Services.TryAddSingleton(); + + return options; + } +} diff --git a/src/ChannelMediator.Tests/AzureBusQueueReaderTests.cs b/src/ChannelMediator.Tests/AzureBusQueueReaderTests.cs new file mode 100644 index 0000000..30439b1 --- /dev/null +++ b/src/ChannelMediator.Tests/AzureBusQueueReaderTests.cs @@ -0,0 +1,65 @@ +ο»Ώusing System.Reflection; +using System.Text.Json; + +using Azure.Messaging.ServiceBus; +using Azure.Messaging.ServiceBus.Administration; + +using ChannelMediator.AzureBus; +using ChannelMediator.Tests.Helpers; + +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; + +namespace ChannelMediator.Tests; + +public class AzureBusQueueReaderTests +{ + [Fact] + public async Task WhenQueueNameDoesNotMatchMessageType_ThenRequestIsStillConsumed() + { + // Arrange + var expectedValue = $"forced-value-{Guid.NewGuid():N}"; + var services = new ServiceCollection(); + services.AddChannelMediator(null, typeof(TestCommandHandler).Assembly); + var serviceProvider = services.BuildServiceProvider(); + using var loggerFactory = LoggerFactory.Create(_ => { }); + var serializerOptions = new JsonSerializerOptions + { + PropertyNamingPolicy = JsonNamingPolicy.CamelCase + }; + + await using var queueReader = new QueueReader( + new ServiceBusClient("Endpoint=sb://unit-test.servicebus.windows.net/;SharedAccessKeyName=fake;SharedAccessKey=fake="), + new AzureServiceBusEntityManager( + new ServiceBusAdministrationClient("Endpoint=sb://unit-test.servicebus.windows.net/;SharedAccessKeyName=fake;SharedAccessKey=fake="), + NullLogger.Instance), + new QueueReaderOptions + { + QueueName = "forced-queue-name", + RequestType = typeof(TestCommand) + }, + serviceProvider, + loggerFactory.CreateLogger()); + + var message = ServiceBusModelFactory.ServiceBusReceivedMessage( + body: new BinaryData(JsonSerializer.SerializeToUtf8Bytes(new TestCommand(expectedValue), serializerOptions)), + messageId: "forced-message-id", + properties: new Dictionary + { + ["messagetype"] = typeof(TestCommand).AssemblyQualifiedName! + }); + + var args = new ProcessMessageEventArgs(message, receiver: null!, CancellationToken.None); + var processMessageAsync = typeof(QueueReader).GetMethod("ProcessMessageAsync", BindingFlags.Instance | BindingFlags.NonPublic); + + // Act + Assert.NotNull(processMessageAsync); + var task = processMessageAsync.Invoke(queueReader, new object[] { args }) as Task; + Assert.NotNull(task); + await task!; + + // Assert + Assert.Contains(expectedValue, TestCommandHandler.ExecutedValues); + } +} diff --git a/src/ChannelMediator.Tests/AzureBusTopicSubscriptionReaderTests.cs b/src/ChannelMediator.Tests/AzureBusTopicSubscriptionReaderTests.cs new file mode 100644 index 0000000..ec1d450 --- /dev/null +++ b/src/ChannelMediator.Tests/AzureBusTopicSubscriptionReaderTests.cs @@ -0,0 +1,68 @@ +ο»Ώusing System.Reflection; +using System.Text.Json; + +using Azure.Messaging.ServiceBus; +using Azure.Messaging.ServiceBus.Administration; + +using ChannelMediator.AzureBus; +using ChannelMediator.Tests.Helpers; + +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; + +namespace ChannelMediator.Tests; + +public class AzureBusTopicSubscriptionReaderTests +{ + [Fact] + public async Task WhenTopicNameDoesNotMatchMessageType_ThenNotificationIsStillConsumed() + { + // Arrange + var handler = new TestNotificationHandler1(); + var services = new ServiceCollection(); + services.AddSingleton>(handler); + services.AddChannelMediator(null, typeof(TestNotificationHandler1).Assembly); + var serviceProvider = services.BuildServiceProvider(); + using var loggerFactory = LoggerFactory.Create(_ => { }); + var serializerOptions = new JsonSerializerOptions + { + PropertyNamingPolicy = JsonNamingPolicy.CamelCase + }; + + await using var topicReader = new TopicSubscriptionReader( + new ServiceBusClient("Endpoint=sb://unit-test.servicebus.windows.net/;SharedAccessKeyName=fake;SharedAccessKey=fake="), + new AzureServiceBusEntityManager( + new ServiceBusAdministrationClient("Endpoint=sb://unit-test.servicebus.windows.net/;SharedAccessKeyName=fake;SharedAccessKey=fake="), + NullLogger.Instance), + new TopicSubscriptionReaderOptions + { + TopicName = "forced-topic-name", + SubscriptionName = "forced-subscription-name", + MessageType = typeof(TestNotification) + }, + serviceProvider, + loggerFactory.CreateLogger()); + + var expectedMessage = $"forced-topic-{Guid.NewGuid():N}"; + var message = ServiceBusModelFactory.ServiceBusReceivedMessage( + body: new BinaryData(JsonSerializer.SerializeToUtf8Bytes(new TestNotification(expectedMessage), serializerOptions)), + messageId: "forced-topic-message-id", + properties: new Dictionary + { + ["messagetype"] = typeof(TestNotification).AssemblyQualifiedName! + }); + + var args = new ProcessMessageEventArgs(message, receiver: null!, CancellationToken.None); + var processMessageAsync = typeof(TopicSubscriptionReader).GetMethod("ProcessMessageAsync", BindingFlags.Instance | BindingFlags.NonPublic); + + // Act + Assert.NotNull(processMessageAsync); + var task = processMessageAsync.Invoke(topicReader, new object[] { args }) as Task; + Assert.NotNull(task); + await task!; + + // Assert + Assert.Contains($"Handler1: {expectedMessage}", handler.HandledMessages); + } +} diff --git a/src/ChannelMediator.Tests/ChannelMediator.Tests.csproj b/src/ChannelMediator.Tests/ChannelMediator.Tests.csproj index 397c7f5..afc9a4b 100644 --- a/src/ChannelMediator.Tests/ChannelMediator.Tests.csproj +++ b/src/ChannelMediator.Tests/ChannelMediator.Tests.csproj @@ -34,6 +34,7 @@ + diff --git a/src/ChannelMediator.Tests/MemoryPublisherTests.cs b/src/ChannelMediator.Tests/MemoryPublisherTests.cs new file mode 100644 index 0000000..8e9897b --- /dev/null +++ b/src/ChannelMediator.Tests/MemoryPublisherTests.cs @@ -0,0 +1,167 @@ +using ChannelMediator.InMemory; +using ChannelMediator.Tests.Helpers; + +using Microsoft.Extensions.Hosting; + +using System.Diagnostics; + +namespace ChannelMediator.Tests; + +[Collection("MemoryPublisher")] +public class MemoryPublisherTests +{ + [Fact] + public async Task WhenNotifyIsCalled_NotificationHandlersAreCalled() + { + var handler1 = new TestNotificationHandler1(); + var handler2 = new TestNotificationHandler2(); + + var host = Host.CreateDefaultBuilder() + .ConfigureServices(services => + { + services.AddSingleton>(handler1); + services.AddSingleton>(handler2); + + services.AddChannelMediator(config => + { + config.UseChannelMediatorInMemory(); + }, typeof(TestNotificationHandler1).Assembly); + }) + .Build(); + + await host.StartAsync(); + + var mediator = host.Services.GetRequiredService(); + + await mediator.Notify(new TestNotification("memory-test")); + + await WaitUntilAsync(() => handler1.HandledMessages.Count == 1 && handler2.HandledMessages.Count == 1); + + Assert.Single(handler1.HandledMessages); + Assert.Equal("Handler1: memory-test", handler1.HandledMessages[0]); + Assert.Single(handler2.HandledMessages); + Assert.Equal("Handler2: memory-test", handler2.HandledMessages[0]); + + await host.StopAsync(); + } + + [Fact] + public async Task WhenEnqueueRequestIsCalled_RequestHandlerIsCalled() + { + MemoryTestCommandHandler.ExecutedValues.Clear(); + + var host = Host.CreateDefaultBuilder() + .ConfigureServices(services => + { + services.AddChannelMediator(config => + { + config.UseChannelMediatorInMemory(); + }, typeof(TestCommandHandler).Assembly); + }) + .Build(); + + await host.StartAsync(); + + var mediator = host.Services.GetRequiredService(); + + await mediator.EnqueueRequest(new MemoryTestCommand("memory-command")); + + await WaitUntilAsync(() => MemoryTestCommandHandler.ExecutedValues.Contains("memory-command")); + + Assert.Contains("memory-command", MemoryTestCommandHandler.ExecutedValues); + + await host.StopAsync(); + } + + [Fact] + public async Task WhenEnqueueRequestIsCalled_ReturnsBeforeRequestHandlerCompletes() + { + SlowMemoryTestCommandHandler.Reset(); + + var host = Host.CreateDefaultBuilder() + .ConfigureServices(services => + { + services.AddChannelMediator(config => + { + config.UseChannelMediatorInMemory(); + }, typeof(SlowMemoryTestCommandHandler).Assembly); + }) + .Build(); + + await host.StartAsync(); + + var mediator = host.Services.GetRequiredService(); + var stopwatch = Stopwatch.StartNew(); + + await mediator.EnqueueRequest(new SlowMemoryTestCommand("slow-memory-command")); + + stopwatch.Stop(); + + Assert.True(stopwatch.Elapsed < TimeSpan.FromMilliseconds(100)); + Assert.False(SlowMemoryTestCommandHandler.Completed.Task.IsCompleted); + + SlowMemoryTestCommandHandler.Release.SetResult(); + await SlowMemoryTestCommandHandler.Completed.Task.WaitAsync(TimeSpan.FromSeconds(2)); + + await host.StopAsync(); + } + + private static async Task WaitUntilAsync(Func condition) + { + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(2)); + + while (!condition()) + { + await Task.Delay(10, cts.Token); + } + } +} + +public record MemoryTestCommand(string Value) : IRequest; + +public class MemoryTestCommandHandler : IRequestHandler +{ + public static List ExecutedValues { get; } = new(); + + public ValueTask HandleAsync(MemoryTestCommand request, CancellationToken cancellationToken) + { + ExecutedValues.Add(request.Value); + return ValueTask.CompletedTask; + } + + public Task Handle(MemoryTestCommand request, CancellationToken cancellationToken) + { + return HandleAsync(request, cancellationToken).AsTask(); + } +} + +public record SlowMemoryTestCommand(string Value) : IRequest; + +public class SlowMemoryTestCommandHandler : IRequestHandler +{ + public static TaskCompletionSource Release { get; private set; } = CreateTaskCompletionSource(); + + public static TaskCompletionSource Completed { get; private set; } = CreateTaskCompletionSource(); + + public static void Reset() + { + Release = CreateTaskCompletionSource(); + Completed = CreateTaskCompletionSource(); + } + + public async ValueTask HandleAsync(SlowMemoryTestCommand request, CancellationToken cancellationToken) + { + await Release.Task.WaitAsync(cancellationToken); + Completed.SetResult(); + } + + public Task Handle(SlowMemoryTestCommand request, CancellationToken cancellationToken) + { + return HandleAsync(request, cancellationToken).AsTask(); + } + + private static TaskCompletionSource CreateTaskCompletionSource() + { + return new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + } +} diff --git a/src/ChannelMediator.Tests/TestCollections.cs b/src/ChannelMediator.Tests/TestCollections.cs index 12f679f..b0f3b50 100644 --- a/src/ChannelMediator.Tests/TestCollections.cs +++ b/src/ChannelMediator.Tests/TestCollections.cs @@ -4,3 +4,8 @@ public class CommandHandlerTestsCollection { } + +[CollectionDefinition("MemoryPublisher", DisableParallelization = true)] +public class MemoryPublisherCollection +{ +} diff --git a/src/Samples/AzureBusReaderSampleConsole/Program.cs b/src/Samples/AzureBusReaderSampleConsole/Program.cs index 1aeaa88..cc5abb8 100644 --- a/src/Samples/AzureBusReaderSampleConsole/Program.cs +++ b/src/Samples/AzureBusReaderSampleConsole/Program.cs @@ -23,8 +23,12 @@ opts.ConnectionString = connectionString!; opts.TopicSubscriberName = "my-subscriber-name"; - opts.AddAzureQueueRequestReader(); - opts.AddAllAzureBusTopicNotification(); + opts.AddAzureQueueRequestReader(ForcedQueueNames.MyRequest); + opts.AddAzureBusTopicNotificationReader(opts.TopicSubscriberName, reader => + { + reader.TopicName = $"{opts.Prefix}{ForcedQueueNames.ProductAddedNotification}"; + }); + opts.AddAzureBusTopicNotificationReader(opts.TopicSubscriberName); }); }, Assembly.GetExecutingAssembly()); diff --git a/src/Samples/AzureBusWriterSampleConsole/Program.cs b/src/Samples/AzureBusWriterSampleConsole/Program.cs index 2f53516..838ad7f 100644 --- a/src/Samples/AzureBusWriterSampleConsole/Program.cs +++ b/src/Samples/AzureBusWriterSampleConsole/Program.cs @@ -1,4 +1,8 @@ ο»Ώusing System.Reflection; +using System.Text.Json; + +using Azure.Messaging.ServiceBus; +using Azure.Messaging.ServiceBus.Administration; using ChannelMediator; using ChannelMediator.AzureBus; @@ -33,9 +37,57 @@ await app.StartAsync(); var mediator = app.Services.GetRequiredService(); +var serviceBusClient = app.Services.GetRequiredService(); +var administrationClient = app.Services.GetRequiredService(); + +const string queuePrefix = "sampleapp-"; +var forcedQueueName = $"{queuePrefix}{ForcedQueueNames.MyRequest}"; +var forcedTopicName = $"{queuePrefix}{ForcedQueueNames.ProductAddedNotification}"; + +if (!await administrationClient.QueueExistsAsync(forcedQueueName)) +{ + await administrationClient.CreateQueueAsync(new CreateQueueOptions(forcedQueueName)); +} + +await using var forcedQueueSender = serviceBusClient.CreateSender(forcedQueueName); +await using var forcedTopicSender = serviceBusClient.CreateSender(forcedTopicName); +var serializerOptions = new JsonSerializerOptions +{ + PropertyNamingPolicy = JsonNamingPolicy.CamelCase +}; + +var forcedQueueRequest = new MyRequest("enqueue-test-via-forced-queue"); +var forcedQueueMessage = new ServiceBusMessage(JsonSerializer.SerializeToUtf8Bytes(forcedQueueRequest, serializerOptions)) +{ + ContentType = "application/json", + Subject = nameof(MyRequest), + ApplicationProperties = + { + ["messagetype"] = typeof(MyRequest).AssemblyQualifiedName + } +}; + +await forcedQueueSender.SendMessageAsync(forcedQueueMessage); +Console.WriteLine($"[WRITER] Sent MyRequest to forced queue '{forcedQueueName}' even though the queue name differs from the message type."); + +if (!await administrationClient.TopicExistsAsync(forcedTopicName)) +{ + await administrationClient.CreateTopicAsync(new CreateTopicOptions(forcedTopicName)); +} + +var forcedTopicNotification = new ProductAddedNotification("p01", 10, 100); +var forcedTopicMessage = new ServiceBusMessage(JsonSerializer.SerializeToUtf8Bytes(forcedTopicNotification, serializerOptions)) +{ + ContentType = "application/json", + Subject = nameof(ProductAddedNotification), + ApplicationProperties = + { + ["messagetype"] = typeof(ProductAddedNotification).AssemblyQualifiedName + } +}; -await mediator.EnqueueRequest(new MyRequest("enqueue-test")); -await mediator.Notify(new ProductAddedNotification("p01", 10, 100)); +await forcedTopicSender.SendMessageAsync(forcedTopicMessage); +Console.WriteLine($"[WRITER] Sent ProductAddedNotification to forced topic '{forcedTopicName}' even though the topic name differs from the message type."); // Publish multiple OrderShippedNotification messages and verify delivery via logs var orders = new[] diff --git a/src/Samples/ChannelMediatorSampleShared/ForcedQueueNames.cs b/src/Samples/ChannelMediatorSampleShared/ForcedQueueNames.cs new file mode 100644 index 0000000..1a7efb5 --- /dev/null +++ b/src/Samples/ChannelMediatorSampleShared/ForcedQueueNames.cs @@ -0,0 +1,7 @@ +ο»Ώnamespace ChannelMediatorSampleShared; + +public static class ForcedQueueNames +{ + public const string MyRequest = "forced-my-request-queue"; + public const string ProductAddedNotification = "forced-product-added-topic"; +}