How to check the connection to a queue?

[aeldhuset]

Hi there,

A project I'm working on is using Helsenorge.Messaging (LinkFactoryPool → IAmqpReceiver, IAmqpSender) for receiving and sending messages using AMQP. We would like to implement an AMQP health check, i.e. some logic checking whether the connection to each of our configured queues is working (without actually removing any message from the queue or adding a message to the queue).

May you suggest a best approach for how to do this?

[aeldhuset]

Inspired by this example, our work-in-progress implementation for checking the connection for a queue is as follows:

var receiverClient = await _linkFactoryPool.CreateCachedMessageReceiverAsync(queue.QueueName);

try
{
	var message = await receiverClient.ReceiveAsync(TimeSpan.FromSeconds(2));

	if (message != null)
	{
		await message.RelaseAsync();
	}

	//...
}
catch (Exception ex)
{
	//...
}

where _linkFactoryPool is of type LinkFactoryPool.

This seems to be working fine when testing locally.

Yet, it would feel safer if a connection check could have been performed without the need of actually trying to retrieve anything from the queue (or sending anything to the queue). Is that possible?

If not: When calling IAmqpReceiver.ReceiveAsync() followed by IAmqpMessage.RelaseAsync() like above, is it guaranteed that every message "received" by the health check logic will be released back to the queue, so that the real receiver will receive the same message on the same queue (queue.QueueName) sometime after it has been released by the health check logic?

[aeldhuset]

Thank you for your informative reply and suggestion, Thor Halvor.

I have now implemented the GetMessageCount approach, and it seems to cover most of what we would like to cover.

We would like for the health check to catch all possibly wrong configurations; i.e. if any erroneous configuration causes a call to AmqpReceiver.ReceiveAsync(...) or AmqpSender.SendAsync(...) to fail, the health check should also fail.

As per now, I can seemingly use whichever value for Port and Exchange for the ConnectionString object provided to the QueueClient, without causing a call to GetMessageCount to fail. But with such erroneous config, calls to AmqpReceiver.ReceiveAsync(...) or AmqpSender.SendAsync(...) do fail.

Do you have any thoughts on how to check the validity of the combined config, including the Port and Exchange values?

[thorhalvor]

you can construct ConnectionString based on AmqpSettings:Connectionstring? then they will always be synced.
or the other way around : construct a MessagingSettings:AmqpSettings.ConnectionString based on the Connectionstring model used in Adminlib

[aeldhuset]

I agree, that is a good idea. That is not what I was asking about, though. Let me rephrase:

If the connection string I use for both AmqpSettings and for QueueConfig contains an erroneous value for Port or Exchange, an attempt to receive or send messages from/to any queue will throw an exception, but a call to GetMessageCount will not throw any exception. Ideally, we would like to be able to identify any faulty values for Port or Exchange without having to receive or send anything from/to any queue. Do you know if that is possible?

[thorhalvor]

if value for Port is incorrect then nothing should work. not even GetMessageCount
if Exchange is wrong, you will not get error when fetching messagecount. That is true. Since you connect directly to queue (not through exchange). I do not think you are able to check if exchange exists and is correct without sending a message the way it is implemented now. It is possible to send a Test-message though, -to HelseNorge or to yourself...

another approach can be to set a property every time you send and receive message, and expose this property to your Healthcheck-module.

If you are integrating with HelseNorge, please log into one of the Slack-channels for easier discussions.

[aeldhuset]

I agree that GetMessageCount should not have worked with an incorrect Port value. After inspecting the source code of QueueClient further, I see that the Port property is never set upon the creation of the ConnectionFactory object, and hence the default port value (5671) is being used regardless of which port is provided in the connection string.

Thank you for the explanation regarding Exchange. You are absolutely right about GetMessageCount naturally not being a viable approach for verifying the exchange value.

[kennethmyhra]

@aeldhuset I haven't tested this approach myself, but can you try to create a connection using the RabbitMQ Client library and calling the method CreateModel() on the IConnection interface.

For reference: https://github.com/Xabaril/AspNetCore.Diagnostics.HealthChecks/blob/2f6a10f89e3c6d97f232f4157a80e3a9e1470dc5/src/HealthChecks.Rabbitmq/RabbitMQHealthCheck.cs#L34

[aeldhuset]

Thank you for the suggestion! Inspired by that approach, the outline of our health check now looks like the following:

• Create the URI (format: "uriScheme://username:password@host:port")
• Based on the URI: create a ConnectionFactory object
• On the ConnectionFactory object: .CreateConnection() → creates an IConnection object
  • This action will fail if either of the URI scheme, user name, password, host name, port values are wrong
• On the IConnection object: .CreateModel() → creates an IModel object
• On the IModel object:
  • .ExchangeDeclarePassive(exchange) → verifies the exchange's existence
  • .MessageCount(queueName) → verifies each queue's existence

With this, we are able to verify the values of URI scheme, user name, password, host name, port, exchange and queue names.

[thorhalvor]

Hi, calling the xxxPassive methods as a good approach. Remember they throw exception and closes channel if not exist.
maybe instead of calling MessageCount you can call the QueueDeclarePassive and you get messagecount in response. then you can check the "consumercount" also. -and maybe log it as warning. if the consumercount is >1 and you are only supposed to have 1 consumer, then something is missconfigured somewhere and other applications "stealing" your messages...
name of exchange is case-sensitive, so this ExchangeDeclarePassive will help alot of people. ex" ...TEST Test Prod PROD..."

[aeldhuset]

That sounds like a good idea, thank you for the extra suggestions!

[ Edit: Every method call is wrapped in try/catch blocks, and both the IConnection and IModel objects are disposed of and recreated if they are either null or closed when the health check is triggered. So the exception and closing scenarios should be covered ]