Использование Rabbit MQ с Rebus продолжает публиковать события - PullRequest
Новая
       45

Использование Rabbit MQ с Rebus продолжает публиковать события

0 голосов
/ 22 октября 2019

Я использую Rebus и подписываюсь на RabbitMq, как описано здесь в документе:

https://github.com/rebus-org/Rebus/wiki/RabbitMQ-transport

Использование контейнера Autofac.

Регистрация обработчиков: 

_builder.RegisterAssemblyTypes(assembly)
                        .Where(t => t.IsClass && !t.IsAbstract && t.GetInterfaces().Any(IsRebusHandler))
                        .As(GetImplementedHandlerInterfaces)
                        .InstancePerDependency()
                        .PropertiesAutowired();

Action<OptionsConfigurer> optionsConfigurer = o =>
        {
            o.SetNumberOfWorkers(2);
            o.SetMaxParallelism(30);
        };

_rebusConfig = (configurer, context) => configurer
                    .Routing(r => r.TypeBased().MapAssemblyOf<MyMessage>(destination))
                    .Transport(t=> t.UseRabbitMq(connectionString, endPointName))
                    .Options(OptionsConfigurer );
    _builder.RegisterRebus(_rebusConfig);

А в коде запуска: 

var bus = container.Resolve<IBus>()
bus.Start();

Где есть подписчик: 

var events = container.Resolve<IEnumerable<IAppEvent>>();
            foreach (var evt in events)
            {
                bus.Subscribe(evt.GetType());
            }

Код обработчика: 

public async Task Handle(FundsTransfer_InitiateFundsTransferCommand message)
        {
            FundsTransferCompletedEvent @eventSuccess = new FundsTransferCompletedEvent
            {
                    Desc=_packet.cmd.model.Desc 
                };

            await bus.Publish(@eventSuccess);

        }

Когда явыполняя конечную точку, я получаю непрерывное сообщение, как показано ниже: оно непрерывно публикует сообщение без выполнения обработчика.

Может кто-нибудь помочь, где это может быть неправильно. 

