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.bridge.AMQPBridgeConstants.ADDRESS_RECEIVER_IDLE_TIMEOUT;
  import static org.hamcrest.CoreMatchers.allOf;
  import static org.hamcrest.CoreMatchers.containsString;
  import static org.junit.jupiter.api.Assertions.assertNotNull;
  
  import java.lang.invoke.MethodHandles;
  import java.net.URI;
  import java.util.Collections;
  import java.util.concurrent.TimeUnit;
  
  import javax.jms.Connection;
  import javax.jms.ConnectionFactory;
  import javax.jms.MessageConsumer;
  import javax.jms.Session;
  
  import org.apache.activemq.artemis.api.core.RoutingType;
  import org.apache.activemq.artemis.api.core.SimpleString;
  import org.apache.activemq.artemis.core.config.amqpBrokerConnectivity.AMQPBridgeAddressPolicyElement;
  import org.apache.activemq.artemis.core.config.amqpBrokerConnectivity.AMQPBridgeBrokerConnectionElement;
  import org.apache.activemq.artemis.core.config.amqpBrokerConnectivity.AMQPBridgeQueuePolicyElement;
  import org.apache.activemq.artemis.core.config.amqpBrokerConnectivity.AMQPBrokerConnectConfiguration;
  import org.apache.activemq.artemis.core.server.ActiveMQServer;
  import org.apache.activemq.artemis.core.server.impl.AddressInfo;
  import org.apache.activemq.artemis.protocol.amqp.broker.ProtonProtocolManagerFactory;
  import org.apache.activemq.artemis.tests.integration.amqp.AmqpClientTestSupport;
  import org.apache.activemq.artemis.tests.util.CFUtil;
  import org.apache.qpid.protonj2.test.driver.ProtonTestServer;
  import org.junit.jupiter.api.Test;
  import org.junit.jupiter.api.Timeout;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  /**
   * Test for reload handling in the broker connection bridge implementation
   */
  class AMQPBridgeConfigurationReloadTest extends AmqpClientTestSupport {
  
     private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
  
     @Override
     protected String getConfiguredProtocols() {
        return "AMQP,CORE";
     }
  
     @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 testBridgeConfigurationWithoutChangesIsIgnoredOnUpdate() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
           peer.expectSASLAnonymousConnect();
           peer.expectOpen().respond();
           peer.expectBegin().respond();
           peer.start();
  
           final URI remoteURI = peer.getServerURI();
           logger.info("Test started, peer listening on: {}", remoteURI);
  
           final AMQPBridgeAddressPolicyElement receiveFromAddress = new AMQPBridgeAddressPolicyElement();
           receiveFromAddress.setName("address-policy");
           receiveFromAddress.addToIncludes(getTestName());
           receiveFromAddress.addToExcludes("test.ignore.#");
           receiveFromAddress.setPriority(1);
  
           final AMQPBridgeBrokerConnectionElement element = new AMQPBridgeBrokerConnectionElement();
           element.setName(getTestName());
           element.addBridgeFromAddressPolicy(receiveFromAddress);
           element.addProperty(ADDRESS_RECEIVER_IDLE_TIMEOUT, 0);
  
           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.addAddressInfo(new AddressInfo(SimpleString.of(getTestName()), RoutingType.MULTICAST));
  
           peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
           peer.expectAttach().ofReceiver()
                              .withName(allOf(containsString(getTestName()),
                                              containsString("address-receiver"),
                                              containsString("amqp-bridge"),
                                              containsString(server.getNodeID().toString())))
                              .respond();
           peer.expectFlow().withLinkCredit(1000);
  
           final ConnectionFactory factory = CFUtil.createConnectionFactory("AMQP", "tcp://localhost:" + AMQP_PORT);
  
           try (Connection connection = factory.createConnection()) {
              final Session session = connection.createSession(Session.AUTO_ACKNOWLEDGE);
              final MessageConsumer consumer = session.createConsumer(session.createTopic(getTestName()));
  
              final ProtonProtocolManagerFactory protocolFactory = (ProtonProtocolManagerFactory)
                 server.getRemotingService().getProtocolFactoryMap().get("AMQP");
              assertNotNull(protocolFactory);
  
              final AMQPBridgeAddressPolicyElement updatedReceiveFromAddress = new AMQPBridgeAddressPolicyElement();
              updatedReceiveFromAddress.setName("address-policy");
              updatedReceiveFromAddress.addToIncludes(getTestName());
              updatedReceiveFromAddress.addToExcludes("test.ignore.#");
              updatedReceiveFromAddress.setPriority(1);
  
              final AMQPBridgeBrokerConnectionElement updatedElement = new AMQPBridgeBrokerConnectionElement();
              updatedElement.setName(getTestName());
              updatedElement.addBridgeFromAddressPolicy(updatedReceiveFromAddress);
  
              amqpConnection.getConnectionElements().clear();
              amqpConnection.addElement(updatedElement); // This should be equivalent to replacing the previous instance.
  
              server.getConfiguration().getAMQPConnection().clear();
              server.getConfiguration().addAMQPConnection(amqpConnection);
  
              protocolFactory.updateProtocolServices(server, Collections.emptyList());
  
              connection.start();
  
              peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
              peer.expectFlow().withLinkCredit(1000).withDrain(true)
                               .respond()
                               .withLinkCredit(0).withDeliveryCount(1000).withDrain(true);
              peer.expectDetach().respond();
  
              consumer.close();
  
              peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
              peer.close();
           }
        }
     }
  
     @Test
     @Timeout(20)
     public void testBridgeConnectsToSecondPeerWhenConfigurationUpdatedWithNewConnection() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
  
           peer.expectSASLAnonymousConnect();
           peer.expectOpen().respond();
           peer.expectBegin().respond();
           peer.start();
  
           final URI remoteURI = peer.getServerURI();
           logger.info("Test started, peer listening on: {}", remoteURI);
  
           final AMQPBridgeAddressPolicyElement receiveFromAddress = new AMQPBridgeAddressPolicyElement();
           receiveFromAddress.setName("address-policy");
           receiveFromAddress.addToIncludes(getTestName());
           receiveFromAddress.addToExcludes("test.ignore.#");
           receiveFromAddress.setPriority(1);
  
           final AMQPBridgeBrokerConnectionElement element = new AMQPBridgeBrokerConnectionElement();
           element.setName(getTestName() + ":1");
           element.addBridgeFromAddressPolicy(receiveFromAddress);
  
           final AMQPBrokerConnectConfiguration amqpConnection =
              new AMQPBrokerConnectConfiguration(getTestName() + ":1", "tcp://" + remoteURI.getHost() + ":" + remoteURI.getPort());
           amqpConnection.setReconnectAttempts(0);// No reconnects
           amqpConnection.addElement(element);
  
           server.getConfiguration().addAMQPConnection(amqpConnection);
           server.start();
           server.addAddressInfo(new AddressInfo(SimpleString.of(getTestName()), RoutingType.MULTICAST));
  
           peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
           peer.expectAttach().ofReceiver()
                              .withName(allOf(containsString(getTestName() + ":1"),
                                              containsString("address-receiver"),
                                              containsString("amqp-bridge"),
                                              containsString(server.getNodeID().toString())))
                              .respond();
           peer.expectFlow().withLinkCredit(1000);
  
           final ConnectionFactory factory = CFUtil.createConnectionFactory("AMQP", "tcp://localhost:" + AMQP_PORT);
  
           try (Connection connection = factory.createConnection()) {
              final Session session = connection.createSession(Session.AUTO_ACKNOWLEDGE);
              session.createConsumer(session.createTopic(getTestName()));
  
              connection.start();
  
              try (ProtonTestServer peer2 = new ProtonTestServer()) {
                 peer2.expectSASLAnonymousConnect();
                 peer2.expectOpen().respond();
                 peer2.expectBegin().respond();
                 peer2.expectAttach().ofReceiver()
                                     .withName(allOf(containsString(getTestName() + ":2"),
                                                     containsString("address-receiver"),
                                                     containsString("amqp-bridge"),
                                                     containsString(server.getNodeID().toString())))
                                     .respond();
                 peer2.expectFlow().withLinkCredit(1000);
                 peer2.start();
  
                 final URI remoteURI2 = peer2.getServerURI();
                 logger.info("Test peer 2 started, peer listening on: {}", remoteURI2);
  
                 final ProtonProtocolManagerFactory protocolFactory = (ProtonProtocolManagerFactory)
                    server.getRemotingService().getProtocolFactoryMap().get("AMQP");
                 assertNotNull(protocolFactory);
  
                 final AMQPBridgeAddressPolicyElement updatedReceiveFromAddress = new AMQPBridgeAddressPolicyElement();
                 updatedReceiveFromAddress.setName("address-policy");
                 updatedReceiveFromAddress.addToIncludes(getTestName());
                 updatedReceiveFromAddress.addToExcludes("test.ignore.#");
                 updatedReceiveFromAddress.setPriority(1);
  
                 final AMQPBridgeBrokerConnectionElement updatedElement = new AMQPBridgeBrokerConnectionElement();
                 updatedElement.setName(getTestName() + ":2");
                 updatedElement.addBridgeFromAddressPolicy(updatedReceiveFromAddress);
  
                 final AMQPBrokerConnectConfiguration updatedAmqpConnection =
                    new AMQPBrokerConnectConfiguration(getTestName() + ":2", "tcp://" + remoteURI2.getHost() + ":" + remoteURI2.getPort());
                 updatedAmqpConnection.setReconnectAttempts(0);// No reconnects
                 updatedAmqpConnection.addElement(updatedElement);
  
                 server.getConfiguration().addAMQPConnection(updatedAmqpConnection);
  
                 protocolFactory.updateProtocolServices(server, Collections.emptyList());
  
                 peer2.waitForScriptToComplete(5, TimeUnit.SECONDS);
                 peer2.close();
              }
  
              peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
              peer.close();
           }
        }
     }
  
     @Test
     @Timeout(20)
     public void testBridgeDisconnectsFromExistingPeerIfConfigurationRemoved() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
           peer.expectSASLAnonymousConnect();
           peer.expectOpen().respond();
           peer.expectBegin().respond();
           peer.start();
  
           final URI remoteURI = peer.getServerURI();
           logger.info("Test started, peer listening on: {}", remoteURI);
  
           final AMQPBridgeAddressPolicyElement receiveFromAddress = new AMQPBridgeAddressPolicyElement();
           receiveFromAddress.setName("address-policy");
           receiveFromAddress.addToIncludes(getTestName());
           receiveFromAddress.addToExcludes("test.ignore.#");
           receiveFromAddress.setPriority(1);
  
           final AMQPBridgeBrokerConnectionElement element = new AMQPBridgeBrokerConnectionElement();
           element.setName(getTestName());
           element.addBridgeFromAddressPolicy(receiveFromAddress);
  
           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.addAddressInfo(new AddressInfo(SimpleString.of(getTestName()), RoutingType.MULTICAST));
  
           peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
           peer.expectAttach().ofReceiver()
                              .withName(allOf(containsString(getTestName()),
                                              containsString("address-receiver"),
                                              containsString("amqp-bridge"),
                                              containsString(server.getNodeID().toString())))
                              .respond();
           peer.expectFlow().withLinkCredit(1000);
  
           final ConnectionFactory factory = CFUtil.createConnectionFactory("AMQP", "tcp://localhost:" + AMQP_PORT);
  
           try (Connection connection = factory.createConnection()) {
              final Session session = connection.createSession(Session.AUTO_ACKNOWLEDGE);
              session.createConsumer(session.createTopic(getTestName()));
  
              connection.start();
  
              peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
              peer.expectDetach().optional();
              peer.expectClose().optional();
              peer.expectConnectionToDrop();
  
              final ProtonProtocolManagerFactory protocolFactory = (ProtonProtocolManagerFactory)
                 server.getRemotingService().getProtocolFactoryMap().get("AMQP");
              assertNotNull(protocolFactory);
  
              server.getConfiguration().clearAMQPConnectionConfigurations();
  
              protocolFactory.updateProtocolServices(server, Collections.emptyList());
  
              peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
  
              // Create more demand, no federation should be initiated
              session.createConsumer(session.createTopic(getTestName()));
  
              peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
              peer.close();
           }
        }
     }
  
     @Test
     @Timeout(20)
     public void testBidgeUpdatesPolicyAndBridgesQueueInsteadOfAddress() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
           peer.expectSASLAnonymousConnect();
           peer.expectOpen().respond();
           peer.expectBegin().respond();
           peer.start();
  
           final URI remoteURI = peer.getServerURI();
           logger.info("Test started, peer listening on: {}", remoteURI);
  
           final AMQPBridgeAddressPolicyElement receiveFromAddress = new AMQPBridgeAddressPolicyElement();
           receiveFromAddress.setName("address-policy");
           receiveFromAddress.addToIncludes(getTestName());
           receiveFromAddress.setPriority(1);
  
           final AMQPBridgeBrokerConnectionElement element = new AMQPBridgeBrokerConnectionElement();
           element.setName(getTestName());
           element.addBridgeFromAddressPolicy(receiveFromAddress);
  
           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.addAddressInfo(new AddressInfo(SimpleString.of(getTestName()), RoutingType.MULTICAST));
  
           peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
           peer.expectAttach().ofReceiver()
                              .withName(allOf(containsString(getTestName()),
                                              containsString("address-receiver"),
                                              containsString("amqp-bridge"),
                                              containsString(server.getNodeID().toString())))
                              .respond();
           peer.expectFlow().withLinkCredit(1000);
  
           final ConnectionFactory factory = CFUtil.createConnectionFactory("AMQP", "tcp://localhost:" + AMQP_PORT);
  
           try (Connection connection = factory.createConnection()) {
              final Session session = connection.createSession(Session.AUTO_ACKNOWLEDGE);
              session.createConsumer(session.createTopic(getTestName()));
              session.createConsumer(session.createQueue("queue"));
  
              connection.start();
  
              peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
              peer.expectDetach().optional();
              peer.expectClose().optional();
              peer.expectConnectionToDrop();
              peer.expectSASLAnonymousConnect();
              peer.expectOpen().respond();
              peer.expectBegin().respond();
              peer.expectAttach().ofReceiver()
                                 .withName(allOf(containsString(getTestName()),
                                                 containsString("queue"),
                                                 containsString("queue-receiver"),
                                                 containsString("amqp-bridge"),
                                                 containsString(server.getNodeID().toString())))
                                 .respond();
              peer.expectFlow().withLinkCredit(1000);
  
              final ProtonProtocolManagerFactory protocolFactory = (ProtonProtocolManagerFactory)
                 server.getRemotingService().getProtocolFactoryMap().get("AMQP");
              assertNotNull(protocolFactory);
  
              final AMQPBridgeQueuePolicyElement updatedReceiveFromQueue = new AMQPBridgeQueuePolicyElement();
              updatedReceiveFromQueue.setName("queue-policy");
              updatedReceiveFromQueue.addToIncludes("*", "queue");
  
              final AMQPBridgeBrokerConnectionElement updatedElement = new AMQPBridgeBrokerConnectionElement();
              updatedElement.setName(getTestName());
              updatedElement.addBridgeFromQueuePolicy(updatedReceiveFromQueue);
  
              final AMQPBrokerConnectConfiguration updatedAmqpConnection =
                 new AMQPBrokerConnectConfiguration(getTestName(), "tcp://" + remoteURI.getHost() + ":" + remoteURI.getPort());
              updatedAmqpConnection.setReconnectAttempts(0);// No reconnects
              updatedAmqpConnection.addElement(updatedElement);
  
              server.getConfiguration().getAMQPConnection().clear();
              server.getConfiguration().addAMQPConnection(updatedAmqpConnection);
  
              protocolFactory.updateProtocolServices(server, Collections.emptyList());
  
              peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
              peer.close();
           }
        }
     }
  }
  