ARTEMIS-5304 Allow delayed close of federation consumers

When demand is removed from a federation consumer and the AMQP link is
drained allow the consumer to idle for a configurable time period before
closing the link. If demand returns before the idled link is closed we
can recover the link by simply granting more credit.

artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/connect/federation/AMQPFederationAddressConsumer.java

@@ -102,6 +102,11 @@ private String generateLinkName() {
              "-" + LINK_SEQUENCE_ID.incrementAndGet();
    }
 
+   @Override
+   public int getReceiverIdleTimeout() {
+      return configuration.getAddressReceiverIdleTimeout();
+   }
+
    @Override
    protected void doCreateReceiver() {
       try {
@@ -200,7 +205,9 @@ protected void doCreateReceiver() {
 
                receiver = new AMQPFederatedAddressDeliveryReceiver(session, consumerInfo, protonReceiver);
 
-               if (protonReceiver.getRemoteSource() != null) {
+               final boolean linkOpened = protonReceiver.getRemoteSource() != null;
+
+               if (linkOpened) {
                   logger.debug("AMQP Federation {} address consumer {} completed open", federation.getName(), consumerInfo);
                } else {
                   logger.debug("AMQP Federation {} address consumer {} rejected by remote", federation.getName(), consumerInfo);
@@ -209,6 +216,10 @@ protected void doCreateReceiver() {
                session.addReceiver(protonReceiver, (session, protonRcvr) -> {
                   return this.receiver;
                });
+
+               if (linkOpened && remoteOpenHandler != null) {
+                  remoteOpenHandler.accept(this);
+               }
             } catch (Exception e) {
                federation.signalError(e);
             }