2019-10-22 08:40:33.183 +05:30 [Debug] Sending NewStructure4.FundsTransferCompletedEvent -> "NewStructure4.FundsTransferCompletedEvent, NewStructure4.Messages@RebusTopics"
2019-10-22 08:40:33.190 +05:30 [Debug] Bus Instance: RebusBus
2019-10-22 08:40:33.193 +05:30 [Debug] Dispatching "NewStructure4.FundsTransferCompletedEvent, NewStructure4.Messages" "d729e3f8-ced9-47e5-8c79-96c8a99b7473" to 1 handlers took 42 ms
2019-10-22 08:40:33.197 +05:30 [Debug] Sending NewStructure4.FundsTransferCompletedEvent -> "NewStructure4.FundsTransferCompletedEvent, NewStructure4.Messages@RebusTopics"
2019-10-22 08:40:33.209 +05:30 [Debug] Publishing message via Bus
2019-10-22 08:40:33.210 +05:30 [Debug] Dispatching "NewStructure4.FundsTransferCompletedEvent, NewStructure4.Messages" "e1dc6be7-bc18-4c21-9ed0-422ca19b2ad5" to 1 handlers took 35 ms
2019-10-22 08:40:33.212 +05:30 [Debug] Bus Context: FlexBusOmegaContext
2019-10-22 08:40:33.222 +05:30 [Debug] Publishing message via Bus
2019-10-22 08:40:33.223 +05:30 [Debug] Bus Instance: RebusBus
2019-10-22 08:40:33.224 +05:30 [Debug] Bus Context: FlexBusOmegaContext
2019-10-22 08:40:33.228 +05:30 [Debug] Sending NewStructure4.FundsTransferCompletedEvent -> "NewStructure4.FundsTransferCompletedEvent, NewStructure4.Messages@RebusTopics"
2019-10-22 08:40:33.230 +05:30 [Debug] Bus Instance: RebusBus
2019-10-22 08:40:33.233 +05:30 [Debug] Sending NewStructure4.FundsTransferCompletedEvent -> "NewStructure4.FundsTransferCompletedEvent, NewStructure4.Messages@RebusTopics"
2019-10-22 08:40:33.239 +05:30 [Debug] Dispatching "NewStructure4.FundsTransferCompletedEvent, NewStructure4.Messages" "8f1fd52c-dd87-4cdb-9341-c71e4ac0801b" to 1 handlers took 30 ms
2019-10-22 08:40:33.242 +05:30 [Debug] Dispatching "NewStructure4.FundsTransferCompletedEvent, NewStructure4.Messages" "ee1e2ccb-eeec-4802-9f5e-7767efcab678" to 1 handlers took 19 ms
2019-10-22 08:40:33.246 +05:30 [Debug] Publishing message via Bus
2019-10-22 08:40:33.253 +05:30 [Debug] Publishing message via Bus
2019-10-22 08:40:33.253 +05:30 [Debug] Bus Context: FlexBusOmegaContext
2019-10-22 08:40:33.255 +05:30 [Debug] Bus Context: FlexBusOmegaContext
2019-10-22 08:40:33.256 +05:30 [Debug] Bus Instance: RebusBus
2019-10-22 08:40:33.260 +05:30 [Debug] Bus Instance: RebusBus
2019-10-22 08:40:33.263 +05:30 [Debug] Sending NewStructure4.FundsTransferCompletedEvent -> "NewStructure4.FundsTransferCompletedEvent, NewStructure4.Messages@RebusTopics"
2019-10-22 08:40:33.267 +05:30 [Debug] Sending NewStructure4.FundsTransferCompletedEvent -> "NewStructure4.FundsTransferCompletedEvent, NewStructure4.Messages@RebusTopics"
2019-10-22 08:40:33.271 +05:30 [Debug] Dispatching "NewStructure4.FundsTransferCompletedEvent, NewStructure4.Messages" "e8a6c158-13e1-4ca6-a351-c1e6df37ac61" to 1 handlers took 17 ms
2019-10-22 08:40:33.273 +05:30 [Debug] Dispatching "NewStructure4.FundsTransferCompletedEvent, NewStructure4.Messages" "95ac2bac-a947-4757-9b24-c693909f2224" to 1 handlers took 27 ms
2019-10-22 08:40:33.285 +05:30 [Debug] Publishing message via Bus
2019-10-22 08:40:33.285 +05:30 [Debug] Publishing message via Bus
2019-10-22 08:40:33.286 +05:30 [Debug] Bus Context: FlexBusOmegaContext
2019-10-22 08:40:33.289 +05:30 [Debug] Bus Context: FlexBusOmegaContext
2019-10-22 08:40:33.291 +05:30 [Debug] Bus Instance: RebusBus
2019-10-22 08:40:33.293 +05:30 [Debug] Bus Instance: RebusBus
2019-10-22 08:40:33.300 +05:30 [Debug] Sending NewStructure4.FundsTransferCompletedEvent -> "NewStructure4.FundsTransferCompletedEvent, NewStructure4.Messages@RebusTopics"
2019-10-22 08:40:33.302 +05:30 [Debug] Sending NewStructure4.FundsTransferCompletedEvent -> "NewStructure4.FundsTransferCompletedEvent, NewStructure4.Messages@RebusTopics"
2019-10-22 08:40:33.304 +05:30 [Debug] Dispatching "NewStructure4.FundsTransferCompletedEvent, NewStructure4.Messages" "4af26345-73fd-44b1-9127-514769337c3c" to 1 handlers took 19 ms
2019-10-22 08:40:33.306 +05:30 [Debug] Dispatching "NewStructure4.FundsTransferCompletedEvent, NewStructure4.Messages" "49825f15-c7a0-4dde-a8e4-fa249b8d38d5" to 1 handlers took 21 ms
2019-10-22 08:40:33.318 +05:30 [Debug] Publishing message via Bus
2019-10-22 08:40:33.318 +05:30 [Debug] Publishing message via Bus
2019-10-22 08:40:33.320 +05:30 [Debug] Bus Context: FlexBusOmegaContext
2019-10-22 08:40:33.322 +05:30 [Debug] Bus Context: FlexBusOmegaContext
2019-10-22 08:40:33.324 +05:30 [Debug] Bus Instance: RebusBus
2019-10-22 08:40:33.329 +05:30 [Debug] Bus Instance: RebusBus
2019-10-22 08:40:33.332 +05:30 [Debug] Sending NewStructure4.FundsTransferCompletedEvent -> "NewStructure4.FundsTransferCompletedEvent, NewStructure4.Messages@RebusTopics"
2019-10-22 08:40:33.333 +05:30 [Debug] Sending NewStructure4.FundsTransferCompletedEvent -> "NewStructure4.FundsTransferCompletedEvent, NewStructure4.Messages@RebusTopics"
2019-10-22 08:40:33.336 +05:30 [Debug] Dispatching "NewStructure4.FundsTransferCompletedEvent, NewStructure4.Messages" "92803028-5c0f-47d0-bb52-5765377b03c0" to 1 handlers took 18 ms
2019-10-22 08:40:33.339 +05:30 [Debug] Dispatching "NewStructure4.FundsTransferCompletedEvent, NewStructure4.Messages" "0fac55d6-653b-462f-bc9b-26ba86a8d069" to 1 handlers took 21 ms
2019-10-22 08:40:33.348 +05:30 [Debug] Publishing message via Bus
2019-10-22 08:40:33.348 +05:30 [Debug] Publishing message via Bus
2019-10-22 08:40:33.350 +05:30 [Debug] Bus Context: FlexBusOmegaContext
2019-10-22 08:40:33.352 +05:30 [Debug] Bus Context: FlexBusOmegaContext
2019-10-22 08:40:33.354 +05:30 [Debug] Bus Instance: RebusBus
2019-10-22 08:40:33.356 +05:30 [Debug] Bus Instance: RebusBus
2019-10-22 08:40:33.361 +05:30 [Debug] Sending NewStructure4.FundsTransferCompletedEvent -> "NewStructure4.FundsTransferCompletedEvent, NewStructure4.Messages@RebusTopics"
2019-10-22 08:40:33.366 +05:30 [Debug] Sending NewStructure4.FundsTransferCompletedEvent -> "NewStructure4.FundsTransferCompletedEvent, NewStructure4.Messages@RebusTopics"
2019-10-22 08:40:33.384 +05:30 [Debug] Dispatching "NewStructure4.FundsTransferCompletedEvent, NewStructure4.Messages" "5c7e1870-7746-4053-bb1d-e4b6d1f519eb" to 1 handlers took 35 ms
2019-10-22 08:40:33.385 +05:30 [Debug] Dispatching "NewStructure4.FundsTransferCompletedEvent, NewStructure4.Messages" "1ac9d320-b026-449d-9c05-4aead10250d7" to 1 handlers took 36 ms
2019-10-22 08:40:33.405 +05:30 [Debug] Publishing message via Bus
2019-10-22 08:40:33.409 +05:30 [Debug] Bus Context: FlexBusOmegaContext
2019-10-22 08:40:33.410 +05:30 [Debug] Publishing message via Bus
2019-10-22 08:40:33.412 +05:30 [Debug] Bus Instance: RebusBus
2019-10-22 08:40:33.416 +05:30 [Debug] Bus Context: FlexBusOmegaContext
2019-10-22 08:40:33.420 +05:30 [Debug] Sending NewStructure4.FundsTransferCompletedEvent -> "NewStructure4.FundsTransferCompletedEvent, NewStructure4.Messages@RebusTopics"
2019-10-22 08:40:33.422 +05:30 [Debug] Bus Instance: RebusBus
2019-10-22 08:40:33.426 +05:30 [Debug] Dispatching "NewStructure4.FundsTransferCompletedEvent, NewStructure4.Messages" "a7090778-3c3a-41ca-bd88-fa5819f7af29" to 1 handlers took 20 ms
2019-10-22 08:40:33.429 +05:30 [Debug] Sending NewStructure4.FundsTransferCompletedEvent -> "NewStructure4.FundsTransferCompletedEvent, NewStructure4.Messages@RebusTopics"
2019-10-22 08:40:33.437 +05:30 [Debug] Publishing message via Bus
2019-10-22 08:40:33.438 +05:30 [Debug] Dispatching "NewStructure4.FundsTransferCompletedEvent, NewStructure4.Messages" "aa267f55-27d0-48da-918d-6cc6c90f0be6" to 1 handlers took 28 ms
2019-10-22 08:40:33.440 +05:30 [Debug] Bus Context: FlexBusOmegaContext
2019-10-22 08:40:33.447 +05:30 [Debug] Publishing message via Bus
2019-10-22 08:40:33.449 +05:30 [Debug] Bus Instance: RebusBus
2019-10-22 08:40:33.454 +05:30 [Debug] Bus Context: FlexBusOmegaContext
2019-10-22 08:40:33.456 +05:30 [Debug] Sending NewStructure4.FundsTransferCompletedEvent -> "NewStructure4.FundsTransferCompletedEvent, NewStructure4.Messages@RebusTopics"
2019-10-22 08:40:33.459 +05:30 [Debug] Bus Instance: RebusBus
2019-10-22 08:40:33.462 +05:30 [Debug] Dispatching "NewStructure4.FundsTransferCompletedEvent, NewStructure4.Messages" "7cfd754c-4a3c-48a3-9d9d-6ffe625e1187" to 1 handlers took 25 ms
2019-10-22 08:40:33.465 +05:30 [Debug] Sending NewStructure4.FundsTransferCompletedEvent -> "NewStructure4.FundsTransferCompletedEvent, NewStructure4.Messages@RebusTopics"
2019-10-22 08:40:33.473 +05:30 [Debug] Publishing message via Bus
2019-10-22 08:40:33.474 +05:30 [Debug] Dispatching "NewStructure4.FundsTransferCompletedEvent, NewStructure4.Messages" "c161e531-51e1-498c-88a1-98a63f9c846e" to 1 handlers took 27 ms
2019-10-22 08:40:33.476 +05:30 [Debug] Bus Context: FlexBusOmegaContext
2019-10-22 08:40:33.483 +05:30 [Debug] Bus Instance: RebusBus
2019-10-22 08:40:33.483 +05:30 [Debug] Publishing message via Bus
2019-10-22 08:40:33.485 +05:30 [Debug] Sending NewStructure4.FundsTransferCompletedEvent -> "NewStructure4.FundsTransferCompletedEvent, NewStructure4.Messages@RebusTopics"
2019-10-22 08:40:33.486 +05:30 [Debug] Bus Context: FlexBusOmegaContext
2019-10-22 08:40:33.493 +05:30 [Debug] Dispatching "NewStructure4.FundsTransferCompletedEvent, NewStructure4.Messages" "914705dc-dd7c-4ed6-8b54-bc29a96a4d5d" to 1 handlers took 19 ms
2019-10-22 08:40:33.495 +05:30 [Debug] Bus Instance: RebusBus
2019-10-22 08:40:33.503 +05:30 [Debug] Publishing message via Bus
2019-10-22 08:40:33.503 +05:30 [Debug] Sending NewStructure4.FundsTransferCompletedEvent -> "NewStructure4.FundsTransferCompletedEvent, NewStructure4.Messages@RebusTopics"
2019-10-22 08:40:33.505 +05:30 [Debug] Bus Context: FlexBusOmegaContext
2019-10-22 08:40:33.508 +05:30 [Debug] Dispatching "NewStructure4.FundsTransferCompletedEvent, NewStructure4.Messages" "6aab45ca-f875-47ec-985c-76ab26a3b2a8" to 1 handlers took 24 ms
2019-10-22 08:40:33.510 +05:30 [Debug] Bus Instance: RebusBus
2019-10-22 08:40:33.518 +05:30 [Debug] Sending NewStructure4.FundsTransferCompletedEvent -> "NewStructure4.FundsTransferCompletedEvent, NewStructure4.Messages@RebusTopics"
2019-10-22 08:40:33.518 +05:30 [Debug] Publishing message via Bus
2019-10-22 08:40:33.524 +05:30 [Debug] Dispatching "NewStructure4.FundsTransferCompletedEvent, NewStructure4.Messages" "c8dba472-27f3-46cb-8ee0-abfae49a222d" to 1 handlers took 21 ms
2019-10-22 08:40:33.526 +05:30 [Debug] Bus Context: FlexBusOmegaContext
2019-10-22 08:40:33.533 +05:30 [Debug] Publishing message via Bus
2019-10-22 08:40:33.534 +05:30 [Debug] Bus Instance: RebusBus
2019-10-22 08:40:33.537 +05:30 [Debug] Bus Context: FlexBusOmegaContext
2019-10-22 08:40:33.539 +05:30 [Debug] Sending NewStructure4.FundsTransferCompletedEvent -> "NewStructure4.FundsTransferCompletedEvent, NewStructure4.Messages@RebusTopics"
2019-10-22 08:40:33.540 +05:30 [Debug] Bus Instance: RebusBus
2019-10-22 08:40:33.544 +05:30 [Debug] Dispatching "NewStructure4.FundsTransferCompletedEvent, NewStructure4.Messages" "4f4d5bbe-a700-4562-a41e-09d2ce4e8999" to 1 handlers took 26 ms
2019-10-22 08:40:33.546 +05:30 [Debug] Sending NewStructure4.FundsTransferCompletedEvent -> "NewStructure4.FundsTransferCompletedEvent, NewStructure4.Messages@RebusTopics"
2019-10-22 08:40:33.556 +05:30 [Debug] Publishing message via Bus
2019-10-22 08:40:33.557 +05:30 [Debug] Dispatching "NewStructure4.FundsTransferCompletedEvent, NewStructure4.Messages" "40847a7a-b789-4202-bc90-32fef097f01d" to 1 handlers took 23 ms
2

