Add in-memory pub/sub integration and tests

.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

ChannelMediator.slnx

@@ -5,6 +5,7 @@
   </Folder>
   <Folder Name="/2-PubSub/">
     <Project Path="src/ChannelMediator.AzureBus/ChannelMediator.AzureBus.csproj" />
+    <Project Path="src/ChannelMediator.InMemory/ChannelMediator.InMemory.csproj" />
     <Project Path="src/ChannelMediator.RabbitMQ/ChannelMediator.RabbitMQ.csproj" />
   </Folder>
   <Folder Name="/3-Api/">
@@ -35,7 +36,6 @@
   <Folder Name="/Solution Items/">
     <File Path=".azure-pipelines/workflows/azure-pipeline.yml" />
     <File Path=".azure-pipelines/workflows/channelmediator-publish.yml" />
-    <File Path=".github/workflows/ci-publish.yml" />
     <File Path="AZURE_SERVICE_BUS.md" />
     <File Path="CODE_EXAMPLES.md" />
     <File Path="README.md" />

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<T>` with `IStreamRequest<T>` 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<IStreamPipelineBehavior<GetOrderLinesQuery, OrderLineDto>,
 | `Publish<TNotification>(TNotification, CancellationToken)` | `Task` | Publishes a notification to multiple handlers |
 | `CreateStream<TResponse>(IStreamRequest<TResponse>, CancellationToken)` | `IAsyncEnumerable<TResponse>` | 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<IMediator>();
+
+// 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**:
 

src/ChannelMediator.InMemory/ChannelMediator.InMemory.csproj

@@ -0,0 +1,61 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+	<PropertyGroup>
+		<TargetFrameworks>net10.0;net9.0;net8.0</TargetFrameworks>
+		<ImplicitUsings>enable</ImplicitUsings>
+		<Nullable>enable</Nullable>
+		<GenerateDocumentationFile>true</GenerateDocumentationFile>
+		<IncludeSymbols>true</IncludeSymbols>
+		<SymbolPackageFormat>snupkg</SymbolPackageFormat>
+		<EmbedUntrackedSources>true</EmbedUntrackedSources>
+		<Version>1.1.14.0</Version>
+		<Authors>Appliman</Authors>
+		<PackageProjectUrl>https://github.com/appliman/channelmediator</PackageProjectUrl>
+		<RepositoryUrl>https://github.com/appliman/channelmediator</RepositoryUrl>
+		<RepositoryType>git</RepositoryType>
+		<PublishRepositoryUrl>true</PublishRepositoryUrl>
+		<PackageTags>ChannelMediator;InMemory;Messaging;Queues;Topics</PackageTags>
+		<Description>In-memory messaging integration for ChannelMediator notifications and queued requests.</Description>
+		<PackageReadmeFile>README.md</PackageReadmeFile>
+	</PropertyGroup>
+
+	<ItemGroup>
+		<None Include="..\..\README.md" Pack="true" PackagePath="\" />
+	</ItemGroup>
+
+	<ItemGroup>
+		<PackageReference Include="Microsoft.Extensions.DependencyInjection" Version="10.0.8" />
+		<PackageReference Include="Microsoft.Extensions.Hosting.Abstractions" Version="10.0.8" />
+		<PackageReference Include="Microsoft.Extensions.Logging.Abstractions" Version="10.0.8" />
+		<PackageReference Include="Microsoft.SourceLink.GitHub" Version="10.0.300" PrivateAssets="All" />
+	</ItemGroup>
+
+	<ItemGroup>
+		<ProjectReference Include="..\ChannelMediator\ChannelMediator.csproj" />
+	</ItemGroup>
+
+	<ItemGroup Condition="'$(TargetFramework)' == 'net8.0'">
+		<Content Include="$(OutputPath)\ChannelMediator.dll">
+			<BuildAction>None</BuildAction>
+			<PackagePath>lib\net8.0</PackagePath>
+			<Pack>true</Pack>
+		</Content>
+	</ItemGroup>
+
+	<ItemGroup Condition="'$(TargetFramework)' == 'net9.0'">
+		<Content Include="$(OutputPath)\ChannelMediator.dll">
+			<BuildAction>None</BuildAction>
+			<PackagePath>lib\net9.0</PackagePath>
+			<Pack>true</Pack>
+		</Content>
+	</ItemGroup>
+
+	<ItemGroup Condition="'$(TargetFramework)' == 'net10.0'">
+		<Content Include="$(OutputPath)\ChannelMediator.dll">
+			<BuildAction>None</BuildAction>
+			<PackagePath>lib\net10.0</PackagePath>
+			<Pack>true</Pack>
+		</Content>
+	</ItemGroup>
+
+</Project>

src/ChannelMediator.InMemory/GlobalInitializerHostedService.cs

@@ -0,0 +1,27 @@
+using Microsoft.Extensions.Hosting;
+
+namespace ChannelMediator.InMemory;
+
+/// <summary>
+/// Hosted service that initializes the global publisher for the memory mediator extensions.
+/// </summary>
+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;
+	}
+}

src/ChannelMediator.InMemory/GlobalUsings.cs

@@ -0,0 +1 @@
+global using ChannelMediator;

src/ChannelMediator.InMemory/IMemoryPublisher.cs

@@ -0,0 +1,27 @@
+namespace ChannelMediator.InMemory;
+
+/// <summary>
+/// Interface for publishing notifications and enqueuing requests in memory.
+/// </summary>
+internal interface IMemoryPublisher
+{
+	/// <summary>
+	/// Publishes the specified notification through the local mediator.
+	/// </summary>
+	/// <typeparam name="T">The notification type.</typeparam>
+	/// <param name="notification">The notification to publish.</param>
+	/// <param name="cancellationToken">Cancellation token.</param>
+	/// <returns>A task representing the asynchronous operation.</returns>
+	Task Notify<T>(T notification, CancellationToken cancellationToken = default)
+		where T : INotification;
+
+	/// <summary>
+	/// Dispatches the specified request through the local mediator.
+	/// </summary>
+	/// <typeparam name="R">The request type.</typeparam>
+	/// <param name="request">The request to enqueue.</param>
+	/// <param name="cancellationToken">Cancellation token.</param>
+	/// <returns>A task representing the asynchronous operation.</returns>
+	Task EnqueueRequest<R>(R request, CancellationToken cancellationToken = default)
+		where R : IRequest;
+}

src/ChannelMediator.InMemory/InMemoryOptions.cs

@@ -0,0 +1,18 @@
+using Microsoft.Extensions.DependencyInjection;
+
+namespace ChannelMediator.InMemory;
+
+/// <summary>
+/// Configuration options for ChannelMediator in-memory publishing.
+/// </summary>
+public sealed class InMemoryOptions
+{
+	internal InMemoryOptions()
+	{
+	}
+
+	/// <summary>
+	/// Gets or sets the collection of service descriptors for dependency injection.
+	/// </summary>
+	public IServiceCollection Services { get; set; } = default!;
+}

src/ChannelMediator.InMemory/MediatorExtensions.cs

@@ -0,0 +1,69 @@
+namespace ChannelMediator.InMemory;
+
+/// <summary>
+/// Extension methods for IMediator to support in-memory publishing.
+/// </summary>
+public static class MediatorExtensions
+{
+	private static IMemoryPublisher? _globalPublisher;
+	private static readonly object Lock = new();
+
+	/// <summary>
+	/// Sets the global publisher instance. This is called internally during service configuration.
+	/// </summary>
+	/// <param name="globalPublisher">The global publisher instance.</param>
+	internal static void SetGlobalPublisher(IMemoryPublisher globalPublisher)
+	{
+		lock (Lock)
+		{
+			_globalPublisher = globalPublisher;
+		}
+	}
+
+	/// <summary>
+	/// Publishes a notification in memory through the configured mediator.
+	/// </summary>
+	/// <typeparam name="TNotification">The notification type.</typeparam>
+	/// <param name="mediator">The mediator instance.</param>
+	/// <param name="notification">The notification to publish.</param>
+	/// <param name="cancellationToken">Cancellation token.</param>
+	/// <returns>A completed task once the background work has been scheduled.</returns>
+	public static Task Notify<TNotification>(
+		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;
+	}
+
+	/// <summary>
+	/// Enqueues a request in memory through the configured mediator.
+	/// </summary>
+	/// <typeparam name="R">The request type.</typeparam>
+	/// <param name="mediator">The mediator instance.</param>
+	/// <param name="request">The request to enqueue.</param>
+	/// <param name="cancellationToken">Cancellation token.</param>
+	/// <returns>A completed task once the background work has been scheduled.</returns>
+	public static Task EnqueueRequest<R>(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;
+	}
+}

src/ChannelMediator.InMemory/MemoryPublisher.cs

@@ -0,0 +1,26 @@
+using Microsoft.Extensions.Logging;
+
+namespace ChannelMediator.InMemory;
+
+internal sealed class MemoryPublisher(
+	IMediator mediator,
+	ILogger<MemoryPublisher> logger) : IMemoryPublisher
+{
+	public async Task Notify<TNotification>(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>(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);
+	}
+}