DelayedRedelivery with a Consumer factory

[thurfir]

Hi,

Sorry if this is the wrong communication channel, I first posted on the google group but my post seems to have disappeared.

We would like to implement a retry process for our queues. We are looking to run retries with long delays (~1 hour) so non-blocking jobs are required.
From what I read DelayedRedelivery is the way to achieve that in masstransit?

I found a few examples on these topics about how to use DelayedRedelivery like this one: #646

however I was not able to apply the UseScheduledRedelivery to test it by myself: we are using the non-generic .Consumer() method on our endpoints, and there is no configuration parameter available.

Here is our code:

                cfg.ReceiveEndpoint(host, queue.QueueName, e =>
                {
                    // create the request consumer
                    e.Consumer(queue.ConsumerType, type => queue.ConsumerBuilder(type));

Is there another way to apply this configuration (maybe from the IConsumer) ?

Thanks for your help.

[phatboyg]

There is a way, but it isn't obvious how to use it. Let me think about this one and see if I can make it easier.

[phatboyg]

So I added extensions to support this, which work for consumers and sagas.

[thurfir]

Much appreciated @phatboyg , this is exactly what i was looking for!

We will test it in our solution today and apply it as soon as it moves to release.

[thurfir]

Hello @phatboyg , there seems to be an issue with the new extensions + the non-generic consumer, the retries never occurs:

            var busControl = Bus.Factory.CreateUsingRabbitMq(cfg =>
            {
                cfg.UseDelayedExchangeMessageScheduler();

                var host = cfg.Host(new Uri("rabbitmq://localhost"), "main", h =>
                {
                    h.Username("guest");
                    h.Password("guest");
                });

                cfg.ReceiveEndpoint(host, "mytempqueue", e =>
                {
                    e.UseDelayedRedelivery(r => r.Intervals(100, 200));

                    //e.Consumer<TestConsumer>();
                    e.Consumer(typeof(TestConsumer), type => new TestConsumer());
                });
            });

            await busControl.StartAsync();

            var uri = new Uri("rabbitmq://localhost/mytempqueue");

            var sendEndpoint = await busControl.GetSendEndpoint(uri);
            await sendEndpoint.Send(new TestRequest { Data = "ok" });

When I switch to e.Consumer<TestConsumer>(); the retries are triggered properly.

Update: Same issue with UseInMemoryScheduler + UseScheduledRedelivery.

[phatboyg]

This should be fixed in develop, will be released at the next version (might be 5.x instead of 4.1, but whatever).

[phatboyg]

There was more work required for this to work, and it's coming.

[thurfir]

Thank you @phatboyg , we really appreciate the time you are spending on this issue!

[phatboyg]

The latest develop should have this all sorted out now. There is a new extension methods to setup retry inside of redelivery.

e.UseDelayedRedelivery(...);
e.UseMessageRetry(...);

This properly nestes retry inside of redelivery.

[thurfir]

I'm a bit confused, should we replace .UseDelayedRedelivery(..) by .UseMessageRetry(..) ?
Is it the same method for .UseScheduledRedelivery(..) ?

[phatboyg]

If you want both retry and redelivery you need to use both in the order I specified above.