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

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.artemis.api.core.RoutingType;
import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.api.jms.ActiveMQJMSClient;
import org.apache.activemq.artemis.core.postoffice.Binding;
import org.apache.activemq.artemis.core.postoffice.Bindings;
import org.apache.activemq.artemis.core.server.ActiveMQServer;
import org.apache.activemq.artemis.core.server.cluster.RemoteQueueBinding;
import org.apache.activemq.artemis.core.server.cluster.impl.MessageLoadBalancingType;
import org.apache.activemq.artemis.tests.integration.cluster.distribution.ClusterTestBase;
import org.apache.activemq.artemis.tests.util.Wait;
import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.util.ConsumerThread;
import org.junit.Assert;
import org.junit.Test;

import javax.jms.Connection;
import javax.jms.Destination;
import javax.jms.Message;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;
import javax.jms.TopicSubscriber;
import java.util.Collection;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

public class MessageRedistributionTest extends ClusterTestBase {

   @Override
   protected boolean isForceUniqueStorageManagerIds() {
      // we want to verify messageId uniqueness across brokers
      return false;
   }

   @Test
   public void testRemoteConsumerClose() throws Exception {

      setupServer(0, true, true);
      setupServer(1, true, true);

      setupClusterConnection("cluster0", "queues", MessageLoadBalancingType.ON_DEMAND, 1, true, 0, 1);
      setupClusterConnection("cluster1", "queues", MessageLoadBalancingType.ON_DEMAND, 1, true, 1, 0);

      startServers(0, 1);

      waitForTopology(servers[0], 2);
      waitForTopology(servers[1], 2);

      setupSessionFactory(0, true);
      setupSessionFactory(1, true);

      createAddressInfo(0, "queues.testAddress", RoutingType.ANYCAST, -1, false);
      createAddressInfo(1, "queues.testAddress", RoutingType.ANYCAST, -1, false);
      createQueue(0, "queues.testaddress", "queue0", null, true);
      createQueue(1, "queues.testaddress", "queue0", null, true);

      //alternately create consumers to the 2 nodes
      //close the connection then close consumer quickly
      //check server's consumer count
      for (int i = 0; i < 50; i++) {
         int target = i % 2;
         int remote = (i + 1) % 2;
         closeConsumerAndConnectionConcurrently(target, remote);
      }
   }

   @Test
   public void testFailoverNonClusteredBrokersInteropWithCoreProducer() throws Exception {

      setupServer(0, true, true);
      setupServer(1, true, true);

      startServers(0, 1);

      servers[0].getAddressSettingsRepository().getMatch("#").setRedeliveryDelay(0).setRedistributionDelay(0);
      servers[1].getAddressSettingsRepository().getMatch("#").setRedeliveryDelay(0).setRedistributionDelay(0);

      setupSessionFactory(0, true);
      setupSessionFactory(1, true);

      createAddressInfo(0, "q", RoutingType.ANYCAST, -1, false);
      createAddressInfo(1, "q", RoutingType.ANYCAST, -1, false);
      createQueue(0, "q", "q", null, true, RoutingType.ANYCAST);
      createQueue(1, "q", "q", null, true, RoutingType.ANYCAST);


      final int numMessagesPerNode = 1000;
      produceWithCoreTo(0, numMessagesPerNode);
      produceWithCoreTo(1, numMessagesPerNode);

      // consume with openwire from both brokers which both start with journal id = 0, should be in lock step

      String zero = getServerUri(0);
      String one = getServerUri(1);

      ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("failover:(" + zero + "," + one + ")?jms.prefetchPolicy.all=10&randomize=false&timeout=400&reconnectDelay=500&useExponentialBackOff=false&initialReconnectDelay=500&nested.wireFormat.maxInactivityDuration=500&nested.wireFormat.maxInactivityDurationInitalDelay=500&nested.ignoreRemoteWireFormat=true&nested.soTimeout=500&nested.connectionTimeout=400&jms.connectResponseTimeout=400&jms.sendTimeout=400&jms.closeTimeout=400");
      factory.setWatchTopicAdvisories(false);

      CountDownLatch continueLatch = new CountDownLatch(1);
      CountDownLatch received = new CountDownLatch(numMessagesPerNode * 2);
      final Connection conn = factory.createConnection();
      conn.start();

      ((ActiveMQConnection)conn).setClientInternalExceptionListener(Throwable::printStackTrace);

      Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
      Destination dest = ActiveMQDestination.createDestination("q", ActiveMQDestination.QUEUE_TYPE);
      session.createConsumer(dest).setMessageListener(message -> {
         try {
            received.countDown();
         } catch (Exception exception) {
            exception.printStackTrace();
         }
      });


      assertTrue(Wait.waitFor(new org.apache.activemq.artemis.utils.Wait.Condition() {
         @Override
         public boolean isSatisfied() throws Exception {
            return received.getCount() <= numMessagesPerNode;
         }
      }));

      // force a failover to the other broker
      servers[0].stop(false, true);

      // get all the messages, our openwire audit does not detect any duplicate
      assertTrue(Wait.waitFor(() -> {
         return received.await(1, TimeUnit.SECONDS);
      }));

      conn.close();
   }

