Brighter Command Dispatcher and Processor

The Paramore.Brighter.AsyncAPI NuGet package automatically generates AsyncAPI 3.0 JSON or YAML documents from a Brighter service's runtime configuration. This eliminates the need for manual documentation of messaging contracts by extracting information directly from the service's configuration.

Key capabilities include:

• Automatic Contract Collection: Gathers messaging contracts from subscriptions (consumers), publications (producers), and [PublicationTopic]-decorated IRequest types discovered via assembly scanning.
• Schema Generation: Automatically generates JSON Schema payloads for IRequest types to define message shapes.
• Pluggable Schema Logic: Allows developers to substitute custom schema generation logic via a pluggable interface.
• Standard Compliance: Uses the official AsyncAPI .NET SDK as the underlying document model to ensure full support for bindings, tags, and protocol-specific metadata.

Brighter provides a pipeline validation mechanism to ensure that your messaging infrastructure (Producers, Consumers, and Validation Steps) is correctly configured. This includes checking for resolvable transformers (wrap/unwrap) and ensuring that validation providers are properly registered for any declared validation steps.

Key validation behaviors:

• Transformer Resolvability: Checks if [Wrap] (on Publications) or [Unwrap] (on Subscriptions) transformers can be resolved by the DI container.
• Validation Providers: Ensures that if a validation step is present in a pipeline, a corresponding provider is registered.
• Severity Levels: New validation findings (like missing transformers) are surfaced as ValidationSeverity.Warning and do not block execution when throwOnError: true is used. However, critical issues like a subscription with no registered handler still produce a ValidationSeverity.Error and will block execution.

The Paramore.Brighter.Analyzer package provides Roslyn analyzer diagnostics to catch common pipeline configuration mistakes at compile time. These diagnostics appear as Warnings in your IDE, providing immediate feedback before runtime.

Key misconfigurations detected include:

• Incorrect Backstop Ordering: Detecting when a backstop attribute (e.g., [RejectMessageOnError]) is ordered after a resilience pipeline (e.g., [UseResiliencePipeline]). If the resilience pipeline comes first, it may catch exceptions before the backstop can act, rendering the backstop ineffective.
• Sync/Async Attribute Mismatches: Detecting when a synchronous attribute is applied to an asynchronous handler (e.g., using [RejectMessageOnError] on a RequestHandlerAsync<T>) or vice versa. In these cases, the attribute is silently ignored at runtime.
• Subscription Pump Type Mismatches: Detecting when a Subscription configuration (e.g., MessagePumpType.Reactor) is incompatible with the handler type (e.g., an async handler), which would otherwise cause a "no handlers found" error at runtime.

Note: The analyzer is distributed as a DevelopmentDependency NuGet package.

Scheduling in Brighter is split into two parts: the Producer and the Consumer.

1. Producer

The producer is responsible for scheduling the message. Depending on the implementation, this might involve creating a timer (In-Memory) or a job (Quartz/Hangfire).

• IAmAMessageScheduler: Used by IAmAMessageProducer to schedule raw Message objects. This is used if the producer doesn't natively support delays or if a specialized scheduler is provided.
• IAmARequestScheduler: Used by IAmACommandProcessor to schedule Send, Publish, and Post operations.

2. Consumer

Once a message/request is scheduled, the scheduler must eventually route it back to the correct producer. To facilitate this, Brighter uses two internal command types:

• FireSchedulerMessage: Wraps a Message.
• FireSchedulerRequest: Wraps a request with metadata like RequestType and RequestData.

A scheduler job (e.g., a Quartz IJob) handles these by calling ICommandProcessor.SendAsync to re-trigger the original intent.

public class MessageSchedulerJob(ICommandProcessor processor) : IJob
{
    public async Task ExecuteAsync(FireSchedulerMessage message)
    {
        await processor.SendAsync(message);
    }
    
    public async Task ExecuteAsync(FireSchedulerRequest message)
    {
        await processor.SendAsync(message);
    }
}

The HandlerPipelineDescription and TransformPipelineDescription classes provide a read-only, reflection-based view of your configured pipelines. This model is used by the validation and diagnostic tools to ensure pipelines are correctly constructed without the overhead of full handler instantiation.

Key Components

HandlerPipelineDescription

Represents a command or event pipeline.

• RequestType: The type of the incoming request.
• HandlerType: The main handler type.
• BeforeSteps / AfterSteps: Lists of PipelineStepDescription defining the pipeline sequence.

PipelineStepDescription

Represents an individual step (attribute) in the pipeline.

• AttributeType: The type of the attribute (e.g., RejectMessageOnErrorAsyncAttribute).
• HandlerType: The type of the handler associated with this step.
• Step: The integer position in the pipeline.
• Timing: The HandlerTiming (Before/After).
• InitializerParams: Parameters passed to the handler during initialization (e.g., policy names).

TransformPipelineDescription

Represents a transformation pipeline (mapping).

• RequestType: The type of the request.
• MapperType: The type of the mapper being used.
• IsDefaultMapper: Whether it is the default mapper.
• WrapTransforms: Outgoing transforms.
• UnwrapTransforms: Incoming transforms.

This ADR (Architectural Decision Record) documents the fixes for two critical bugs in the RabbitMQ messaging gateway within Brighter:

1. Shutdown Deadlock: Resolved a sync-over-async deadlock occurring when a ServiceActivator using the Proactor (async message pump) receives a shutdown signal. The deadlock was caused by the BrighterSynchronizationContext being blocked by synchronous waits on async operations that required the same context to complete.
2. Connection Pool Race Condition: Resolved a race condition in RmqMessageGatewayConnectionPool where a stale ConnectionShutdown event handler could inadvertently dispose of a new, active connection sharing the same endpoint credentials/host.

Key architectural changes implemented to resolve these issues:

• Added IAsyncDisposable to IAmAChannelAsync to allow proper asynchronous cleanup.
• Updated the Proactor to await Channel.DisposeAsync() upon receiving MT_QUIT.
• Applied ConfigureAwait(false) to connection pool async methods to prevent continuations from posting back to the BrighterSynchronizationContext.
• Guarded ConnectionShutdown handlers with ReferenceEquals to ensure only the correct connection instance is handled.
• Fixed a Time-of-Check to Time-of-Use (TOCTOU) vulnerability in the synchronous RemoveConnection method.
• Ensured DisposeAsync is the primary path for the RmqMessageGateway async variant.