Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
51 changes: 51 additions & 0 deletions src/Cortex.Mediator/Behaviors/VoidLoggingCommandBehavior.cs
Original file line number Diff line number Diff line change
@@ -0,0 +1,51 @@
using Cortex.Mediator.Commands;
using Microsoft.Extensions.Logging;
using System;
using System.Diagnostics;
using System.Threading;
using System.Threading.Tasks;

namespace Cortex.Mediator.Behaviors
{

public sealed class LoggingCommandBehavior<TCommand> : ICommandPipelineBehavior<TCommand> where TCommand : ICommand
{
private readonly ILogger<LoggingCommandBehavior<TCommand>> _logger;

public LoggingCommandBehavior(ILogger<LoggingCommandBehavior<TCommand>> logger)
{
_logger = logger;
}

public async Task Handle(
TCommand command,
CommandHandlerDelegate next,
CancellationToken cancellationToken)
{
var commandName = typeof(TCommand).Name;
_logger.LogInformation("Executing command {CommandName}", commandName);

var stopwatch = Stopwatch.StartNew(); // start timing
try
{
await next();

stopwatch.Stop();
_logger.LogInformation(
"Command {CommandName} executed successfully in {ElapsedMilliseconds} ms",
commandName,
stopwatch.ElapsedMilliseconds);
}
catch (Exception ex)
{
stopwatch.Stop();
_logger.LogError(
ex,
"Error executing command {CommandName} after {ElapsedMilliseconds} ms",
commandName,
stopwatch.ElapsedMilliseconds);
throw;
}
}
}
}
10 changes: 10 additions & 0 deletions src/Cortex.Mediator/Commands/ICommand.cs
Original file line number Diff line number Diff line change
Expand Up @@ -8,4 +8,14 @@
public interface ICommand<TResult>
{
}

// feature #141

/// <summary>
/// Represents a command in the CQRS pattern.
/// Commands are used to change the system state and do not return a value.
/// </summary>
public interface ICommand
{
}
}
19 changes: 19 additions & 0 deletions src/Cortex.Mediator/Commands/ICommandHandler.cs
Original file line number Diff line number Diff line change
Expand Up @@ -18,4 +18,23 @@ public interface ICommandHandler<in TCommand, TResult>
/// <param name="cancellationToken">The cancellation token.</param>
Task<TResult> Handle(TCommand command, CancellationToken cancellationToken);
}



// feature #141

/// <summary>
/// Defines a handler for a command.
/// </summary>
/// <typeparam name="TCommand">The type of command being handled.</typeparam>
public interface ICommandHandler<in TCommand>
where TCommand : ICommand
{
/// <summary>
/// Handles the specified command.
/// </summary>
/// <param name="command">The command to handle.</param>
/// <param name="cancellationToken">The cancellation token.</param>
Task Handle(TCommand command, CancellationToken cancellationToken);
}
}
25 changes: 25 additions & 0 deletions src/Cortex.Mediator/Commands/ICommandPipelineBehavior.cs
Original file line number Diff line number Diff line change
Expand Up @@ -19,8 +19,33 @@ Task<TResult> Handle(
CancellationToken cancellationToken);
}


// For non returning commands
// feature #141

/// <summary>
/// Defines a pipeline behavior for wrapping command handlers.
/// </summary>
/// <typeparam name="TCommand">The type of command being handled.</typeparam>
public interface ICommandPipelineBehavior<in TCommand>
where TCommand : ICommand
{
/// <summary>
/// Handles the command and invokes the next behavior in the pipeline.
/// </summary>
Task Handle(
TCommand command,
CommandHandlerDelegate next,
CancellationToken cancellationToken);
}

/// <summary>
/// Represents a delegate that wraps the command handler execution.
/// </summary>
public delegate Task<TResult> CommandHandlerDelegate<TResult>();