   private void produceWithCoreTo(int serveId, final int numMessagesPerNode) throws Exception {

      String targetUrl = getServerUri(serveId);
      Connection jmsConn = null;
      try {
         org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory coreCf = ActiveMQJMSClient.createConnectionFactory(targetUrl, "cf" + serveId);
         jmsConn = coreCf.createConnection();
         jmsConn.setClientID("theProducer");
         Session coreSession = jmsConn.createSession(false, Session.AUTO_ACKNOWLEDGE);
         TextMessage msg = coreSession.createTextMessage("TEXT");
         Queue queue = coreSession.createQueue("q");
         MessageProducer producer = coreSession.createProducer(queue);
         for (int i = 0; i < numMessagesPerNode; i++) {
            msg.setIntProperty("MM", i);
            msg.setIntProperty("SN", serveId);
            producer.send(msg);
         }
      } finally {
         if (jmsConn != null) {
            jmsConn.close();
         }
      }
   }

   @Test
   public void testAdvisoriesNotClustered() throws Exception {

      setupServer(0, true, true);
      setupServer(1, true, true);

      setupClusterConnection("cluster0", "", MessageLoadBalancingType.ON_DEMAND, 1, true, 0, 1);
      setupClusterConnection("cluster1", "", MessageLoadBalancingType.ON_DEMAND, 1, true, 1, 0);

      startServers(0, 1);

      waitForTopology(servers[0], 2);
      waitForTopology(servers[1], 2);

      setupSessionFactory(0, true);
      setupSessionFactory(1, true);

      createAddressInfo(0, "testAddress", RoutingType.MULTICAST, -1, false);
      createAddressInfo(1, "testAddress", RoutingType.MULTICAST, -1, false);

      ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory();
      ActiveMQConnectionFactory factory2 = new ActiveMQConnectionFactory(getServerUri(1));
      Connection conn = null;
      Connection conn2 = null;
      CountDownLatch active = new CountDownLatch(1);
      try {
         conn = factory.createConnection();
         conn2 = factory2.createConnection();
         conn2.setClientID("id");
         conn2.start();
         Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
         Session session2 = conn2.createSession(false, Session.AUTO_ACKNOWLEDGE);
         Topic dest = (Topic) ActiveMQDestination.createDestination("testAddress", ActiveMQDestination.TOPIC_TYPE);
         TopicSubscriber mySubscriber = session2.createDurableSubscriber(dest, "mySubscriber");
         MessageProducer producer = session.createProducer(dest);
         producer.send(session.createTextMessage("test message"));
         Message message = mySubscriber.receive(5000);
         SimpleString  advQueue = new SimpleString("ActiveMQ.Advisory.TempQueue");
         SimpleString  advTopic = new SimpleString("ActiveMQ.Advisory.TempTopic");
         //we create a consumer on node 2 and assert that the advisory subscription queue is not clustered
         Assert.assertEquals("", 1, servers[0].getPostOffice().getBindingsForAddress(advQueue).getBindings().size());
         Assert.assertEquals("", 1, servers[0].getPostOffice().getBindingsForAddress(advTopic).getBindings().size());
         Assert.assertEquals("", 1, servers[1].getPostOffice().getBindingsForAddress(advQueue).getBindings().size());
         Assert.assertEquals("", 1, servers[1].getPostOffice().getBindingsForAddress(advTopic).getBindings().size());

      } finally {
         conn.close();
         conn2.close();
      }
   }

   @Override
   protected boolean isResolveProtocols() {
      return true;
   }

   private void closeConsumerAndConnectionConcurrently(int targetNode, int remoteNode) throws Exception {

      String targetUri = getServerUri(targetNode);
      ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(targetUri);
      Connection conn = null;
      CountDownLatch active = new CountDownLatch(1);
      try {
         conn = factory.createConnection();
         conn.start();
         Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
         Destination dest = ActiveMQDestination.createDestination("queue0", ActiveMQDestination.QUEUE_TYPE);
         ConsumerThread consumer = new ConsumerThread(session, dest);
         consumer.setMessageCount(0);
         consumer.setFinished(active);
         consumer.start();

         assertTrue("consumer takes too long to finish!", active.await(5, TimeUnit.SECONDS));
      } finally {
         conn.close();
      }

      Wait.waitFor(() -> getRemoteQueueBinding(servers[remoteNode]) != null);

      //check remote server's consumer count
      RemoteQueueBinding remoteBinding = getRemoteQueueBinding(servers[remoteNode]);

      assertNotNull(remoteBinding);

      Wait.waitFor(() -> remoteBinding.consumerCount() >= 0);
      int count = remoteBinding.consumerCount();
      assertTrue("consumer count should never be negative " + count, count >= 0);
   }

   private RemoteQueueBinding getRemoteQueueBinding(ActiveMQServer server) throws Exception {
      ActiveMQServer remoteServer = server;
      Bindings bindings = remoteServer.getPostOffice().getBindingsForAddress(new SimpleString("queues.testaddress"));
      Collection<Binding> bindingSet = bindings.getBindings();

      return getRemoteQueueBinding(bindingSet);
   }

   private RemoteQueueBinding getRemoteQueueBinding(Collection<Binding> bindingSet) {
      RemoteQueueBinding remoteBinding = null;
      for (Binding b : bindingSet) {
         if (b instanceof RemoteQueueBinding) {
            remoteBinding = (RemoteQueueBinding) b;
            break;
         }
      }
      return remoteBinding;
   }
}
