amqp_client: Emit 'connection.blocked' in direct connections

[the-mikedavis]

Fixes #14657

Previously direct 0-9-1 connections did not notice when memory or disk alarms were set. This could allow an 0-9-1 shovel where the destination is a direct connection to completely overload a broker which is already in alarm. With this change, direct connections register the connection process with rabbit_alarm and emit connection.blocked and
connection.unblocked to the blocked handler if one is registered. rabbit_amqp091_shovel already respects the connection.blocked, so the destination will not receive any messages.

[ansd]

Just in case it helps, the following shows how to easily trigger memory and/or disk alarms in a test case:

rabbitmq-server/deps/rabbit/test/amqp_client_SUITE.erl

Lines 3200 to 3234 in 2303570

	%% Set memory alarm before beginning the session.
	DefaultWatermark = rpc(Config, vm_memory_monitor, get_vm_memory_high_watermark, []),
	ok = rpc(Config, vm_memory_monitor, set_vm_memory_high_watermark, [0]),
	timer:sleep(100),
	{ok, Session1} = amqp10_client:begin_session_sync(Connection),
	Address = rabbitmq_amqp_address:queue(QName),
	{ok, Sender} = amqp10_client:attach_sender_link(Session1, <<"test-sender">>, Address, unsettled),
	%% We should still receive link credit since the target queue is fine.
	ok = wait_for_credit(Sender),
	%% However, RabbitMQ's incoming window shouldn't allow our client to send any TRANSFER.
	%% In other words, the client is limited by session flow control, but not by link flow control.
	Tag1 = <<"tag1">>,
	Msg1 = amqp10_msg:new(Tag1, <<"m1">>, false),
	?assertEqual({error, remote_incoming_window_exceeded},
	amqp10_client:send_msg(Sender, Msg1)),
	%% Set additionally disk alarm.
	DefaultDiskFreeLimit = rpc(Config, rabbit_disk_monitor, get_disk_free_limit, []),
	ok = rpc(Config, rabbit_disk_monitor, set_disk_free_limit, [999_000_000_000_000]), % 999 TB
	timer:sleep(100),
	?assertEqual({error, remote_incoming_window_exceeded},
	amqp10_client:send_msg(Sender, Msg1)),
	%% Clear memory alarm.
	ok = rpc(Config, vm_memory_monitor, set_vm_memory_high_watermark, [DefaultWatermark]),
	timer:sleep(100),
	?assertEqual({error, remote_incoming_window_exceeded},
	amqp10_client:send_msg(Sender, Msg1)),
	%% Clear disk alarm.
	ok = rpc(Config, rabbit_disk_monitor, set_disk_free_limit, [DefaultDiskFreeLimit]),
	timer:sleep(100),

[the-mikedavis]

Thanks for the pointers @ansd & @gomoripeti! I moved the existing tests for alarms and 0-9-1 shovels to a new suite and added a group there that uses a direct connection