/*
 * 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.routing;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.Map;

import org.apache.activemq.artemis.api.core.QueueConfiguration;
import org.apache.activemq.artemis.api.core.RoutingType;
import org.apache.activemq.artemis.api.core.management.QueueControl;
import org.apache.activemq.artemis.api.core.management.ResourceNames;
import org.apache.activemq.artemis.core.remoting.impl.netty.TransportConstants;
import org.apache.activemq.artemis.core.server.cluster.impl.MessageLoadBalancingType;
import org.apache.activemq.artemis.core.server.routing.KeyType;
import org.apache.activemq.artemis.core.server.routing.policies.ConsistentHashPolicy;
import org.apache.activemq.artemis.core.server.routing.policies.FirstElementPolicy;
import org.apache.activemq.artemis.core.server.routing.policies.LeastConnectionsPolicy;
import org.apache.activemq.artemis.core.server.routing.policies.RoundRobinPolicy;
import org.apache.activemq.artemis.tests.extensions.parameterized.ParameterizedTestExtension;
import org.apache.activemq.artemis.tests.extensions.parameterized.Parameters;
import org.junit.jupiter.api.TestTemplate;
import org.junit.jupiter.api.extension.ExtendWith;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertTrue;

@ExtendWith(ParameterizedTestExtension.class)
public class RedirectTest extends RoutingTestBase {

   @Parameters(name = "pool: {0}")
   public static Collection<Object[]> parameters() {
      return Arrays.asList(new Object[][] {{CLUSTER_POOL}, {DISCOVERY_POOL}, {STATIC_POOL} });
   }


   private final String pool;


   public RedirectTest(String pool) {
      this.pool = pool;
   }

   @TestTemplate
   public void testSimpleRedirect() throws Exception {
      final String queueName = "RedirectTestQueue";

      setupPrimaryServerWithDiscovery(0, GROUP_ADDRESS, GROUP_PORT, true, true, false);
      setupPrimaryServerWithDiscovery(1, GROUP_ADDRESS, GROUP_PORT, true, true, false);
      if (CLUSTER_POOL.equals(pool)) {
         setupDiscoveryClusterConnection("cluster0", 0, "dg1", "queues", MessageLoadBalancingType.OFF, 1, true);
         setupDiscoveryClusterConnection("cluster1", 1, "dg1", "queues", MessageLoadBalancingType.OFF, 1, true);
         setupRouterServerWithCluster(0, KeyType.USER_NAME, FirstElementPolicy.NAME, null, false, "ACTIVEMQ.CLUSTER.ADMIN.USER", 1, "cluster0");
      } else if (DISCOVERY_POOL.equals(pool)) {
         setupRouterServerWithDiscovery(0, KeyType.USER_NAME, FirstElementPolicy.NAME, null, false, null, 1);
      } else {
         setupRouterServerWithStaticConnectors(0, KeyType.USER_NAME, FirstElementPolicy.NAME, null, false, null, 1, 1);
      }

      startServers(0, 1);

      getServer(0).createQueue(QueueConfiguration.of(queueName).setRoutingType(RoutingType.ANYCAST));
      getServer(1).createQueue(QueueConfiguration.of(queueName).setRoutingType(RoutingType.ANYCAST));

      testSimpleRedirectWithProtocol(AMQP_PROTOCOL, queueName);
      testSimpleRedirectWithProtocol(CORE_PROTOCOL, queueName);
      testSimpleRedirectWithProtocol(OPENWIRE_PROTOCOL, queueName);

      stopServers(0, 1);
   }

   private void testSimpleRedirectWithProtocol(final String protocol, final String queueName) throws Exception {
      QueueControl queueControl0 = (QueueControl)getServer(0).getManagementService()
          .getResource(ResourceNames.QUEUE + queueName);
      QueueControl queueControl1 = (QueueControl)getServer(1).getManagementService()
          .getResource(ResourceNames.QUEUE + queueName);

      assertEquals(0, queueControl0.countMessages());
      assertEquals(0, queueControl1.countMessages());

      ConnectionFactory connectionFactory = createFactory(protocol, false, TransportConstants.DEFAULT_HOST,
          TransportConstants.DEFAULT_PORT + 0, null, "admin", "admin");


      try (Connection connection = connectionFactory.createConnection()) {
         connection.start();
         try (Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE)) {
            javax.jms.Queue queue = session.createQueue(queueName);
            try (MessageProducer producer = session.createProducer(queue)) {
               producer.send(session.createTextMessage("TEST" + protocol));
            }
         }
      }

      assertEquals(0, queueControl0.countMessages());
      assertEquals(1, queueControl1.countMessages());

      try (Connection connection = connectionFactory.createConnection()) {
         connection.start();
         try (Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE)) {
            try (MessageConsumer consumer = session.createConsumer(session.createQueue(queueName))) {
               TextMessage message = (TextMessage) consumer.receive(1000);
               assertNotNull(message);
               assertEquals("TEST" + protocol, message.getText());
            }
         }
      }

      assertEquals(0, queueControl0.countMessages());
      assertEquals(0, queueControl1.countMessages());
   }

   @TestTemplate
   public void testRoundRobinRedirect() throws Exception {
      testEvenlyRedirect(RoundRobinPolicy.NAME, null, false);
   }

   @TestTemplate
   public void testLeastConnectionsRedirect() throws Exception {
      testEvenlyRedirect(LeastConnectionsPolicy.NAME, Collections.singletonMap(
         LeastConnectionsPolicy.CONNECTION_COUNT_THRESHOLD, String.valueOf(30)), false);
   }

   @TestTemplate
   public void testRoundRobinRedirectWithFailure() throws Exception {
      testEvenlyRedirect(RoundRobinPolicy.NAME, null, true);
   }

   @TestTemplate
   public void testLeastConnectionsRedirectWithFailure() throws Exception {
      testEvenlyRedirect(LeastConnectionsPolicy.NAME, Collections.singletonMap(
         LeastConnectionsPolicy.CONNECTION_COUNT_THRESHOLD, String.valueOf(30)), true);
   }

   private void testEvenlyRedirect(final String policyName, final Map<String, String> properties, final boolean withFailure) throws Exception {
      final String queueName = "RedirectTestQueue";
      final int[] nodes = new int[] {0, 1, 2, 3};

      for (int node : nodes) {
         setupPrimaryServerWithDiscovery(node, GROUP_ADDRESS, GROUP_PORT, true, true, false);
      }

      if (CLUSTER_POOL.equals(pool)) {
         for (int node : nodes) {
            setupDiscoveryClusterConnection("cluster" + node, node, "dg1", "queues", MessageLoadBalancingType.OFF, 1, true);
         }
         setupRouterServerWithCluster(0, KeyType.USER_NAME, policyName, properties, false, "ACTIVEMQ.CLUSTER.ADMIN.USER", nodes.length - 1, "cluster0");
      } else if (DISCOVERY_POOL.equals(pool)) {
         setupRouterServerWithDiscovery(0, KeyType.USER_NAME, policyName, properties, false, null, nodes.length - 1);
      } else {
         setupRouterServerWithStaticConnectors(0, KeyType.USER_NAME, policyName, properties, false, null, nodes.length - 1, 1, 2, 3);
      }

      if (withFailure) {
         setupRouterLocalCache(0, true, 0);
      }

      startServers(nodes);

      for (int node : nodes) {
         getServer(node).createQueue(QueueConfiguration.of(queueName).setRoutingType(RoutingType.ANYCAST));
      }

      testEvenlyRedirectWithProtocol(AMQP_PROTOCOL, queueName, nodes, withFailure);
      testEvenlyRedirectWithProtocol(CORE_PROTOCOL, queueName, nodes, withFailure);
      testEvenlyRedirectWithProtocol(OPENWIRE_PROTOCOL, queueName, nodes, withFailure);

      stopServers(nodes);
   }

   private void testEvenlyRedirectWithProtocol(final String protocol, final String queueName, final int[] nodes, final boolean withFailure) throws Exception {
      QueueControl[] queueControls = new QueueControl[nodes.length];

      for (int node : nodes) {
         queueControls[node] = (QueueControl)getServer(node).getManagementService()
             .getResource(ResourceNames.QUEUE + queueName);

         assertEquals(0, queueControls[node].countMessages(), "Unexpected message count for node " + node);
      }

      ConnectionFactory[] connectionFactories = new ConnectionFactory[nodes.length - 1];
      Connection[] connections = new Connection[nodes.length - 1];
      Session[] sessions = new Session[nodes.length - 1];

      for (int i = 0; i < nodes.length - 1; i++) {
         connectionFactories[i] = createFactory(protocol, false, TransportConstants.DEFAULT_HOST,
             TransportConstants.DEFAULT_PORT + 0, null, "user" + i, "user" + i);

         connections[i] = connectionFactories[i].createConnection();
         connections[i].start();

         sessions[i] = connections[i].createSession(false, Session.AUTO_ACKNOWLEDGE);
      }

      for (int i = 0; i < nodes.length - 1; i++) {
         try (MessageProducer producer = sessions[i].createProducer(sessions[i].createQueue(queueName))) {
            producer.send(sessions[i].createTextMessage("TEST" + protocol + i));
         }

         sessions[i].close();
         connections[i].close();
      }

      assertEquals(0, queueControls[0].countMessages());
      for (int targetNode = 1; targetNode < nodes.length - 1; targetNode++) {
         assertEquals(1, queueControls[targetNode].countMessages(), "Messages of node " + targetNode);
      }

      if (withFailure) {
         crashAndWaitForFailure(getServer(0));

         startServers(0);

         queueControls[0] = (QueueControl)getServer(0).getManagementService()
             .getResource(ResourceNames.QUEUE + queueName);
      }

      for (int i = 0; i < nodes.length - 1; i++) {
         try (Connection connection = connectionFactories[i].createConnection()) {
            connection.start();
            try (Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE)) {
               try (MessageConsumer consumer = session.createConsumer(session.createQueue(queueName))) {
                  TextMessage message = (TextMessage) consumer.receive(1000);
                  assertNotNull(message);
                  assertEquals("TEST" + protocol + i, message.getText());
               }
            }
         }
      }

      for (int node : nodes) {
         assertEquals(0, queueControls[node].countMessages(), "Unexpected message count for node " + node);
      }
   }

   @TestTemplate
   public void testSymmetricRedirect() throws Exception {
      final String queueName = "RedirectTestQueue";

      setupPrimaryServerWithDiscovery(0, GROUP_ADDRESS, GROUP_PORT, true, true, false);
      setupPrimaryServerWithDiscovery(1, GROUP_ADDRESS, GROUP_PORT, true, true, false);
      if (CLUSTER_POOL.equals(pool)) {
         setupDiscoveryClusterConnection("cluster0", 0, "dg1", "queues", MessageLoadBalancingType.OFF, 1, true);
         setupDiscoveryClusterConnection("cluster1", 1, "dg1", "queues", MessageLoadBalancingType.OFF, 1, true);
         setupRouterServerWithCluster(0, KeyType.USER_NAME, ConsistentHashPolicy.NAME, null, true, "ACTIVEMQ.CLUSTER.ADMIN.USER", 2, "cluster0");
         setupRouterServerWithCluster(1, KeyType.USER_NAME, ConsistentHashPolicy.NAME, null, true, "ACTIVEMQ.CLUSTER.ADMIN.USER", 2, "cluster1");
      } else if (DISCOVERY_POOL.equals(pool)) {
         setupRouterServerWithDiscovery(0, KeyType.USER_NAME, ConsistentHashPolicy.NAME, null, true, null, 2);
         setupRouterServerWithDiscovery(1, KeyType.USER_NAME, ConsistentHashPolicy.NAME, null, true, null, 2);
      } else {
         setupRouterServerWithStaticConnectors(0, KeyType.USER_NAME, ConsistentHashPolicy.NAME, null, true, null, 2, 1);
         setupRouterServerWithStaticConnectors(1, KeyType.USER_NAME, ConsistentHashPolicy.NAME, null, true, null, 2, 0);
      }

      startServers(0, 1);

      assertNotEquals(getServer(0).getNodeID(), getServer(1).getNodeID());

      getServer(0).createQueue(QueueConfiguration.of(queueName).setRoutingType(RoutingType.ANYCAST));
      getServer(1).createQueue(QueueConfiguration.of(queueName).setRoutingType(RoutingType.ANYCAST));

      testSymmetricRedirectWithProtocol(AMQP_PROTOCOL, queueName);
      testSymmetricRedirectWithProtocol(CORE_PROTOCOL, queueName);
      testSymmetricRedirectWithProtocol(OPENWIRE_PROTOCOL, queueName);

      stopServers(0, 1);
   }

   private void testSymmetricRedirectWithProtocol(final String protocol, final String queueName) throws Exception {
      QueueControl queueControl0 = (QueueControl)getServer(0).getManagementService()
          .getResource(ResourceNames.QUEUE + queueName);
      QueueControl queueControl1 = (QueueControl)getServer(1).getManagementService()
          .getResource(ResourceNames.QUEUE + queueName);

      assertEquals(0, queueControl0.countMessages(), "Unexpected message count for node 0");
      assertEquals(0, queueControl1.countMessages(), "Unexpected message count for node 1");

      ConnectionFactory connectionFactory0 = createFactory(protocol, false, TransportConstants.DEFAULT_HOST,
          TransportConstants.DEFAULT_PORT + 0, null, "admin", "admin");


      try (Connection connection = connectionFactory0.createConnection()) {
         connection.start();
         try (Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE)) {
            javax.jms.Queue queue = session.createQueue(queueName);
            try (MessageProducer producer = session.createProducer(queue)) {
               producer.send(session.createTextMessage("TEST" + protocol));
            }
         }
      }

      assertTrue((queueControl0.countMessages() == 0 && queueControl1.countMessages() == 1) ||
          (queueControl0.countMessages() == 1 && queueControl1.countMessages() == 0));

      assertNotEquals(getServer(0).getNodeID(), getServer(1).getNodeID());

      ConnectionFactory connectionFactory1 = createFactory(protocol, false, TransportConstants.DEFAULT_HOST,
          TransportConstants.DEFAULT_PORT + 1, null, "admin", "admin");

      try (Connection connection = connectionFactory1.createConnection()) {
         connection.start();
         try (Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE)) {
            try (MessageConsumer consumer = session.createConsumer(session.createQueue(queueName))) {
               TextMessage message = (TextMessage) consumer.receive(1000);
               assertNotNull(message);
               assertEquals("TEST" + protocol, message.getText());
            }
         }
      }

      assertEquals(0, queueControl0.countMessages(), "Unexpected message count for node 0");
      assertEquals(0, queueControl1.countMessages(), "Unexpected message count for node 1");
   }

   @TestTemplate
   public void testRedirectAfterFailure() throws Exception {
      final String queueName = "RedirectTestQueue";
      final int[] nodes = new int[] {0, 1, 2};

      for (int node : nodes) {
         setupPrimaryServerWithDiscovery(node, GROUP_ADDRESS, GROUP_PORT, true, true, false);
      }
      if (CLUSTER_POOL.equals(pool)) {
         for (int node : nodes) {
            setupDiscoveryClusterConnection("cluster" + node, node, "dg1", "queues", MessageLoadBalancingType.OFF, 1, true);
         }
         setupRouterServerWithCluster(0, KeyType.USER_NAME, FirstElementPolicy.NAME, null, false, "ACTIVEMQ.CLUSTER.ADMIN.USER", 1, "cluster0");
      } else if (DISCOVERY_POOL.equals(pool)) {
         setupRouterServerWithDiscovery(0, KeyType.USER_NAME, FirstElementPolicy.NAME, null, false, null, 1);
      } else {
         setupRouterServerWithStaticConnectors(0, KeyType.USER_NAME, FirstElementPolicy.NAME, null, false, null, 1, 1, 2);
      }

      startServers(0, 1, 2);

      for (int node : nodes) {
         getServer(node).createQueue(QueueConfiguration.of(queueName).setRoutingType(RoutingType.ANYCAST));
      }

      testRedirectAfterFailureWithProtocol(AMQP_PROTOCOL, queueName, nodes);
      testRedirectAfterFailureWithProtocol(CORE_PROTOCOL, queueName, nodes);
      testRedirectAfterFailureWithProtocol(OPENWIRE_PROTOCOL, queueName, nodes);

      stopServers(0, 1, 2);
   }

   private void testRedirectAfterFailureWithProtocol(final String protocol, final String queueName, final int[] nodes) throws Exception {
      QueueControl[] queueControls = new QueueControl[nodes.length];
      for (int node : nodes) {
         queueControls[node] = (QueueControl)getServer(node).getManagementService()
             .getResource(ResourceNames.QUEUE + queueName);

         assertEquals(0, queueControls[node].countMessages(), "Unexpected message count for node " + node);
      }

      int failedNode;
      ConnectionFactory connectionFactory = createFactory(protocol, false, TransportConstants.DEFAULT_HOST,
          TransportConstants.DEFAULT_PORT + 0, null, "admin", "admin");


      try (Connection connection = connectionFactory.createConnection()) {
         connection.start();
         try (Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE)) {
            javax.jms.Queue queue = session.createQueue(queueName);
            try (MessageProducer producer = session.createProducer(queue)) {
               producer.send(session.createTextMessage("TEST_BEFORE_FAILURE"));

               if (queueControls[1].countMessages() > 0) {
                  failedNode = 1;
               } else {
                  failedNode = 2;
               }

               stopServers(failedNode);

               producer.send(session.createTextMessage("TEST_AFTER_FAILURE"));
            }
         }
      }

      startServers(failedNode);

      queueControls[failedNode] = (QueueControl)getServer(failedNode).getManagementService()
          .getResource(ResourceNames.QUEUE + queueName);

      try (Connection connection = connectionFactory.createConnection()) {
         connection.start();
         try (Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE)) {
            try (MessageConsumer consumer = session.createConsumer(session.createQueue(queueName))) {
               TextMessage message = (TextMessage) consumer.receive(1000);
               assertNotNull(message);
               assertEquals("TEST_AFTER_FAILURE", message.getText());
            }
         }
      }

      for (int node : nodes) {
         if (node == failedNode) {
            assertEquals(1, queueControls[node].countMessages(), "Unexpected message count for node " + node);
            queueControls[node].removeAllMessages();
         } else {
            assertEquals(0, queueControls[node].countMessages(), "Unexpected message count for node " + node);
         }
      }
   }
}