И в моем конце обработчикая получаю это сообщение, даже если обработчик выполняется и публикует событие (используя транспорт RabbitMq): 

2019-10-22 10:15:37.962 +05:30 [Debug] Bus Instance: RebusBus
2019-10-22 10:15:38.079 +05:30 [Debug] Sending NewStructure4.FundsTransferCompletedEvent -> "NewStructure4.FundsTransferCompletedEvent, NewStructure4.Messages@RebusTopics"
2019-10-22 10:15:40.146 +05:30 [Information] awaiting task
2019-10-22 10:15:44.268 +05:30 [Debug] Dispatching "NewStructure4.FundsTransfer_InitiateFundsTransferCommand, NewStructure4.Messages" "c38dea6e-e77f-4900-bdd0-ebaff4cedecc" to 1 handlers took 18803 ms
2019-10-22 10:15:47.112 +05:30 [Warning] Unhandled exception 1 (FINAL) while handling message with ID "e717b3a9-02b3-47f5-b519-2f8e1fdbaf2e"
Rebus.Exceptions.MessageCouldNotBeDispatchedToAnyHandlersException: Message with ID e717b3a9-02b3-47f5-b519-2f8e1fdbaf2e and type NewStructure4.FundsTransferCompletedEvent, NewStructure4.Messages could not be dispatched to any handlers (and will not be retried under the default fail-fast settings)
   at Rebus.Pipeline.Receive.DispatchIncomingMessageStep.Process(IncomingStepContext context, Func`1 next)
   at Rebus.Sagas.LoadSagaDataStep.Process(IncomingStepContext context, Func`1 next)
   at Rebus.Pipeline.Receive.ActivateHandlersStep.Process(IncomingStepContext context, Func`1 next)
   at Rebus.Pipeline.Receive.HandleRoutingSlipsStep.Process(IncomingStepContext context, Func`1 next)
   at Rebus.Pipeline.Receive.DeserializeIncomingMessageStep.Process(IncomingStepContext context, Func`1 next)
   at Rebus.Pipeline.Receive.HandleDeferredMessagesStep.Process(IncomingStepContext context, Func`1 next)
   at Rebus.Retry.FailFast.FailFastStep.Process(IncomingStepContext context, Func`1 next)
   at Rebus.Retry.Simple.SimpleRetryStrategyStep.DispatchWithTrackerIdentifier(Func`1 next, String identifierToTrackMessageBy, ITransactionContext transactionContext, String messageId, String secondLevelMessageId)
