NServiceBus behavior that results in incoming messages of the same type to be processed sequentially even though the endpoint is configured for concurrent processing

SequentialProcessingByMessageTypeBehavior.cs

/// <summary>
/// Usage:
///   endpointConfiguration.Pipeline.Register(new SequentialProcessingByMessageTypeBehavior(), "Processes incoming messages of the same type sequentially.");
/// </summary>
class SequentialProcessingByMessageTypeBehavior : IBehavior<IIncomingPhysicalMessageContext, IIncomingPhysicalMessageContext>
{
    TimeSpan Timeout = TimeSpan.FromSeconds(1);
    ConcurrentDictionary<string, SemaphoreSlim> semaphores = new ConcurrentDictionary<string, SemaphoreSlim>();

    public async Task Invoke(IIncomingPhysicalMessageContext context, Func<IIncomingPhysicalMessageContext, Task> next)
    {
        var type = context.MessageHeaders[Headers.EnclosedMessageTypes];
        var semaphore = semaphores.GetOrAdd(type, x => new SemaphoreSlim(1));
        var success = await semaphore.WaitAsync(Timeout)
            .ConfigureAwait(false);

        if (!success) throw new InvalidOperationException($"Could not obtain semaphore within {Timeout}");

        try
        {
            await next(context)
                .ConfigureAwait(false);
        }
        finally
        {
            semaphore.Release();
        }
    }
}