/// <summary>
/// Represents a delegate that wraps the command handler execution.
/// </summary>
public delegate Task CommandHandlerDelegate();
}
61 changes: 31 additions & 30 deletions src/Cortex.Mediator/DependencyInjection/MediatorOptions.cs
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@ namespace Cortex.Mediator.DependencyInjection
public class MediatorOptions
{
internal List<Type> CommandBehaviors { get; } = new();
internal List<Type> VoidCommandBehaviors { get; } = new();
internal List<Type> QueryBehaviors { get; } = new();

public bool OnlyPublicClasses { get; set; } = true;
Expand All @@ -23,21 +24,25 @@ public MediatorOptions AddCommandPipelineBehavior<TBehavior>()
var behaviorType = typeof(TBehavior);

if (behaviorType.IsGenericTypeDefinition)
{
throw new ArgumentException("Open generic types must be registered using AddOpenCommandPipelineBehavior");
}

var implementsInterface = behaviorType
.GetInterfaces()
.Any(i => i.IsGenericType &&
i.GetGenericTypeDefinition() == typeof(ICommandPipelineBehavior<,>));
var implementsReturning =
behaviorType.GetInterfaces().Any(i => i.IsGenericType &&
i.GetGenericTypeDefinition() == typeof(ICommandPipelineBehavior<,>));

if (!implementsInterface)
{
throw new ArgumentException("Type must implement ICommandPipelineBehavior<,>");
}
var implementsNonReturning =
behaviorType.GetInterfaces().Any(i => i.IsGenericType &&
i.GetGenericTypeDefinition() == typeof(ICommandPipelineBehavior<>));

if (!implementsReturning && !implementsNonReturning)
throw new ArgumentException("Type must implement ICommandPipelineBehavior<,> or ICommandPipelineBehavior<>");

if (implementsReturning)
CommandBehaviors.Add(behaviorType);

if (implementsNonReturning)
VoidCommandBehaviors.Add(behaviorType);

CommandBehaviors.Add(behaviorType);
return this;
}

Expand All @@ -47,29 +52,25 @@ public MediatorOptions AddCommandPipelineBehavior<TBehavior>()
public MediatorOptions AddOpenCommandPipelineBehavior(Type openGenericBehaviorType)
{
if (!openGenericBehaviorType.IsGenericTypeDefinition)
{
throw new ArgumentException("Type must be an open generic type definition");
}

var implementsInterface = openGenericBehaviorType
.GetInterfaces()
.Any(i => i.IsGenericType &&
i.GetGenericTypeDefinition() == typeof(ICommandPipelineBehavior<,>));
var implementsReturning =
openGenericBehaviorType.GetInterfaces().Any(i => i.IsGenericType &&
i.GetGenericTypeDefinition() == typeof(ICommandPipelineBehavior<,>));

// For open generics, interface might not appear in GetInterfaces() yet; check by definition instead.
if (!implementsInterface &&
!(openGenericBehaviorType.IsGenericTypeDefinition &&
openGenericBehaviorType.GetGenericTypeDefinition() == openGenericBehaviorType))
{
// Fall back to checking generic arguments count to give a clear error
var ok = openGenericBehaviorType.GetGenericArguments().Length == 2;
if (!ok)
{
throw new ArgumentException("Type must implement ICommandPipelineBehavior<,>");
}
}
var implementsNonReturning =
openGenericBehaviorType.GetInterfaces().Any(i => i.IsGenericType &&
i.GetGenericTypeDefinition() == typeof(ICommandPipelineBehavior<>));

if (!implementsReturning && !implementsNonReturning)
throw new ArgumentException("Type must implement ICommandPipelineBehavior<,> or ICommandPipelineBehavior<>");

if (implementsReturning)
CommandBehaviors.Add(openGenericBehaviorType);

if (implementsNonReturning)
VoidCommandBehaviors.Add(openGenericBehaviorType);

CommandBehaviors.Add(openGenericBehaviorType);
return this;
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,8 @@ public static MediatorOptions AddDefaultBehaviors(this MediatorOptions options)
return options
// Register the open generic logging behavior for commands that return TResult
.AddOpenCommandPipelineBehavior(typeof(LoggingCommandBehavior<,>))
.AddOpenQueryPipelineBehavior(typeof(LoggingQueryBehavior<,>));
.AddOpenQueryPipelineBehavior(typeof(LoggingQueryBehavior<,>))
.AddOpenCommandPipelineBehavior(typeof(LoggingCommandBehavior<>)); // Add void command logging
}
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,14 @@ private static void RegisterHandlers(
.AsImplementedInterfaces()
.WithScopedLifetime());