2019-10-22 10:15:47.225 +05:30 [Error] Moving message with ID "e717b3a9-02b3-47f5-b519-2f8e1fdbaf2e" to error queue "error"
System.AggregateException: 1 unhandled exceptions (Message with ID e717b3a9-02b3-47f5-b519-2f8e1fdbaf2e and type NewStructure4.FundsTransferCompletedEvent, NewStructure4.Messages could not be dispatched to any handlers (and will not be retried under the default fail-fast settings)) ---> Rebus.Exceptions.MessageCouldNotBeDispatchedToAnyHandlersException: Message with ID e717b3a9-02b3-47f5-b519-2f8e1fdbaf2e and type NewStructure4.FundsTransferCompletedEvent, NewStructure4.Messages could not be dispatched to any handlers (and will not be retried under the default fail-fast settings)
   at Rebus.Pipeline.Receive.DispatchIncomingMessageStep.Process(IncomingStepContext context, Func`1 next)
   at Rebus.Sagas.LoadSagaDataStep.Process(IncomingStepContext context, Func`1 next)
   at Rebus.Pipeline.Receive.ActivateHandlersStep.Process(IncomingStepContext context, Func`1 next)
   at Rebus.Pipeline.Receive.HandleRoutingSlipsStep.Process(IncomingStepContext context, Func`1 next)
   at Rebus.Pipeline.Receive.DeserializeIncomingMessageStep.Process(IncomingStepContext context, Func`1 next)
   at Rebus.Pipeline.Receive.HandleDeferredMessagesStep.Process(IncomingStepContext context, Func`1 next)
   at Rebus.Retry.FailFast.FailFastStep.Process(IncomingStepContext context, Func`1 next)
   at Rebus.Retry.Simple.SimpleRetryStrategyStep.DispatchWithTrackerIdentifier(Func`1 next, String identifierToTrackMessageBy, ITransactionContext transactionContext, String messageId, String secondLevelMessageId)
   --- End of inner exception stack trace ---
