rebusrebus-rabbitmq

using rabbit mq with rebus keeps publishing the events


I am using Rebus and I am subscribing to RabbitMq as explained here in the document:

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

Using Autofac container.

Registering handlers:

_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);

And in startup code:

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

Where there is a subscriber:

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

Handler code:

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

            await bus.Publish(@eventSuccess);

        }

When I execute the end point, I get continuous message as below: It continuously publishes message without executing the handler.

Can anyone help where it might be wrong.

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

And in my handler end point, I get this message even if the handler is getting executed and publishing the event (using RabbitMq transport) :

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)<---

Solution

  • Found the mistake. Was calling the bus.Subscribe from a common library in all end points.

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

    Need to be configured for the subscriber end point only. By default, it does not ignore configuration in the absence of any event handlers present in the end point. To get the correct behaviour, I deleted the queues and recreated those in RabbitMq.