Как мне написать промежуточное ПО для Rebus?

Question

Я хочу иметь возможность написать собственное промежуточное программное обеспечение для Rebus, которое будет охватывать обработчики сообщений: выполнение до и после каждого обработчика, независимо от типа сообщения.Возможно ли это с Rebus, и если да, то как мне это сделать?

Просматривая исходный код Rebus, я думаю, что должно быть возможным, поскольку библиотека очень аккуратно построена вокругКонцепция IPipeline с настраиваемыми «шагами».Но хотя выглядит так, как будто легко добавлять пользовательские шаги, я не могу найти публичного API, предоставляющего доступ к этой линии.Возможно ли изменить конвейер из клиентского кода?

То, что я ищу, по сути эквивалентно MassTransit's IFilter<>.Мы собираемся выбрать между MassTransit и Rebus (или, возможно, пользовательской реализацией поверх Microsoft.Azure.ServiceBus) для нового проекта, и на данный момент это похоже на главную функцию, отсутствующую в Rebus.

Answer 1

С помощью Rebus вы добавляете пользовательские шаги, декорируя IPipeline, что позволяет изменять конвейеры входящих и исходящих сообщений так, как вы хотите.

Чтобы было удобнее вводить шаги в определенную позициюв конвейере Rebus поставляется с PipelineStepInjector, который является декоратором, который позволяет позиционировать введенный шаг относительно другого шага.

Извлечь раздел Pipelines на вики-странице о расширяемости - показывает, как исходящий шаг можно использовать для автоматического обогащения заголовков всех исходящих сообщений.

Входящие шаги можно добавить аналогичным образом, используя только метод OnReceive вместо OnSend для добавленияшаг:

Configure.With(...)
    .(...)
    .Options(o => {
        o.Decorate<IPipeline>(c =>
        {
            var pipeline = c.Get<IPipeline>();
            var step = new YourIncomingStep();

            return new PipelineStepInjector(pipeline)
                .OnReceive(step, PipelineRelativePosition.Before, typeof(DispatchIncomingMessageStep));
        });
    })
    .Start();