content: /*
   * Licensed to the Apache Software Foundation (ASF) under one or more
   * contributor license agreements. See the NOTICE file distributed with
   * this work for additional information regarding copyright ownership.
   * The ASF licenses this file to You under the Apache License, Version 2.0
   * (the "License"); you may not use this file except in compliance with
   * the License. You may obtain a copy of the License at
   *
   *     http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  
  package org.apache.activemq.artemis.tests.integration.amqp.connect;
  
  import static org.apache.activemq.artemis.protocol.amqp.connect.AMQPBrokerConnectionConstants.BROKER_CONNECTION_INFO;
  import static org.apache.activemq.artemis.protocol.amqp.connect.AMQPBrokerConnectionConstants.CONNECTION_NAME;
  import static org.apache.activemq.artemis.protocol.amqp.connect.AMQPBrokerConnectionConstants.NODE_ID;
  import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationConstants.ADDRESS_AUTO_DELETE;
  import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationConstants.ADDRESS_AUTO_DELETE_DELAY;
  import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationConstants.ADDRESS_AUTO_DELETE_MSG_COUNT;
  import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationConstants.ADDRESS_ENABLE_DIVERT_BINDINGS;
  import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationConstants.ADDRESS_EXCLUDES;
  import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationConstants.ADDRESS_INCLUDES;
  import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationConstants.ADDRESS_MAX_HOPS;
  import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationConstants.ADDRESS_RECEIVER_IDLE_TIMEOUT;
  import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationConstants.ADD_ADDRESS_POLICY;
  import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationConstants.ADD_QUEUE_POLICY;
  import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationConstants.FEDERATION_BASE_VALIDATION_ADDRESS;
  import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationConstants.FEDERATION_CONFIGURATION;
  import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationConstants.FEDERATION_CONTROL_LINK;
  import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationConstants.FEDERATION_CONTROL_LINK_PREFIX;
  import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationConstants.FEDERATION_EVENTS_LINK_PREFIX;
  import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationConstants.FEDERATION_EVENT_LINK;
  import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationConstants.FEDERATION_NAME;
  import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationConstants.FEDERATION_V1;
  import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationConstants.FEDERATION_V2;
  import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationConstants.FEDERATION_VERSION;
  import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationConstants.IGNORE_ADDRESS_BINDING_FILTERS;
  import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationConstants.IGNORE_QUEUE_CONSUMER_FILTERS;
  import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationConstants.IGNORE_QUEUE_CONSUMER_PRIORITIES;
  import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationConstants.LARGE_MESSAGE_THRESHOLD;
  import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationConstants.LINK_ATTACH_TIMEOUT;
  import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationConstants.OPERATION_TYPE;
  import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationConstants.POLICY_NAME;
  import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationConstants.PULL_RECEIVER_BATCH_SIZE;
  import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationConstants.QUEUE_EXCLUDES;
  import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationConstants.QUEUE_INCLUDES;
  import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationConstants.QUEUE_INCLUDE_FEDERATED;
  import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationConstants.QUEUE_PRIORITY_ADJUSTMENT;
  import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationConstants.QUEUE_RECEIVER_IDLE_TIMEOUT;
  import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationConstants.RECEIVER_CREDITS;
  import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationConstants.RECEIVER_CREDITS_LOW;
  import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationConstants.RECEIVER_DRAIN_ON_TRANSIENT_DELIVERY_ERRORS;
  import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationConstants.RECEIVER_LINK_QUIESCE_TIMEOUT;
  import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationConstants.RECEIVER_QUIESCE_TIMEOUT;
  import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationConstants.USE_MODIFIED_FOR_TRANSIENT_DELIVERY_ERRORS;
  import static org.hamcrest.CoreMatchers.allOf;
  import static org.hamcrest.CoreMatchers.containsString;
  import static org.hamcrest.CoreMatchers.notNullValue;
  import static org.hamcrest.CoreMatchers.nullValue;
  import static org.junit.jupiter.api.Assertions.assertEquals;
  import static org.junit.jupiter.api.Assertions.assertFalse;
  import static org.junit.jupiter.api.Assertions.assertNotNull;
  import static org.junit.jupiter.api.Assertions.assertTrue;
  import static org.junit.jupiter.api.Assertions.fail;
  
  import java.lang.invoke.MethodHandles;
  import java.net.URI;
  import java.util.ArrayList;
  import java.util.HashMap;
  import java.util.LinkedHashMap;
  import java.util.List;
  import java.util.Map;
  import java.util.Random;
  import java.util.concurrent.ForkJoinPool;
  import java.util.concurrent.TimeUnit;
  
  import javax.jms.Connection;
  import javax.jms.ConnectionFactory;
  import javax.jms.JMSException;
  import javax.jms.Session;
  
  import org.apache.activemq.artemis.api.core.ActiveMQException;
  import org.apache.activemq.artemis.api.core.SimpleString;
  import org.apache.activemq.artemis.api.core.management.BrokerConnectionControl;
  import org.apache.activemq.artemis.api.core.management.ResourceNames;
  import org.apache.activemq.artemis.core.config.amqpBrokerConnectivity.AMQPBrokerConnectConfiguration;
  import org.apache.activemq.artemis.core.config.amqpBrokerConnectivity.AMQPFederatedBrokerConnectionElement;
  import org.apache.activemq.artemis.core.config.amqpBrokerConnectivity.AMQPFederationAddressPolicyElement;
  import org.apache.activemq.artemis.core.config.amqpBrokerConnectivity.AMQPFederationQueuePolicyElement;
  import org.apache.activemq.artemis.core.server.ActiveMQServer;
  import org.apache.activemq.artemis.core.server.Queue;
  import org.apache.activemq.artemis.protocol.amqp.connect.federation.ActiveMQServerAMQPFederationPlugin;
  import org.apache.activemq.artemis.protocol.amqp.federation.FederationConsumerInfo;
  import org.apache.activemq.artemis.protocol.amqp.proton.AmqpSupport;
  import org.apache.activemq.artemis.tests.integration.amqp.AmqpClientTestSupport;
  import org.apache.activemq.artemis.tests.util.CFUtil;
  import org.apache.activemq.artemis.tests.util.Wait;
  import org.apache.qpid.proton.amqp.transport.AmqpError;
  import org.apache.qpid.protonj2.test.driver.ProtonTestClient;
  import org.apache.qpid.protonj2.test.driver.ProtonTestServer;
  import org.apache.qpid.protonj2.test.driver.matchers.messaging.MessageAnnotationsMatcher;
  import org.apache.qpid.protonj2.test.driver.matchers.transport.TransferPayloadCompositeMatcher;
  import org.apache.qpid.protonj2.test.driver.matchers.types.EncodedAmqpValueMatcher;
  import org.hamcrest.Matchers;
  import org.jgroups.util.UUID;
  import org.junit.jupiter.api.Test;
  import org.junit.jupiter.api.Timeout;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  /**
   * Tests basic connect handling of the AMQP federation feature.
   */
  public class AMQPFederationConnectTest extends AmqpClientTestSupport {
  
     private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
  
     @Override
     protected ActiveMQServer createServer() throws Exception {
        // Creates the broker used to make the outgoing connection. The port passed is for
        // that brokers acceptor. The test server connected to by the broker binds to a random port.
        return createServer(AMQP_PORT, false);
     }
  
     @Test
     @Timeout(20)
     public void testBrokerConnectsWithAnonymous() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
           peer.expectSASLAnonymousConnect("PLAIN", "ANONYMOUS");
           peer.expectOpen().respond();
           peer.expectBegin().respond();
           peer.start();
  
           final URI remoteURI = peer.getServerURI();
           logger.info("Connect test started, peer listening on: {}", remoteURI);
  
           // No user or pass given, it will have to select ANONYMOUS even though PLAIN also offered
           AMQPBrokerConnectConfiguration amqpConnection =
                 new AMQPBrokerConnectConfiguration(getTestName(), "tcp://" + remoteURI.getHost() + ":" + remoteURI.getPort());
           amqpConnection.setReconnectAttempts(0);// No reconnects
           server.getConfiguration().addAMQPConnection(amqpConnection);
           server.start();
  
           peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
        }
     }
  
     @Test
     @Timeout(20)
     public void testFederatedBrokerConnectsWithPlain() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
           peer.expectSASLPlainConnect("user", "pass", "PLAIN", "ANONYMOUS");
           peer.expectOpen().respond();
           peer.expectBegin().respond();
           peer.start();
  
           final URI remoteURI = peer.getServerURI();
           logger.info("Connect test started, peer listening on: {}", remoteURI);
  
           AMQPBrokerConnectConfiguration amqpConnection =
                 new AMQPBrokerConnectConfiguration(getTestName(), "tcp://" + remoteURI.getHost() + ":" + remoteURI.getPort());
           amqpConnection.setReconnectAttempts(0);// No reconnects
           server.getConfiguration().addAMQPConnection(amqpConnection);
           amqpConnection.setUser("user");
           amqpConnection.setPassword("pass");
           server.start();
  
           peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
        }
     }
  
     @Test
     @Timeout(20)
     public void testBrokerConnectsAndCreateManagementResource() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
           peer.expectSASLPlainConnect("testUser", "pass", "PLAIN", "ANONYMOUS");
           peer.expectOpen().respond();
           peer.expectBegin().respond();
           peer.start();
  
           final URI remoteURI = peer.getServerURI();
           logger.info("Connect test started, peer listening on: {}", remoteURI);
  
           // No user or pass given, it will have to select ANONYMOUS even though PLAIN also offered
           AMQPBrokerConnectConfiguration amqpConnection =
                 new AMQPBrokerConnectConfiguration(getTestName(), "tcp://" + remoteURI.getHost() + ":" + remoteURI.getPort());
           amqpConnection.setReconnectAttempts(0);// No reconnects
           amqpConnection.setRetryInterval(100);
           amqpConnection.setUser("testUser");
           amqpConnection.setPassword("pass");
           server.getConfiguration().addAMQPConnection(amqpConnection);
           server.start();
  
           peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
  
           final BrokerConnectionControl brokerConnection = (BrokerConnectionControl)
              server.getManagementService().getResource(ResourceNames.BROKER_CONNECTION + getTestName());
  
           assertNotNull(brokerConnection);
           assertTrue(brokerConnection.isConnected());
           assertTrue(brokerConnection.isStarted());
           assertNotNull(brokerConnection.getUri());
           assertEquals(getTestName(), brokerConnection.getName());
           assertEquals("testUser", brokerConnection.getUser());
           assertEquals(0, brokerConnection.getReconnectAttempts());
           assertEquals(100, brokerConnection.getRetryInterval());
           assertEquals("AMQP", brokerConnection.getProtocol());
  
           peer.expectClose().optional();
           peer.expectConnectionToDrop();
  
           brokerConnection.stop();
  
           peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
  
           // Reflects state change but no NPE when connection is stopped.
           assertFalse(brokerConnection.isConnected());
           assertFalse(brokerConnection.isStarted());
           assertNotNull(brokerConnection.getUri());
           assertEquals(getTestName(), brokerConnection.getName());
           assertEquals("testUser", brokerConnection.getUser());
           assertEquals(0, brokerConnection.getReconnectAttempts());
           assertEquals(100, brokerConnection.getRetryInterval());
           assertEquals("AMQP", brokerConnection.getProtocol());
  
           peer.expectSASLPlainConnect("testUser", "pass", "PLAIN", "ANONYMOUS");
           peer.expectOpen().respond();
           peer.expectBegin().respond();
  
           brokerConnection.start();
  
           peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
  
           assertTrue(brokerConnection.isConnected());
           assertTrue(brokerConnection.isStarted());
  
           peer.close();
        }
     }
  
     @Test
     @Timeout(20)
     public void testFederationConfiguredCreatesControlLink() throws Exception {
        final int AMQP_MIN_LARGE_MESSAGE_SIZE = 10_000;
        final int AMQP_CREDITS = 100;
        final int AMQP_CREDITS_LOW = 50;
        final int AMQP_RECEIVER_QUIESCE_TIMEOUT = 10_000;
        final int AMQP_ADDRESS_RECEIVER_IDLE_TIMEOUT = 10_000;
        final int AMQP_QUEUE_RECEIVER_IDLE_TIMEOUT = 7_000;
        final int AMQP_PULL_CREDITS_BATCH = 50;
        final int AMQP_LINK_ATTACH_TIMEOUT = 60;
        final boolean AMQP_TUNNEL_CORE_MESSAGES = false;
        final boolean AMQP_INGNORE_CONSUMER_FILTERS = false;
        final boolean AMQP_INGNORE_CONSUMER_PRIORITIES = false;
        final boolean AMQP_INGNORE_ADDRESS_BINDING_FILTERS = false;
        final boolean AMQP_USE_MODIFIED_FOR_TRANSIENT_DELIVERY_ERRORS = false;
        final int AMQP_RECEIVER_LINK_QUIESCE_TIMEOUT = 250_000;
        final boolean AMQP_RECEIVER_DRAIN_ON_TRANSIENT_DELIVERY_ERRORS = false;
  
        final Map<String, Object> federationConfiguration = new HashMap<>();
        federationConfiguration.put(RECEIVER_CREDITS, AMQP_CREDITS);
        federationConfiguration.put(RECEIVER_CREDITS_LOW, AMQP_CREDITS_LOW);
        federationConfiguration.put(RECEIVER_QUIESCE_TIMEOUT, AMQP_RECEIVER_QUIESCE_TIMEOUT);
        federationConfiguration.put(ADDRESS_RECEIVER_IDLE_TIMEOUT, AMQP_ADDRESS_RECEIVER_IDLE_TIMEOUT);
        federationConfiguration.put(QUEUE_RECEIVER_IDLE_TIMEOUT, AMQP_QUEUE_RECEIVER_IDLE_TIMEOUT);
        federationConfiguration.put(PULL_RECEIVER_BATCH_SIZE, AMQP_PULL_CREDITS_BATCH);
        federationConfiguration.put(LARGE_MESSAGE_THRESHOLD, AMQP_MIN_LARGE_MESSAGE_SIZE);
        federationConfiguration.put(LINK_ATTACH_TIMEOUT, AMQP_LINK_ATTACH_TIMEOUT);
        federationConfiguration.put(IGNORE_QUEUE_CONSUMER_FILTERS, AMQP_INGNORE_CONSUMER_FILTERS);
        federationConfiguration.put(IGNORE_QUEUE_CONSUMER_PRIORITIES, AMQP_INGNORE_CONSUMER_PRIORITIES);
        federationConfiguration.put(IGNORE_ADDRESS_BINDING_FILTERS, AMQP_INGNORE_ADDRESS_BINDING_FILTERS);
        federationConfiguration.put(AmqpSupport.TUNNEL_CORE_MESSAGES, AMQP_TUNNEL_CORE_MESSAGES);
        federationConfiguration.put(USE_MODIFIED_FOR_TRANSIENT_DELIVERY_ERRORS, AMQP_USE_MODIFIED_FOR_TRANSIENT_DELIVERY_ERRORS);
        federationConfiguration.put(RECEIVER_LINK_QUIESCE_TIMEOUT, AMQP_RECEIVER_LINK_QUIESCE_TIMEOUT);
        federationConfiguration.put(RECEIVER_DRAIN_ON_TRANSIENT_DELIVERY_ERRORS, AMQP_RECEIVER_DRAIN_ON_TRANSIENT_DELIVERY_ERRORS);
  
        final String controlLinkAddress = "test-control-address";
  
        try (ProtonTestServer peer = new ProtonTestServer()) {
           peer.expectSASLAnonymousConnect("PLAIN", "ANONYMOUS");
           peer.expectOpen().respond();
           peer.expectBegin().respond();
           peer.expectAttach().ofSender()
                              .withDesiredCapability(FEDERATION_CONTROL_LINK.toString())
                              .withName(allOf(containsString("federation-"), containsString("myFederation")))
                              .withProperty(FEDERATION_VERSION.toString(), FEDERATION_V2)
                              .withProperty(FEDERATION_NAME.toString(), "myFederation")
                              .withProperty(FEDERATION_CONFIGURATION.toString(), federationConfiguration)
                              .withTarget().withDynamic(true)
                                           .withCapabilities("temporary-topic")
                              .and()
                              .respond()
                              .withTarget().withAddress(controlLinkAddress)
                              .and()
                              .withOfferedCapabilities(FEDERATION_CONTROL_LINK.toString())
                              .withProperty(FEDERATION_VERSION.toString(), FEDERATION_V2);
           peer.start();
  
           final URI remoteURI = peer.getServerURI();
           logger.info("Connect test started, peer listening on: {}", remoteURI);
  
           final AMQPBrokerConnectConfiguration amqpConnection = new AMQPBrokerConnectConfiguration(
              getTestName(), "tcp://" + remoteURI.getHost() + ":" + remoteURI.getPort() +
              "?amqpCredits=" + AMQP_CREDITS + "&amqpLowCredits=" + AMQP_CREDITS_LOW +
              "&amqpMinLargeMessageSize=" + AMQP_MIN_LARGE_MESSAGE_SIZE);
           amqpConnection.setReconnectAttempts(0);// No reconnects
           final AMQPFederatedBrokerConnectionElement federation = new AMQPFederatedBrokerConnectionElement("myFederation");
           federation.addProperty(LINK_ATTACH_TIMEOUT, AMQP_LINK_ATTACH_TIMEOUT);
           federation.addProperty(AmqpSupport.TUNNEL_CORE_MESSAGES, Boolean.toString(AMQP_TUNNEL_CORE_MESSAGES));
           federation.addProperty(PULL_RECEIVER_BATCH_SIZE, AMQP_PULL_CREDITS_BATCH);
           federation.addProperty(RECEIVER_QUIESCE_TIMEOUT, AMQP_RECEIVER_QUIESCE_TIMEOUT);
           federation.addProperty(ADDRESS_RECEIVER_IDLE_TIMEOUT, AMQP_ADDRESS_RECEIVER_IDLE_TIMEOUT);
           federation.addProperty(QUEUE_RECEIVER_IDLE_TIMEOUT, AMQP_QUEUE_RECEIVER_IDLE_TIMEOUT);
           federation.addProperty(IGNORE_QUEUE_CONSUMER_PRIORITIES, Boolean.toString(AMQP_INGNORE_CONSUMER_PRIORITIES));
           federation.addProperty(IGNORE_ADDRESS_BINDING_FILTERS, Boolean.toString(AMQP_INGNORE_ADDRESS_BINDING_FILTERS));
           federation.addProperty(USE_MODIFIED_FOR_TRANSIENT_DELIVERY_ERRORS, Boolean.toString(AMQP_USE_MODIFIED_FOR_TRANSIENT_DELIVERY_ERRORS));
           federation.addProperty(RECEIVER_LINK_QUIESCE_TIMEOUT, AMQP_RECEIVER_LINK_QUIESCE_TIMEOUT);
           federation.addProperty(RECEIVER_DRAIN_ON_TRANSIENT_DELIVERY_ERRORS, Boolean.toString(AMQP_RECEIVER_DRAIN_ON_TRANSIENT_DELIVERY_ERRORS));
           amqpConnection.addElement(federation);
           server.getConfiguration().addAMQPConnection(amqpConnection);
           server.start();
  
           peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
  
           Wait.assertTrue(() -> server.locateQueue(FEDERATION_BASE_VALIDATION_ADDRESS +
                                                    "." + FEDERATION_CONTROL_LINK_PREFIX +
                                                    "." + controlLinkAddress) != null);
  
           peer.close();
        }
     }
  
     @Test
     @Timeout(20)
     public void testFederationCreatesControlLinkAndContinuesIfConnectedToOldVersion() throws Exception {
        final String controlLinkAddress = "test-control-address";
  
        try (ProtonTestServer peer = new ProtonTestServer()) {
           peer.expectSASLAnonymousConnect("PLAIN", "ANONYMOUS");
           peer.expectOpen().respond();
           peer.expectBegin().respond();
           peer.expectAttach().ofSender()
                              .withDesiredCapability(FEDERATION_CONTROL_LINK.toString())
                              .withProperty(FEDERATION_VERSION.toString(), FEDERATION_V2)
                              .respond()
                              .withTarget().withAddress(controlLinkAddress)
                              .and()
                              .withProperty(FEDERATION_VERSION.toString(), FEDERATION_V1)
                              .withOfferedCapabilities(FEDERATION_CONTROL_LINK.toString());
           peer.start();
  
           final URI remoteURI = peer.getServerURI();
           logger.info("Connect test started, peer listening on: {}", remoteURI);
  
           AMQPBrokerConnectConfiguration amqpConnection =
                 new AMQPBrokerConnectConfiguration(getTestName(), "tcp://" + remoteURI.getHost() + ":" + remoteURI.getPort());
           amqpConnection.setReconnectAttempts(0);// No reconnects
           amqpConnection.addElement(new AMQPFederatedBrokerConnectionElement(getTestName()));
           server.getConfiguration().addAMQPConnection(amqpConnection);
           server.start();
  
           peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
  
           Wait.assertTrue(() -> server.locateQueue(FEDERATION_BASE_VALIDATION_ADDRESS +
                                                    "." + FEDERATION_CONTROL_LINK_PREFIX +
                                                    "." + controlLinkAddress) != null);
  
           peer.close();
        }
     }
  
     @Test
     @Timeout(20)
     public void testFederationCreatesControlLinkAndClosesConnectionIfCapabilityIsAbsent() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
           peer.expectSASLAnonymousConnect("PLAIN", "ANONYMOUS");
           peer.expectOpen().respond();
           peer.expectBegin().respond();
           peer.expectAttach().ofSender().withDesiredCapability(FEDERATION_CONTROL_LINK.toString()).respond();
           peer.expectClose().optional(); // Can sometimes be sent
           peer.expectConnectionToDrop();
           peer.start();
  
           final URI remoteURI = peer.getServerURI();
           logger.info("Connect test started, peer listening on: {}", remoteURI);
  
           AMQPBrokerConnectConfiguration amqpConnection =
                 new AMQPBrokerConnectConfiguration(getTestName(), "tcp://" + remoteURI.getHost() + ":" + remoteURI.getPort());
           amqpConnection.setReconnectAttempts(0);// No reconnects
           amqpConnection.addElement(new AMQPFederatedBrokerConnectionElement(getTestName()));
           server.getConfiguration().addAMQPConnection(amqpConnection);
           server.start();
  
           peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
        }
     }
  
     @Test
     @Timeout(20)
     public void testFederationCreatesControlLinkAndClosesConnectionDetachIndicatesNotAuthorized() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
           peer.expectSASLAnonymousConnect("PLAIN", "ANONYMOUS");
           peer.expectOpen().respond();
           peer.expectBegin().respond();
           peer.expectAttach().ofSender().withDesiredCapability(FEDERATION_CONTROL_LINK.toString())
                                         .respond()
                                         .withOfferedCapabilities(FEDERATION_CONTROL_LINK.toString())
                                         .withSource()
                                         .also()
                                         .withNullTarget();
           peer.remoteDetach().withErrorCondition("amqp:unauthorized-access", "Not authroized").queue();
           peer.expectDetach().optional();
           peer.expectClose().optional();
           peer.expectConnectionToDrop();
           // Broker reconnect and allow it to attach this time.
           peer.expectSASLAnonymousConnect("PLAIN", "ANONYMOUS");
           peer.expectOpen().respond();
           peer.expectBegin().respond();
           peer.expectAttach().ofSender()
                              .withTarget().withDynamic(true).and()
                              .withDesiredCapability(FEDERATION_CONTROL_LINK.toString())
                              .respondInKind()
                              .withTarget().withAddress("dynamic-name");
           peer.start();
  
           final URI remoteURI = peer.getServerURI();
           logger.info("Connect test started, peer listening on: {}", remoteURI);
  
           AMQPBrokerConnectConfiguration amqpConnection =
                 new AMQPBrokerConnectConfiguration(getTestName(), "tcp://" + remoteURI.getHost() + ":" + remoteURI.getPort());
           amqpConnection.setReconnectAttempts(1);// One reconnects
           amqpConnection.setRetryInterval(200);
           amqpConnection.addElement(new AMQPFederatedBrokerConnectionElement(getTestName()));
           server.getConfiguration().addAMQPConnection(amqpConnection);
           server.start();
  
           peer.waitForScriptToComplete(10, TimeUnit.SECONDS);
  
           Wait.assertTrue(() -> server.locateQueue(FEDERATION_BASE_VALIDATION_ADDRESS +
                                                    "." + FEDERATION_CONTROL_LINK_PREFIX +
                                                    ".dynamic-name") != null);
  
           server.stop();
        }
     }
  
     @Test
     @Timeout(20)
     public void testFederationSendsReceiveFromQueuePolicyToRemoteWhenSendToIsConfigured() throws Exception {
        final MessageAnnotationsMatcher maMatcher = new MessageAnnotationsMatcher(true);
        maMatcher.withEntry(OPERATION_TYPE.toString(), Matchers.is(ADD_QUEUE_POLICY));
        final Map<String, Object> policyMap = new LinkedHashMap<>();
  
        final List<String> includes = new ArrayList<>();
        includes.add("a");
        includes.add("b");
        includes.add("c");
        includes.add("d");
        final List<String> excludes = new ArrayList<>();
        excludes.add("e");
        excludes.add("f");
        excludes.add("g");
        excludes.add("h");
  
        policyMap.put(POLICY_NAME, "test-policy");
        policyMap.put(QUEUE_INCLUDE_FEDERATED, true);
        policyMap.put(QUEUE_PRIORITY_ADJUSTMENT, 42);
        policyMap.put(QUEUE_INCLUDES, includes);
        policyMap.put(QUEUE_EXCLUDES, excludes);
  
        final EncodedAmqpValueMatcher bodyMatcher = new EncodedAmqpValueMatcher(policyMap);
        final TransferPayloadCompositeMatcher payloadMatcher = new TransferPayloadCompositeMatcher();
        payloadMatcher.setMessageAnnotationsMatcher(maMatcher);
        payloadMatcher.addMessageContentMatcher(bodyMatcher);
  
        try (ProtonTestServer peer = new ProtonTestServer()) {
           peer.expectSASLAnonymousConnect();
           peer.expectOpen().respond();
           peer.expectBegin().respond();
           peer.expectAttach().ofSender()
                              .withTarget().withDynamic(true).and()
                              .withDesiredCapability(FEDERATION_CONTROL_LINK.toString())
                              .respondInKind().withTarget().withAddress("test-dynamic");
           peer.remoteFlow().withLinkCredit(10).queue();
           peer.expectAttach().ofSender()
                              .withTarget().withDynamic(true).and()
                              .withDesiredCapability(FEDERATION_EVENT_LINK.toString())
                              .respondInKind().withTarget().withAddress("test-dynamic-events");
           peer.remoteFlow().withLinkCredit(10).queue();
           peer.expectTransfer().withPayload(payloadMatcher);
           peer.start();
  
           final URI remoteURI = peer.getServerURI();
           logger.info("Connect test started, peer listening on: {}", remoteURI);
  
           final AMQPFederationQueuePolicyElement sendToQueue = new AMQPFederationQueuePolicyElement();
           sendToQueue.setName("test-policy");
           sendToQueue.setIncludeFederated(true);
           sendToQueue.setPriorityAdjustment(42);
           sendToQueue.addToIncludes("a", "b");
           sendToQueue.addToIncludes("c", "d");
           sendToQueue.addToExcludes("e", "f");
           sendToQueue.addToExcludes("g", "h");
  
           final AMQPFederatedBrokerConnectionElement element = new AMQPFederatedBrokerConnectionElement();
           element.setName(getTestName());
           element.addRemoteQueuePolicy(sendToQueue);
  
           final AMQPBrokerConnectConfiguration amqpConnection =
              new AMQPBrokerConnectConfiguration(getTestName(), "tcp://" + remoteURI.getHost() + ":" + remoteURI.getPort());
           amqpConnection.setReconnectAttempts(0);// No reconnects
           amqpConnection.addElement(element);
  
           server.getConfiguration().addAMQPConnection(amqpConnection);
           server.start();
  
           peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
           peer.close();
        }
     }
  
     @Test
     @Timeout(20)
     public void testFederationSendsReceiveFromQueuePolicyToRemoteWhenSendToIsConfiguredAndEventSenderRejected() throws Exception {
        final MessageAnnotationsMatcher maMatcher = new MessageAnnotationsMatcher(true);
        maMatcher.withEntry(OPERATION_TYPE.toString(), Matchers.is(ADD_QUEUE_POLICY));
        final Map<String, Object> policyMap = new LinkedHashMap<>();
  
        final List<String> includes = new ArrayList<>();
        includes.add("a");
        includes.add("b");
        includes.add("c");
        includes.add("d");
        final List<String> excludes = new ArrayList<>();
        excludes.add("e");
        excludes.add("f");
        excludes.add("g");
        excludes.add("h");
  
        policyMap.put(POLICY_NAME, "test-policy");
        policyMap.put(QUEUE_INCLUDE_FEDERATED, true);
        policyMap.put(QUEUE_PRIORITY_ADJUSTMENT, 42);
        policyMap.put(QUEUE_INCLUDES, includes);
        policyMap.put(QUEUE_EXCLUDES, excludes);
  
        final EncodedAmqpValueMatcher bodyMatcher = new EncodedAmqpValueMatcher(policyMap);
        final TransferPayloadCompositeMatcher payloadMatcher = new TransferPayloadCompositeMatcher();
        payloadMatcher.setMessageAnnotationsMatcher(maMatcher);
        payloadMatcher.addMessageContentMatcher(bodyMatcher);
  
        try (ProtonTestServer peer = new ProtonTestServer()) {
           peer.expectSASLAnonymousConnect();
           peer.expectOpen().respond();
           peer.expectBegin().respond();
           peer.expectAttach().ofSender()
                              .withTarget().withDynamic(true).and()
                              .withDesiredCapability(FEDERATION_CONTROL_LINK.toString())
                              .respondInKind().withTarget().withAddress("test-dynamic");
           peer.expectAttach().ofSender()
                              .withTarget().withDynamic(true).and()
                              .withDesiredCapability(FEDERATION_EVENT_LINK.toString())
                              .respond()
                              .withNullSource();
           peer.expectDetach().respond();
           peer.remoteFlow().withHandle(0).withLinkCredit(10).queue(); // Ensure order of events
           peer.expectTransfer().withPayload(payloadMatcher);
           peer.start();
  
           final URI remoteURI = peer.getServerURI();
           logger.info("Connect test started, peer listening on: {}", remoteURI);
  
           final AMQPFederationQueuePolicyElement sendToQueue = new AMQPFederationQueuePolicyElement();
           sendToQueue.setName("test-policy");
           sendToQueue.setIncludeFederated(true);
           sendToQueue.setPriorityAdjustment(42);
           sendToQueue.addToIncludes("a", "b");
           sendToQueue.addToIncludes("c", "d");
           sendToQueue.addToExcludes("e", "f");
           sendToQueue.addToExcludes("g", "h");
  
           final AMQPFederatedBrokerConnectionElement element = new AMQPFederatedBrokerConnectionElement();
           element.setName(getTestName());
           element.addRemoteQueuePolicy(sendToQueue);
  
           final AMQPBrokerConnectConfiguration amqpConnection =
              new AMQPBrokerConnectConfiguration(getTestName(), "tcp://" + remoteURI.getHost() + ":" + remoteURI.getPort());
           amqpConnection.setReconnectAttempts(0);// No reconnects
           amqpConnection.addElement(element);
  
           server.getConfiguration().addAMQPConnection(amqpConnection);
           server.start();
  
           peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
           peer.close();
        }
     }
  
     @Test
     @Timeout(20)
     public void testFederationSendsReceiveFromAddressPolicyToRemoteWhenSendToIsConfigured() throws Exception {
        final MessageAnnotationsMatcher maMatcher = new MessageAnnotationsMatcher(true);
        maMatcher.withEntry(OPERATION_TYPE.toString(), Matchers.is(ADD_ADDRESS_POLICY));
        final Map<String, Object> policyMap = new LinkedHashMap<>();
  
        final List<String> includes = new ArrayList<>();
        includes.add("include");
        final List<String> excludes = new ArrayList<>();
        excludes.add("exclude");
  
        policyMap.put(POLICY_NAME, "test-policy");
        policyMap.put(ADDRESS_AUTO_DELETE, true);
        policyMap.put(ADDRESS_AUTO_DELETE_DELAY, 42L);
        policyMap.put(ADDRESS_AUTO_DELETE_MSG_COUNT, 314L);
        policyMap.put(ADDRESS_MAX_HOPS, 5);
        policyMap.put(ADDRESS_ENABLE_DIVERT_BINDINGS, false);
        policyMap.put(ADDRESS_INCLUDES, includes);
        policyMap.put(ADDRESS_EXCLUDES, excludes);
  
        final EncodedAmqpValueMatcher bodyMatcher = new EncodedAmqpValueMatcher(policyMap);
        final TransferPayloadCompositeMatcher payloadMatcher = new TransferPayloadCompositeMatcher();
        payloadMatcher.setMessageAnnotationsMatcher(maMatcher);
        payloadMatcher.addMessageContentMatcher(bodyMatcher);
  
        try (ProtonTestServer peer = new ProtonTestServer()) {
           peer.expectSASLAnonymousConnect();
           peer.expectOpen().respond();
           peer.expectBegin().respond();
           peer.expectAttach().ofSender()
                              .withTarget().withDynamic(true).and()
                              .withDesiredCapability(FEDERATION_CONTROL_LINK.toString())
                              .respondInKind().withTarget().withAddress("test-dynamic");
           peer.remoteFlow().withLinkCredit(10).queue();
           peer.expectAttach().ofSender()
                              .withTarget().withDynamic(true).and()
                              .withDesiredCapability(FEDERATION_EVENT_LINK.toString())
                              .respondInKind().withTarget().withAddress("test-dynamic-events");
           peer.remoteFlow().withLinkCredit(10).queue();
           peer.expectTransfer().withPayload(payloadMatcher);
           peer.start();
  
           final URI remoteURI = peer.getServerURI();
           logger.info("Connect test started, peer listening on: {}", remoteURI);
  
           final AMQPFederationAddressPolicyElement sendToAddress = new AMQPFederationAddressPolicyElement();
           sendToAddress.setName("test-policy");
           sendToAddress.setAutoDelete(true);
           sendToAddress.setAutoDeleteDelay(42L);
           sendToAddress.setAutoDeleteMessageCount(314L);
           sendToAddress.setMaxHops(5);
           sendToAddress.setEnableDivertBindings(false);
           sendToAddress.addToIncludes("include");
           sendToAddress.addToExcludes("exclude");
  
           final AMQPFederatedBrokerConnectionElement element = new AMQPFederatedBrokerConnectionElement();
           element.setName(getTestName());
           element.addRemoteAddressPolicy(sendToAddress);
  
           final AMQPBrokerConnectConfiguration amqpConnection =
              new AMQPBrokerConnectConfiguration(getTestName(), "tcp://" + remoteURI.getHost() + ":" + remoteURI.getPort());
           amqpConnection.setReconnectAttempts(0);// No reconnects
           amqpConnection.addElement(element);
  
           server.getConfiguration().addAMQPConnection(amqpConnection);
           server.start();
  
           peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
           peer.close();
        }
     }
  
     @Test
     @Timeout(20)
     public void testFederationSendsReceiveFromAddressPolicyToRemoteWhenSendToIsConfiguredAndEventSenderRejected() throws Exception {
        final MessageAnnotationsMatcher maMatcher = new MessageAnnotationsMatcher(true);
        maMatcher.withEntry(OPERATION_TYPE.toString(), Matchers.is(ADD_ADDRESS_POLICY));
        final Map<String, Object> policyMap = new LinkedHashMap<>();
  
        final List<String> includes = new ArrayList<>();
        includes.add("include");
        final List<String> excludes = new ArrayList<>();
        excludes.add("exclude");
  
        policyMap.put(POLICY_NAME, "test-policy");
        policyMap.put(ADDRESS_AUTO_DELETE, true);
        policyMap.put(ADDRESS_AUTO_DELETE_DELAY, 42L);
        policyMap.put(ADDRESS_AUTO_DELETE_MSG_COUNT, 314L);
        policyMap.put(ADDRESS_MAX_HOPS, 5);
        policyMap.put(ADDRESS_ENABLE_DIVERT_BINDINGS, false);
        policyMap.put(ADDRESS_INCLUDES, includes);
        policyMap.put(ADDRESS_EXCLUDES, excludes);
  
        final EncodedAmqpValueMatcher bodyMatcher = new EncodedAmqpValueMatcher(policyMap);
        final TransferPayloadCompositeMatcher payloadMatcher = new TransferPayloadCompositeMatcher();
        payloadMatcher.setMessageAnnotationsMatcher(maMatcher);
        payloadMatcher.addMessageContentMatcher(bodyMatcher);
  
        try (ProtonTestServer peer = new ProtonTestServer()) {
           peer.expectSASLAnonymousConnect();
           peer.expectOpen().respond();
           peer.expectBegin().respond();
           peer.expectAttach().ofSender()
                              .withTarget().withDynamic(true).and()
                              .withDesiredCapability(FEDERATION_CONTROL_LINK.toString())
                              .respondInKind().withTarget().withAddress("test-dynamic");
           peer.expectAttach().ofSender()
                              .withTarget().withDynamic(true).and()
                              .withDesiredCapability(FEDERATION_EVENT_LINK.toString())
                              .respond()
                              .withNullTarget();
           peer.expectDetach().respond();
           peer.remoteFlow().withHandle(0).withLinkCredit(10).queue(); // Ensure order of events
           peer.expectTransfer().withPayload(payloadMatcher);
           peer.start();
  
           final URI remoteURI = peer.getServerURI();
           logger.info("Connect test started, peer listening on: {}", remoteURI);
  
           final AMQPFederationAddressPolicyElement sendToAddress = new AMQPFederationAddressPolicyElement();
           sendToAddress.setName("test-policy");
           sendToAddress.setAutoDelete(true);
           sendToAddress.setAutoDeleteDelay(42L);
           sendToAddress.setAutoDeleteMessageCount(314L);
           sendToAddress.setMaxHops(5);
           sendToAddress.setEnableDivertBindings(false);
           sendToAddress.addToIncludes("include");
           sendToAddress.addToExcludes("exclude");
  
           final AMQPFederatedBrokerConnectionElement element = new AMQPFederatedBrokerConnectionElement();
           element.setName(getTestName());
           element.addRemoteAddressPolicy(sendToAddress);
  
           final AMQPBrokerConnectConfiguration amqpConnection =
              new AMQPBrokerConnectConfiguration(getTestName(), "tcp://" + remoteURI.getHost() + ":" + remoteURI.getPort());
           amqpConnection.setReconnectAttempts(0);// No reconnects
           amqpConnection.addElement(element);
  
           server.getConfiguration().addAMQPConnection(amqpConnection);
           server.start();
  
           peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
           peer.close();
        }
     }
  
     @Test
     @Timeout(20)
     public void testConnectToBrokerFromRemoteAsFederatedSourceAndCreateControlLink() throws Exception {
        server.start();
  
        try (ProtonTestClient peer = new ProtonTestClient()) {
           scriptFederationConnectToRemote(peer, getTestName());
           peer.connect("localhost", AMQP_PORT);
  
           peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
           peer.expectClose();
           peer.remoteClose().now();
  
           peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
           peer.close();
  
           server.stop();
  
           logger.info("Test stopped");
        }
     }
  
     @Test
     @Timeout(20)
     public void testConnectToBrokerFromRemoteAsFederatedSourceAndCreateEventsSenderLink() throws Exception {
        server.start();
  
        try (ProtonTestClient peer = new ProtonTestClient()) {
           scriptFederationConnectToRemote(peer, getTestName(), false, null, null, true, false);
           peer.connect("localhost", AMQP_PORT);
  
           peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
           peer.expectClose();
           peer.remoteClose().now();
  
           peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
           peer.close();
  
           server.stop();
  
           logger.info("Test stopped");
        }
     }
  
     @Test
     @Timeout(20)
     public void testConnectToBrokerFromRemoteAsFederatedSourceAndCreateEventsReceiverLink() throws Exception {
        server.start();
  
        try (ProtonTestClient peer = new ProtonTestClient()) {
           scriptFederationConnectToRemote(peer, getTestName(), false, null, null, false, true);
           peer.connect("localhost", AMQP_PORT);
  
           peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
           peer.expectClose();
           peer.remoteClose().now();
  
           peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
           peer.close();
  
           server.stop();
  
           logger.info("Test stopped");
        }
     }
  
     @Test
     @Timeout(20)
     public void testConnectToBrokerFromRemoteAsFederatedSourceAndCreateEventsLinks() throws Exception {
        server.start();
  
        try (ProtonTestClient peer = new ProtonTestClient()) {
           scriptFederationConnectToRemote(peer, getTestName(), false, null, null, true, true);
           peer.connect("localhost", AMQP_PORT);
  
           peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
           peer.expectClose();
           peer.remoteClose().now();
  
           peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
           peer.close();
  
           server.stop();
  
           logger.info("Test stopped");
        }
     }
  
     @Test
     @Timeout(20)
     public void testControlLinkPassesConnectAttemptWhenUserHasPrivledges() throws Exception {
        enableSecurity(server, FEDERATION_BASE_VALIDATION_ADDRESS);
        server.start();
  
        try (ProtonTestClient peer = new ProtonTestClient()) {
           scriptFederationConnectToRemote(peer, getTestName(), fullUser, fullPass);
           peer.connect("localhost", AMQP_PORT);
  
           peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
           peer.expectClose();
           peer.remoteClose().now();
  
           peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
           peer.close();
  
           server.stop();
  
           logger.info("Test stopped");
        }
     }
  
     @Test
     @Timeout(20)
     public void testControlAndEventsLinksPassesConnectAttemptWhenUserHasPrivledges() throws Exception {
        enableSecurity(server, FEDERATION_BASE_VALIDATION_ADDRESS + ".#");
        server.start();
  
        try (ProtonTestClient peer = new ProtonTestClient()) {
           scriptFederationConnectToRemote(peer, getTestName(), true, fullUser, fullPass, true, true);
           peer.connect("localhost", AMQP_PORT);
  
           peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
           peer.expectClose();
           peer.remoteClose().now();
  
           peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
           peer.close();
  
           server.stop();
  
           logger.info("Test stopped");
        }
     }
  
     @Test
     @Timeout(20)
     public void testControlLinkRefusesConnectAttemptWhenUseDoesNotHavePrivledgesForControlAddress() throws Exception {
        enableSecurity(server, FEDERATION_BASE_VALIDATION_ADDRESS);
        server.start();
  
        try (ProtonTestClient peer = new ProtonTestClient()) {
           scriptFederationConnectToRemoteNotAuthorizedForControlAddress(peer, getTestName(), guestUser, guestPass);
           peer.connect("localhost", AMQP_PORT);
  
           peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
           peer.close();
  
           server.stop();
  
           logger.info("Test stopped");
        }
     }
  
     @Test
     @Timeout(20)
     public void testRemoteConnectionCannotAttachEventReceiverLinkWithoutControlLink() throws Exception {
        server.start();
  
        try (ProtonTestClient peer = new ProtonTestClient()) {
           peer.queueClientSaslAnonymousConnect();
           peer.remoteOpen().queue();
           peer.expectOpen();
           peer.remoteBegin().queue();
           peer.expectBegin();
           peer.connect("localhost", AMQP_PORT);
  
           peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
           // Broker should reject the attach since there's no control link
           peer.expectAttach().ofSender().withName("federation-event-receiver")
                                         .withNullSource()
                                         .withTarget();
           peer.expectDetach().withError(AmqpError.ILLEGAL_STATE.toString()).respond();
  
           // Attempt to create a federation event receiver link without existing control link
           peer.remoteAttach().ofReceiver()
                              .withDesiredCapabilities(FEDERATION_EVENT_LINK.toString())
                              .withName("federation-event-receiver")
                              .withSenderSettleModeSettled()
                              .withReceivervSettlesFirst()
                              .withSource().withDurabilityOfNone()
                                           .withExpiryPolicyOnLinkDetach()
                                           .withCapabilities("temporary-topic")
                                           .withDynamic(true)
                                           .and()
                              .withTarget().and()
                              .now();
  
           peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
  
           peer.expectClose();
           peer.remoteClose().now();
  
           peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
           peer.close();
  
           server.stop();
        }
     }
  
     @Test
     @Timeout(20)
     public void testRemoteConnectionCannotAttachEventSenderLinkWithoutControlLink() throws Exception {
        server.start();
  
        try (ProtonTestClient peer = new ProtonTestClient()) {
           peer.queueClientSaslAnonymousConnect();
           peer.remoteOpen().queue();
           peer.expectOpen();
           peer.remoteBegin().queue();
           peer.expectBegin();
           peer.connect("localhost", AMQP_PORT);
  
           peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
           // Broker should reject the attach since there's no control link
           peer.expectAttach().ofReceiver().withName("federation-event-sender")
                                           .withSource().also()
                                           .withNullTarget();
           peer.expectDetach().withError(AmqpError.ILLEGAL_STATE.toString()).respond();
  
           // Attempt to create a federation event receiver link without existing control link
           peer.remoteAttach().ofSender()
                              .withDesiredCapabilities(FEDERATION_EVENT_LINK.toString())
                              .withName("federation-event-sender")
                              .withSenderSettleModeSettled()
                              .withReceivervSettlesFirst()
                              .withTarget().withDurabilityOfNone()
                                           .withExpiryPolicyOnLinkDetach()
                                           .withCapabilities("temporary-topic")
                                           .withDynamic(true)
                                           .and()
                              .withSource().and()
                              .now();
  
           peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
  
           peer.expectClose();
           peer.remoteClose().now();
  
           peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
           peer.close();
  
           server.stop();
        }
     }
  
     @Test
     @Timeout(20)
     public void testControlLinkSenderQueueCreatedWithMaxConsumersOfOne() throws Exception {
        final String controlLinkAddress = "test-control-address";
        final String federationControlSenderAddress = FEDERATION_BASE_VALIDATION_ADDRESS +
                                                      "." + FEDERATION_CONTROL_LINK_PREFIX +
                                                      "." + controlLinkAddress;
  
        try (ProtonTestServer peer = new ProtonTestServer()) {
           peer.expectSASLAnonymousConnect("PLAIN", "ANONYMOUS");
           peer.expectOpen().respond();
           peer.expectBegin().respond();
           peer.expectAttach().ofSender()
                              .withDesiredCapability(FEDERATION_CONTROL_LINK.toString())
                              .withName(allOf(containsString("federation-"), containsString("myFederation")))
                              .withTarget().withDynamic(true)
                                           .withCapabilities("temporary-topic")
                              .and()
                              .respond()
                              .withTarget().withAddress(controlLinkAddress)
                              .and()
                              .withOfferedCapabilities(FEDERATION_CONTROL_LINK.toString());
           peer.start();
  
           final URI remoteURI = peer.getServerURI();
           logger.info("Connect test started, peer listening on: {}", remoteURI);
  
           final AMQPBrokerConnectConfiguration amqpConnection = new AMQPBrokerConnectConfiguration(
              getTestName(), "tcp://" + remoteURI.getHost() + ":" + remoteURI.getPort());
           amqpConnection.setReconnectAttempts(0);// No reconnects
           final AMQPFederatedBrokerConnectionElement federation = new AMQPFederatedBrokerConnectionElement("myFederation");
           amqpConnection.addElement(federation);
  
           server.getConfiguration().addAMQPConnection(amqpConnection);
           server.start();
  
           peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
  
           Wait.assertTrue(() -> server.locateQueue(federationControlSenderAddress) != null);
  
           final Queue result = server.locateQueue(SimpleString.of(federationControlSenderAddress));
  
           assertNotNull(result);
           assertTrue(result.isTemporary());
           assertTrue(result.isInternalQueue());
           assertEquals(1, result.getMaxConsumers());
  
           // Try and bind to the control address which should be rejected as the queue
           // was created with max consumers of one.
           peer.expectAttach().ofSender()
                              .withName("test-control-link-suspect")
                              .withNullSource();
           peer.expectDetach().withClosed(true)
                              .withError(AmqpError.INTERNAL_ERROR.toString());
           peer.remoteAttach().ofReceiver()
                              .withName("test-control-link-suspect")
                              .withSenderSettleModeUnsettled()
                              .withReceivervSettlesFirst()
                              .withTarget().also()
                              .withSource().withAddress(federationControlSenderAddress)
                                           .also()
                                           .now();
  
           peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
        }
     }
  
     @Test
     @Timeout(20)
     public void testEventSenderLinkFromTargetUsesNamespacedDynamicQueue() throws Exception {
        final String federationControlLinkName = "federation-test";
  
        server.start();
  
        try (ProtonTestClient peer = new ProtonTestClient()) {
           peer.queueClientSaslAnonymousConnect();
           peer.remoteOpen().queue();
           peer.expectOpen();
           peer.remoteBegin().queue();
           peer.expectBegin();
           peer.remoteAttach().ofSender()
                              .withName(federationControlLinkName)
                              .withDesiredCapabilities(FEDERATION_CONTROL_LINK.toString())
                              .withSenderSettleModeUnsettled()
                              .withReceivervSettlesFirst()
                              .withSource().also()
                              .withTarget().withDynamic(true)
                                           .withDurabilityOfNone()
                                           .withExpiryPolicyOnLinkDetach()
                                           .withLifetimePolicyOfDeleteOnClose()
                                           .withCapabilities("temporary-topic")
                                           .also()
                              .queue();
           peer.expectAttach().ofReceiver()
                              .withName(federationControlLinkName)
                              .withTarget()
                                 .withAddress(notNullValue())
                              .also()
                              .withOfferedCapability(FEDERATION_CONTROL_LINK.toString());
           peer.expectFlow();
  
           final String federationEventsSenderLinkName = "events-receiver-test";
  
           peer.remoteAttach().ofReceiver()
                              .withName(federationEventsSenderLinkName)
                              .withDesiredCapabilities(FEDERATION_EVENT_LINK.toString())
                              .withSenderSettleModeUnsettled()
                              .withReceivervSettlesFirst()
                              .withTarget().also()
                              .withSource().withDynamic(true)
                                           .withDurabilityOfNone()
                                           .withExpiryPolicyOnLinkDetach()
                                           .withLifetimePolicyOfDeleteOnClose()
                                           .withCapabilities("temporary-topic")
                                           .also()
                                           .queue();
           peer.remoteFlow().withLinkCredit(10).queue();
           peer.expectAttach().ofSender()
                              .withName(federationEventsSenderLinkName)
                              .withSource()
                                 .withAddress(notNullValue())
                              .also()
                              .withOfferedCapability(FEDERATION_EVENT_LINK.toString());
  
           peer.connect("localhost", AMQP_PORT);
           peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
  
           // The events receiver from the remote should trigger a temporary queue to be created on
           // the server to allow sends of events beyond currently available credit.
           Wait.assertTrue(() -> server.locateQueue(FEDERATION_BASE_VALIDATION_ADDRESS +
                                                    "." + FEDERATION_EVENTS_LINK_PREFIX +
                                                    "." + federationEventsSenderLinkName) != null);
  
           server.stop();
        }
     }
  
     @Test
     @Timeout(20)
     public void testEventsLinkAtTargetIsCreatedWithMaxConsumersOfOne() throws Exception {
        final String federationControlLinkName = "federation-test";
  
        server.start();
  
        try (ProtonTestClient peer = new ProtonTestClient()) {
           peer.queueClientSaslAnonymousConnect();
           peer.remoteOpen().queue();
           peer.expectOpen();
           peer.remoteBegin().queue();
           peer.expectBegin();
           peer.remoteAttach().ofSender()
                              .withName(federationControlLinkName)
                              .withDesiredCapabilities(FEDERATION_CONTROL_LINK.toString())
                              .withSenderSettleModeUnsettled()
                              .withReceivervSettlesFirst()
                              .withSource().also()
                              .withTarget().withDynamic(true)
                                           .withDurabilityOfNone()
                                           .withExpiryPolicyOnLinkDetach()
                                           .withLifetimePolicyOfDeleteOnClose()
                                           .withCapabilities("temporary-topic")
                                           .also()
                              .queue();
           peer.expectAttach().ofReceiver()
                              .withName(federationControlLinkName)
                              .withTarget()
                                 .withAddress(notNullValue())
                              .also()
                              .withOfferedCapability(FEDERATION_CONTROL_LINK.toString());
           peer.expectFlow();
  
           final String federationEventsSenderLinkName = "events-receiver-test";
           final String federationEventsSenderAddress = FEDERATION_BASE_VALIDATION_ADDRESS +
                                                        "." + FEDERATION_EVENTS_LINK_PREFIX +
                                                        "." + federationEventsSenderLinkName;
  
           peer.remoteAttach().ofReceiver()
                              .withName(federationEventsSenderLinkName)
                              .withDesiredCapabilities(FEDERATION_EVENT_LINK.toString())
                              .withSenderSettleModeUnsettled()
                              .withReceivervSettlesFirst()
                              .withTarget().also()
                              .withSource().withDynamic(true)
                                           .withDurabilityOfNone()
                                           .withExpiryPolicyOnLinkDetach()
                                           .withLifetimePolicyOfDeleteOnClose()
                                           .withCapabilities("temporary-topic")
                                           .also()
                                           .queue();
           peer.remoteFlow().withLinkCredit(10).queue();
           peer.expectAttach().ofSender()
                              .withName(federationEventsSenderLinkName)
                              .withSource()
                                 .withAddress(notNullValue())
                              .also()
                              .withOfferedCapability(FEDERATION_EVENT_LINK.toString());
  
           peer.connect("localhost", AMQP_PORT);
           peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
  
           // The events receiver from the remote should trigger a temporary queue to be created on
           // the server to allow sends of events beyond currently available credit.
           Wait.assertTrue(() -> server.locateQueue(federationEventsSenderAddress) != null);
  
           final Queue result = server.locateQueue(SimpleString.of(federationEventsSenderAddress));
  
           assertNotNull(result);
           assertTrue(result.isTemporary());
           assertTrue(result.isInternalQueue());
           assertEquals(1, result.getMaxConsumers());
  
           // Try and bind to the events address which should be rejected as the queue
           // was created with max consumers of one.
           peer.expectAttach().ofSender()
                              .withName("test-events-link-suspect")
                              .withNullSource();
           peer.expectDetach().withClosed(true)
                              .withError(AmqpError.INTERNAL_ERROR.toString());
           peer.remoteAttach().ofReceiver()
                              .withName("test-events-link-suspect")
                              .withSenderSettleModeUnsettled()
                              .withReceivervSettlesFirst()
                              .withTarget().also()
                              .withSource().withAddress(federationEventsSenderAddress)
                                           .also()
                                           .now();
  
           peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
  
           server.stop();
        }
     }
  
     @Test
     @Timeout(30)
     public void testFederationDemandAddedAndImmediateBrokerShutdownOverlaps() throws Exception {
        // Testing for a race on broker shutdown if demand was added at the same time and the
        // broker is creating an outbound consumer to match that demand.
        for (int i = 0; i < 2; ++i) {
           doTestFederationDemandAddedAndImmediateBrokerShutdown();
        }
     }
  
     private void doTestFederationDemandAddedAndImmediateBrokerShutdown() throws Exception {
        if (server == null || !server.isStarted()) {
           server = createServer();
        }
  
        try (ProtonTestServer peer = new ProtonTestServer()) {
           peer.expectSASLAnonymousConnect();
           peer.expectOpen().respond();
           peer.expectBegin().respond();
           peer.expectAttach().ofSender()
                              .withDesiredCapability(FEDERATION_CONTROL_LINK.toString())
                              .respond()
                              .withOfferedCapabilities(FEDERATION_CONTROL_LINK.toString());
           peer.expectAttach().ofReceiver()
                              .withSenderSettleModeSettled()
                              .withDesiredCapability(FEDERATION_EVENT_LINK.toString())
                              .respondInKind();
           peer.expectFlow().withLinkCredit(10);
           peer.start();
  
           final URI remoteURI = peer.getServerURI();
           logger.info("Connect test started, peer listening on: {}", remoteURI);
  
           final AMQPFederationQueuePolicyElement receiveFromQueue = new AMQPFederationQueuePolicyElement();
           receiveFromQueue.setName("queue-policy");
           receiveFromQueue.addToIncludes("test", "test");
  
           final AMQPFederatedBrokerConnectionElement element = new AMQPFederatedBrokerConnectionElement();
           element.setName(getTestName());
           element.addLocalQueuePolicy(receiveFromQueue);
  
           final AMQPBrokerConnectConfiguration amqpConnection =
              new AMQPBrokerConnectConfiguration(getTestName(), "tcp://" + remoteURI.getHost() + ":" + remoteURI.getPort());
           amqpConnection.setReconnectAttempts(0);// No reconnects
           amqpConnection.addElement(element);
  
           server.getConfiguration().addAMQPConnection(amqpConnection);
           server.start();
           server.registerBrokerPlugin(new ActiveMQServerAMQPFederationPlugin() {
  
              @Override
              public void beforeCreateFederationConsumer(final FederationConsumerInfo consumerInfo) throws ActiveMQException {
                 logger.debug("Delaying attach of outgoing federation receiver");
                 ForkJoinPool.commonPool().execute(() -> {
                    try {
                       server.stop();
                    } catch (Exception e) {
                    }
                 });
                 // Allow a bit of time for the server stop to get started before allowing
                 // the remote federation consumer to begin being built.
                 try {
                    Thread.sleep(new Random(System.currentTimeMillis()).nextInt(8));
                 } catch (InterruptedException e) {
                 }
              }
           });
  
           peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
           peer.expectAttach().optional();
           peer.expectFlow().optional();
           peer.expectDetach().optional();
           peer.expectClose().optional();
           peer.expectConnectionToDrop();
  
           final ConnectionFactory factory = CFUtil.createConnectionFactory("AMQP", "tcp://localhost:" + AMQP_PORT);
  
           try (Connection connection = factory.createConnection()) {
              final Session session = connection.createSession(Session.AUTO_ACKNOWLEDGE);
  
              connection.start();
  
              try {
                 session.createConsumer(session.createQueue("test"));
              } catch (JMSException ex) {
                 // Ignored as we are asynchronously shutting down the server, this could happen.
              }
  
              peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
              peer.close();
           }
        }
     }
  
     @Test
     @Timeout(20)
     public void testBrokerConnectsCarriesInfoInConnectionProperties() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
           peer.start();
  
           final URI remoteURI = peer.getServerURI();
           logger.info("Connect test started, peer listening on: {}", remoteURI);
  
           // No user or pass given, it will have to select ANONYMOUS even though PLAIN also offered
           AMQPBrokerConnectConfiguration amqpConnection =
                 new AMQPBrokerConnectConfiguration(getTestName(), "tcp://" + remoteURI.getHost() + ":" + remoteURI.getPort());
           amqpConnection.setReconnectAttempts(0);// No reconnects
           amqpConnection.setAutostart(false);
           server.getConfiguration().addAMQPConnection(amqpConnection);
           server.start();
  
           final Map<String, Object> brokerConnectionInfo = new HashMap<>();
  
           brokerConnectionInfo.put(CONNECTION_NAME, getTestName());
           brokerConnectionInfo.put(NODE_ID, server.getNodeID().toString());
  
           peer.expectSASLAnonymousConnect("PLAIN", "ANONYMOUS");
           peer.expectOpen().withProperty(BROKER_CONNECTION_INFO.toString(), brokerConnectionInfo).respond();
           peer.expectBegin().respond();
  
           server.getBrokerConnections().forEach(conn -> {
              try {
                 conn.start();
              } catch (Exception e) {
                 fail("Should not have thrown an error on start: " + e);
              }
           });
  
           peer.waitForScriptToComplete(5, TimeUnit.MINUTES);
        }
     }
  
     @Test
     @Timeout(20)
     public void testBrokerConnectionDoesNotSendOfferedCapabilities() throws Exception {
        // Checks that current behavior is to not send, this will break if the defaults are
        // changed such that the connection starts sending offered capabilities by default.
        try (ProtonTestServer peer = new ProtonTestServer()) {
           peer.expectSASLAnonymousConnect("PLAIN", "ANONYMOUS");
           peer.expectOpen().withOfferedCapabilities(nullValue()).respond();
           peer.expectBegin().respond();
           peer.start();
  
           final URI remoteURI = peer.getServerURI();
           logger.info("Connect test started, peer listening on: {}", remoteURI);
  
           // No user or pass given, it will have to select ANONYMOUS even though PLAIN also offered
           AMQPBrokerConnectConfiguration amqpConnection =
                 new AMQPBrokerConnectConfiguration(getTestName(), "tcp://" + remoteURI.getHost() + ":" + remoteURI.getPort());
           amqpConnection.setReconnectAttempts(0);// No reconnects
           server.getConfiguration().addAMQPConnection(amqpConnection);
           server.start();
  
           peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
        }
     }
  
     // Use these methods to script the initial handshake that a broker that is establishing
     // a federation connection with a remote broker instance would perform.
  
     private void scriptFederationConnectToRemote(ProtonTestClient peer, String federationName) {
        scriptFederationConnectToRemote(peer, federationName, false, null, null);
     }
  
     private void scriptFederationConnectToRemote(ProtonTestClient peer, String federationName, String username, String password) {
        scriptFederationConnectToRemote(peer, federationName, true, username, password);
     }
  
     private void scriptFederationConnectToRemote(ProtonTestClient peer, String federationName, boolean auth, String username, String password) {
        scriptFederationConnectToRemote(peer, federationName, auth, username, password, false, false);
     }
  
     private void scriptFederationConnectToRemote(ProtonTestClient peer, String federationName, boolean auth, String username, String password, boolean eventsSender, boolean eventsReceiver) {
        final String federationControlLinkName = "Federation:test:" + UUID.randomUUID().toString();
  
        if (auth) {
           peer.queueClientSaslPlainConnect(username, password);
        } else {
           peer.queueClientSaslAnonymousConnect();
        }
  
        peer.remoteOpen().queue();
        peer.expectOpen();
        peer.remoteBegin().queue();
        peer.expectBegin();
        peer.remoteAttach().ofSender()
                           .withName(federationControlLinkName)
                           .withDesiredCapabilities(FEDERATION_CONTROL_LINK.toString())
                           .withProperty(FEDERATION_VERSION.toString(), FEDERATION_V2)
                           .withSenderSettleModeUnsettled()
                           .withReceivervSettlesFirst()
                           .withSource().also()
                           .withTarget().withDynamic(true)
                                        .withDurabilityOfNone()
                                        .withExpiryPolicyOnLinkDetach()
                                        .withLifetimePolicyOfDeleteOnClose()
                                        .withCapabilities("temporary-topic")
                                        .also()
                           .queue();
        peer.expectAttach().ofReceiver()
                           .withName(federationControlLinkName)
                           .withTarget()
                              .withAddress(notNullValue())
                           .also()
                           .withProperty(FEDERATION_VERSION.toString(), FEDERATION_V2)
                           .withOfferedCapabilities(FEDERATION_CONTROL_LINK.toString());
        peer.expectFlow();
  
        if (eventsSender) {
           final String federationEventsSenderLinkName = "Federation:events-sender:test:" + UUID.randomUUID().toString();
  
           peer.remoteAttach().ofSender()
                              .withName(federationEventsSenderLinkName)
                              .withDesiredCapabilities(FEDERATION_EVENT_LINK.toString())
                              .withSenderSettleModeUnsettled()
                              .withReceivervSettlesFirst()
                              .withSource().also()
                              .withTarget().withDynamic(true)
                                           .withDurabilityOfNone()
                                           .withExpiryPolicyOnLinkDetach()
                                           .withLifetimePolicyOfDeleteOnClose()
                                           .withCapabilities("temporary-topic")
                                           .also()
                                           .queue();
           peer.expectAttach().ofReceiver()
                              .withName(federationEventsSenderLinkName)
                              .withTarget()
                                 .withAddress(notNullValue())
                              .also()
                              .withOfferedCapability(FEDERATION_EVENT_LINK.toString());
           peer.expectFlow();
        }
  
        if (eventsReceiver) {
           final String federationEventsSenderLinkName = "Federation:events-receiver:test:" + UUID.randomUUID().toString();
  
           peer.remoteAttach().ofReceiver()
                              .withName(federationEventsSenderLinkName)
                              .withDesiredCapabilities(FEDERATION_EVENT_LINK.toString())
                              .withSenderSettleModeUnsettled()
                              .withReceivervSettlesFirst()
                              .withTarget().also()
                              .withSource().withDynamic(true)
                                           .withDurabilityOfNone()
                                           .withExpiryPolicyOnLinkDetach()
                                           .withLifetimePolicyOfDeleteOnClose()
                                           .withCapabilities("temporary-topic")
                                           .also()
                                           .queue();
           peer.remoteFlow().withLinkCredit(10).queue();
           peer.expectAttach().ofSender()
                              .withName(federationEventsSenderLinkName)
                              .withSource()
                                 .withAddress(notNullValue())
                              .also()
                              .withOfferedCapability(FEDERATION_EVENT_LINK.toString());
        }
     }
  
     private void scriptFederationConnectToRemoteNotAuthorizedForControlAddress(ProtonTestClient peer, String federationName, String username, String password) {
        final String federationControlLinkName = "Federation:test:" + UUID.randomUUID().toString();
  
        peer.queueClientSaslPlainConnect(username, password);
        peer.remoteOpen().queue();
        peer.expectOpen();
        peer.remoteBegin().queue();
        peer.expectBegin();
        peer.remoteAttach().ofSender()
                           .withName(federationControlLinkName)
                           .withDesiredCapabilities(FEDERATION_CONTROL_LINK.toString())
                           .withSenderSettleModeUnsettled()
                           .withReceivervSettlesFirst()
                           .withSource().also()
                           .withTarget().withDynamic(true)
                                        .withDurabilityOfNone()
                                        .withExpiryPolicyOnLinkDetach()
                                        .withLifetimePolicyOfDeleteOnClose()
                                        .withCapabilities("temporary-topic")
                                        .also()
                           .queue();
        peer.expectAttach().ofReceiver()
                           .withTarget(nullValue());
        peer.expectDetach().withError("amqp:unauthorized-access",
                                      "User does not have permission to attach to the federation control address").respond();
        peer.remoteClose().queue();
        peer.expectClose();
     }
  }
  