---> (Inner Exception #0) Rebus.Exceptions.MessageCouldNotBeDispatchedToAnyHandlersException: Message with ID e717b3a9-02b3-47f5-b519-2f8e1fdbaf2e and type NewStructure4.FundsTransferCompletedEvent, NewStructure4.Messages could not be dispatched to any handlers (and will not be retried under the default fail-fast settings)
   at Rebus.Pipeline.Receive.DispatchIncomingMessageStep.Process(IncomingStepContext context, Func`1 next)
   at Rebus.Sagas.LoadSagaDataStep.Process(IncomingStepContext context, Func`1 next)
   at Rebus.Pipeline.Receive.ActivateHandlersStep.Process(IncomingStepContext context, Func`1 next)
   at Rebus.Pipeline.Receive.HandleRoutingSlipsStep.Process(IncomingStepContext context, Func`1 next)
   at Rebus.Pipeline.Receive.DeserializeIncomingMessageStep.Process(IncomingStepContext context, Func`1 next)
   at Rebus.Pipeline.Receive.HandleDeferredMessagesStep.Process(IncomingStepContext context, Func`1 next)
   at Rebus.Retry.FailFast.FailFastStep.Process(IncomingStepContext context, Func`1 next)
   at Rebus.Retry.Simple.SimpleRetryStrategyStep.DispatchWithTrackerIdentifier(Func`1 next, String identifierToTrackMessageBy, ITransactionContext transactionContext, String messageId, String secondLevelMessageId)<---

Ответы [ 2 ]

1 голос
/ 22 октября 2019

Нашли ошибку. Вызывается шина. Подписка из общей библиотеки во всех конечных точках. 

foreach (var evt in events)
            {
                bus.Subscribe(evt.GetType());
            }

Необходимо настроить только для конечной точки подписчика . По умолчанию он не игнорирует конфигурацию в отсутствие каких-либо обработчиков событий, присутствующих в конечной точке. Чтобы получить правильное поведение, я удалил очереди и заново создал их в RabbitMq.

1 голос
/ 22 октября 2019

  1. Не могли бы вы включить (суть) код, который публикует событие? Я подозреваю, что там как-то скрывается логическая проблема. 100

  2. Сообщение об ошибке 

2019-10-22 10:15:47.225 +05:30 [Error] Moving message with ID "e717b3a9-02b3-47f5-b519-2f8e1fdbaf2e" to error queue "error"
System.AggregateException: 1 unhandled exceptions (Message with ID e717b3a9-02b3-47f5-b519-2f8e1fdbaf2e and type NewStructure4.FundsTransferCompletedEvent, NewStructure4.Messages could not be dispatched to any handlers (and will not be retried under the default fail-fast settings)) ---> Rebus.Exceptions.MessageCouldNotBeDispatchedToAnyHandlersException: Message with ID e717b3a9-02b3-47f5-b519-2f8e1fdbaf2e and type NewStructure4.FundsTransferCompletedEvent, NewStructure4.Messages could not be dispatched to any handlers (and will not be retried under the default fail-fast settings)
   at Rebus.Pipeline.Receive.DispatchIncomingMessageStep.Process(IncomingStepContext context, Func`1 next)
   at Rebus.Sagas.LoadSagaDataStep.Process(IncomingStepContext context, Func`1 next)
(...)

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

Если вы используете встроенный активатор обработчика, это может быть что-то вроде 

activator.Handle<FundsTransferCompletedEvent>(async message => {
    // handle message in here
});

или, если вы используете контейнер IoC, он будет выглядеть по-разному, в зависимости от того, какой контейнер вы используете. В Castle Windsor Rebus предоставляет регистрационное расширение, которое позволяет сделать это: 

container.RegisterHandler<FundsTransferCompletedEventHandler>();

, где FundsTransferCompletedEventHandler будет обработчиком сообщений: 

public class FundsTransferCompletedEventHandler : IHandleMessages<FundsTransferCompletedEvent>
{
    public async Task Handle(FundsTransferCompletedEvent message)
    {
        // handle message in here
    }
}

Я надеюсь, чтоимеет смысл. 10

Я обновлю ответ, когда к вопросу будет добавлено больше деталей.

...