RabbitMQ transport check not working

[jagofett]

Hi!
First of all thanks for creating this application!

In version CAP 3.0.1, the transport checking may be response false positive results.
My application (which is a microservice) is configured to work with RabbitMQ and Postgre; if I start the application, before the RabbitMQ server is started, the following exception will throw:

None of the specified endpoints were reachable RabbitMQ.Client.Exceptions.BrokerUnreachableException: None of the specified endpoints were reachable ---> System.AggregateException: One or more errors occurred. (Connection failed) ---> RabbitMQ.Client.Exceptions.ConnectFailureException: Connection failed ---> System.Net.Internals.SocketExceptionFactory+ExtendedSocketException (10061): No connection could be made because the target machine actively refused it. 127.0.0.1:5672

Which is what we can expect, but the application launch, and after 30 seconds, the "Transport connection checking..." and "Transport connection healthy!" logs will appear - which is not true, because there is no connection to the broker, no retry to connect.

When I start the RabbitMQ server, nothing changed, the application always thinks the connection is healthy, and therefore there will be no retry. If I restart the microservice, it can successfully connect to the Rabbit broker, and work as expected.

So, steps to reproduce the issue:

1. Start microservices without starting RabbitMQ
2. AFTER all microservices have started: Start RabbitMQ
3. Send a message, publish succeeds (entry in database has succeeded status and I can see queued message in RabbitMQ management console)
4. Message is not received by consumer, even after waiting for a couple of minutes
5. Restart microservice which consumes messages: Message is received

Please see the attached log file to more information. cap_log.txt

As you can see, this issue is similar to #329.

My guess is in this case, the DotNetCore.CAP.RabbitMQ.RabbitMQConsumerClient.Subscribe method throws an exception, which is located in the RabbiMQ client, which is not handled in the ConsumerRegister class - only in case of BrokerConnectionException will the CAP retry, not in any other exception.
I think it would be better, if there is any problem starting the consumer, the health-check service should retry it.

Probably I'll send a pull request about it, because it's quite a simple modification - but in any case, thanks in advance for taking time for this issue.

[jagofett]

Since I cannot push the repo, here is what I would change in it (and it is seemed to work for me):
src/DotNetCore.CAP/Internal/IConsumerRegister.Default.cs

public void Start()
        {
            var groupingMatches = _selector.GetCandidatesMethodsOfGroupNameGrouped();

            foreach (var matchGroup in groupingMatches)
            {
                for (int i = 0; i < _options.ConsumerThreadCount; i++)
                {
                    Task.Factory.StartNew(() =>
                    {
                        try
                        {
                            using (var client = _consumerClientFactory.Create(matchGroup.Key))
                            {
                                _serverAddress = client.BrokerAddress;

                                RegisterMessageProcessor(client);

                                client.Subscribe(matchGroup.Value.Select(x => x.Attribute.Name));

                                client.Listening(_pollingDelay, _cts.Token);
                            }
                        }
                        catch (OperationCanceledException)
                        {
                            //ignore
                        }
                        catch (BrokerConnectionException e)
                        {
                            _isHealthy = false;
                            _logger.LogError(e, e.Message);
                        }
                        catch (Exception e)
                        {
                            _isHealthy = false; //it should retry every case of error.
                            _logger.LogError(e, e.Message);
                        }
                    }, _cts.Token, TaskCreationOptions.LongRunning, TaskScheduler.Default);
                }
            }
            _compositeTask = Task.CompletedTask;
        }

[yang-xiaodong]

Catching all exceptions is not a good idea. In many cases, exceptions may occur, but these exceptions are not always caused by RabbitMQ not starting.

[yang-xiaodong]

@jagofett Thanks for your report , please check whether this issue has been fixed? it works on my machine now

[jagofett]

@yang-xiaodong Thanks for the fast reply and fix! I've just tested it, and it works like a charm in my machine as well!
Could you please release a new nuget package from it, or can you tell when you can release it?
Thanks in advance!

[yang-xiaodong]

I have released version 3.0.2 to nuget and you will see it in a few minutes.

[jagofett]

Thanks!

[lhysrc]

ConsumerRegister.Restart 方法在并发情况下 Start 被调用多次，导致所有消息都被消费多次。