// feature #141 - Register void command handlers
services.Scan(scan => scan
.FromAssemblies(assemblies)
.AddClasses(classes => classes
.AssignableTo(typeof(ICommandHandler<>)), options.OnlyPublicClasses)
.AsImplementedInterfaces()
.WithScopedLifetime());

services.Scan(scan => scan
.FromAssemblies(assemblies)
.AddClasses(classes => classes
Expand All @@ -72,6 +80,12 @@ private static void RegisterPipelineBehaviors(IServiceCollection services, Media
services.AddTransient(typeof(ICommandPipelineBehavior<,>), behaviorType);
}

// feature #141 - Register non-returning command pipeline behaviors
foreach (var behaviorType in options.VoidCommandBehaviors)
{
services.AddTransient(typeof(ICommandPipelineBehavior<>), behaviorType);
}

// Query behaviors (if needed)
foreach (var behaviorType in options.QueryBehaviors)
{
Expand Down
5 changes: 5 additions & 0 deletions src/Cortex.Mediator/IMediator.cs
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,11 @@ Task<TResult> SendCommandAsync<TCommand, TResult>(
CancellationToken cancellationToken = default)
where TCommand : ICommand<TResult>;

Task SendCommandAsync<TCommand>(
TCommand command,
CancellationToken cancellationToken = default)
where TCommand : ICommand;

Task<TResult> SendQueryAsync<TQuery, TResult>(
TQuery query,
CancellationToken cancellationToken = default)
Expand Down
40 changes: 38 additions & 2 deletions src/Cortex.Mediator/Mediator.cs
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,7 @@ public Mediator(IServiceProvider serviceProvider)
}

public async Task<TResult> SendCommandAsync<TCommand, TResult>(TCommand command, CancellationToken cancellationToken = default)
where TCommand : ICommand<TResult>
where TCommand : ICommand<TResult>
{
var handler = _serviceProvider.GetRequiredService<ICommandHandler<TCommand, TResult>>();

Expand All @@ -31,7 +31,19 @@ public async Task<TResult> SendCommandAsync<TCommand, TResult>(TCommand command,
handler = new PipelineBehaviorNextDelegate<TCommand, TResult>(behavior, handler);
}

return await handler.Handle(command, cancellationToken);
return await handler.Handle(command, cancellationToken);
}

public async Task SendCommandAsync<TCommand>(TCommand command, CancellationToken cancellationToken = default) where TCommand : ICommand
{
var handler = _serviceProvider.GetRequiredService<ICommandHandler<TCommand>>();

foreach (var behavior in _serviceProvider.GetServices<ICommandPipelineBehavior<TCommand>>().Reverse())
{
handler = new PipelineBehaviorNextDelegate<TCommand>(behavior, handler);
}

await handler.Handle(command, cancellationToken);
}

public async Task<TResult> SendQueryAsync<TQuery, TResult>(TQuery query, CancellationToken cancellationToken = default)
Expand All @@ -57,6 +69,7 @@ public async Task PublishAsync<TNotification>(
await Task.WhenAll(tasks);
}


private class PipelineBehaviorNextDelegate<TCommand, TResult> : ICommandHandler<TCommand, TResult>
where TCommand : ICommand<TResult>
{
Expand All @@ -80,6 +93,29 @@ public Task<TResult> Handle(TCommand command, CancellationToken cancellationToke
}
}

private class PipelineBehaviorNextDelegate<TCommand> : ICommandHandler<TCommand>
where TCommand : ICommand
{
private readonly ICommandPipelineBehavior<TCommand> _behavior;
private readonly ICommandHandler<TCommand> _next;

public PipelineBehaviorNextDelegate(
ICommandPipelineBehavior<TCommand> behavior,
ICommandHandler<TCommand> next)
{
_behavior = behavior;
_next = next;
}

public Task Handle(TCommand command, CancellationToken cancellationToken)
{
return _behavior.Handle(
command,
() => _next.Handle(command, cancellationToken),
cancellationToken);
}
}

private class QueryPipelineBehaviorNextDelegate<TQuery, TResult>
: IQueryHandler<TQuery, TResult>
where TQuery : IQuery<TResult>
Expand Down