MassTransit JobConsumer middleware for the purpose of validating a message

Question

I would like to have an equivalent behavior but for Job Consumers :

public class MessageValidatorFilter<T> : IFilter<ConsumeContext<T>>
    where T : class
{
    private readonly ILogger<MessageValidatorFilter<T>> _logger;
    private readonly IValidator<T> _validator;

    public MessageValidatorFilter(ILogger<MessageValidatorFilter<T>> logger, IServiceProvider serviceProvider)
    {
        _logger = logger;
        _validator = serviceProvider.GetService<IValidator<T>>();
    }

    public async Task Send(ConsumeContext<T> context, IPipe<ConsumeContext<T>> next)
    {
        var validationResult = _validator is not null
            ? await _validator.ValidateAsync(context.Message, context.CancellationToken)
            : new ValidationResult();

        if (validationResult.IsValid is false)
        {
            _logger.LogError("Message validation errors: {Errors}", validationResult.Errors);

            await context.Send(
                destinationAddress: new($"queue:yourcontext-{KebabCaseEndpointNameFormatter.Instance.SanitizeName(typeof(T).Name)}-validation-errors"),
                message: new ValidationResultMessage<T>(context.Message, validationResult));

            return;
        }

        await next.Send(context);
    }

    public void Probe(ProbeContext context) { }
}

But there is no Middleware for JobConsumers, this documentation (https://masstransit-project.com/advanced/middleware/custom.html) use ConsumeContext which does not work with Job Consumers

Answer 1

You can't, it isn't supported. If you want to validate the message, you'd need to do so in the job consumer, via an injected dependency.