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

import java.lang.invoke.MethodHandles;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.activemq.artemis.api.core.ActiveMQException;
import org.apache.activemq.artemis.api.core.QueueConfiguration;
import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.api.core.client.ClientConsumer;
import org.apache.activemq.artemis.api.core.client.ClientMessage;
import org.apache.activemq.artemis.api.core.client.ClientProducer;
import org.apache.activemq.artemis.api.core.client.ClientSession;
import org.apache.activemq.artemis.api.core.client.ClientSessionFactory;
import org.apache.activemq.artemis.api.core.client.ServerLocator;
import org.apache.activemq.artemis.core.config.Configuration;
import org.apache.activemq.artemis.core.io.nio.NIOSequentialFileFactory;
import org.apache.activemq.artemis.core.journal.LoaderCallback;
import org.apache.activemq.artemis.core.journal.PreparedTransactionInfo;
import org.apache.activemq.artemis.core.journal.RecordInfo;
import org.apache.activemq.artemis.core.journal.impl.JournalImpl;
import org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds;
import org.apache.activemq.artemis.core.server.ActiveMQServer;
import org.apache.activemq.artemis.tests.util.ActiveMQTestBase;
import org.junit.Assert;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class RedeliveryConsumerTest extends ActiveMQTestBase {

   private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());

   ActiveMQServer server;

   final SimpleString ADDRESS = new SimpleString("address");

   ClientSessionFactory factory;

   private ServerLocator locator;



   @Test
   public void testRedeliveryMessageStrict() throws Exception {
      testDedeliveryMessageOnPersistent(true);
   }

   @Test
   public void testRedeliveryMessageSimpleCancel() throws Exception {
      testDedeliveryMessageOnPersistent(false);
   }

   @Test
   public void testDeliveryNonPersistent() throws Exception {
      testDelivery(false);
   }

   @Test
   public void testDeliveryPersistent() throws Exception {
      testDelivery(true);
   }

   public void testDelivery(final boolean persistent) throws Exception {
      setUp(true);
      ClientSession session = factory.createSession(false, false, false);
      ClientProducer prod = session.createProducer(ADDRESS);

      for (int i = 0; i < 10; i++) {
         prod.send(createTextMessage(session, Integer.toString(i), persistent));
      }

      session.commit();
      session.close();

      session = factory.createSession(null, null, false, true, true, true, 0);

      session.start();
      for (int loopAck = 0; loopAck < 5; loopAck++) {
         ClientConsumer browser = session.createConsumer(ADDRESS, null, true);
         for (int i = 0; i < 10; i++) {
            ClientMessage msg = browser.receive(1000);
            Assert.assertNotNull("element i=" + i + " loopAck = " + loopAck + " was expected", msg);
            msg.acknowledge();
            Assert.assertEquals(Integer.toString(i), getTextMessage(msg));

            // We don't change the deliveryCounter on Browser, so this should be always 0
            Assert.assertEquals(0, msg.getDeliveryCount());
         }

         session.commit();
         browser.close();
      }

      session.close();

      session = factory.createSession(false, false, false);
      session.start();

      ClientConsumer consumer = session.createConsumer(ADDRESS);

      for (int loopAck = 0; loopAck < 5; loopAck++) {
         for (int i = 0; i < 10; i++) {
            ClientMessage msg = consumer.receive(1000);
            Assert.assertNotNull(msg);
            Assert.assertEquals(Integer.toString(i), getTextMessage(msg));

            // No ACK done, so deliveryCount should be always = 1
            Assert.assertEquals(1, msg.getDeliveryCount());
         }
         session.rollback();
      }

      if (persistent) {
         session.close();
         server.stop();
         server.start();
         factory = createSessionFactory(locator);
         session = factory.createSession(false, false, false);
         session.start();
         consumer = session.createConsumer(ADDRESS);
      }

      for (int loopAck = 1; loopAck <= 5; loopAck++) {
         for (int i = 0; i < 10; i++) {
            ClientMessage msg = consumer.receive(1000);
            Assert.assertNotNull(msg);
            msg.acknowledge();
            Assert.assertEquals(Integer.toString(i), getTextMessage(msg));
            Assert.assertEquals(loopAck, msg.getDeliveryCount());
         }
         if (loopAck < 5) {
            if (persistent) {
               session.close();
               server.stop();
               server.start();
               factory = createSessionFactory(locator);
               session = factory.createSession(false, false, false);
               session.start();
               consumer = session.createConsumer(ADDRESS);
            } else {
               session.rollback();
            }
         }
      }

      session.close();
   }

   protected void testDedeliveryMessageOnPersistent(final boolean strictUpdate) throws Exception {
      setUp(strictUpdate);
      ClientSession session = factory.createSession(false, false, false);

      logger.debug("created");

      ClientProducer prod = session.createProducer(ADDRESS);
      prod.send(createTextMessage(session, "Hello"));
      session.commit();
      session.close();

      session = factory.createSession(false, false, false);
      session.start();
      ClientConsumer consumer = session.createConsumer(ADDRESS);

      ClientMessage msg = consumer.receive(1000);
      Assert.assertEquals(1, msg.getDeliveryCount());
      session.stop();

      // if strictUpdate == true, this will simulate a crash, where the server is stopped without closing/rolling back
      // the session, but the delivery count still persisted, so the final delivery count is 2 too.
      if (!strictUpdate) {
         // If non Strict, at least rollback/cancel should still update the delivery-counts
         session.rollback(true);

         session.close();
      }

      server.stop();

      // once the server is stopped, we close the session
      // to clean up its client resources
      session.close();

      server.start();

      factory = createSessionFactory(locator);

      session = factory.createSession(false, true, false);
      session.start();
      consumer = session.createConsumer(ADDRESS);
      msg = consumer.receive(1000);
      Assert.assertNotNull(msg);
      Assert.assertEquals(strictUpdate ? 2 : 2, msg.getDeliveryCount());
      session.close();
   }

   @Test
   public void testInfiniteDedeliveryMessageOnPersistent() throws Exception {
      internaltestInfiniteDedeliveryMessageOnPersistent(false);
   }

   private void internaltestInfiniteDedeliveryMessageOnPersistent(final boolean strict) throws Exception {
      setUp(strict);
      ClientSession session = factory.createSession(false, false, false);

      logger.debug("created");

      ClientProducer prod = session.createProducer(ADDRESS);
      prod.send(createTextMessage(session, "Hello"));
      session.commit();
      session.close();

      int expectedCount = 1;
      for (int i = 0; i < 700; i++) {
         session = factory.createSession(false, false, false);
         session.start();
         ClientConsumer consumer = session.createConsumer(ADDRESS);
         ClientMessage msg = consumer.receive(5000);
         assertNotNull(msg);
         assertEquals(expectedCount, msg.getDeliveryCount());

         if (i % 100 == 0) {
            expectedCount++;
            msg.acknowledge();
            session.rollback();
         }
         session.close();
      }

      factory.close();
      server.stop();

      setUp(false);

      for (int i = 0; i < 700; i++) {
         session = factory.createSession(false, false, false);
         session.start();
         ClientConsumer consumer = session.createConsumer(ADDRESS);
         ClientMessage msg = consumer.receive(5000);
         assertNotNull(msg);
         assertEquals(expectedCount, msg.getDeliveryCount());
         session.close();
      }

      server.stop();

      JournalImpl journal = new JournalImpl(server.getConfiguration().getJournalFileSize(), 2, 2, 0, 0, new NIOSequentialFileFactory(server.getConfiguration().getJournalLocation(), 1), "activemq-data", "amq", 1);

      final AtomicInteger updates = new AtomicInteger();

      journal.start();
      journal.load(new LoaderCallback() {

         @Override
         public void failedTransaction(long transactionID, List<RecordInfo> records, List<RecordInfo> recordsToDelete) {
         }

         @Override
         public void updateRecord(RecordInfo info) {
            if (info.userRecordType == JournalRecordIds.UPDATE_DELIVERY_COUNT) {
               updates.incrementAndGet();
            }
         }

         @Override
         public void deleteRecord(long id) {
         }

         @Override
         public void addRecord(RecordInfo info) {
         }

         @Override
         public void addPreparedTransaction(PreparedTransactionInfo preparedTransaction) {
         }
      });

      journal.stop();

      assertEquals(7, updates.get());

   }

   @Test
   public void testRedeliveryCollisionAvoidance() throws Exception {
      setUp(false);
      int numberOfThreads = 10;
      long redeliveryDelay = 1000;
      server.getAddressSettingsRepository().getMatch(ADDRESS.toString()).setRedeliveryDelay(redeliveryDelay).setRedeliveryCollisionAvoidanceFactor(0.5);

      ClientSession session = factory.createSession(false, false, false);
      ClientProducer prod = session.createProducer(ADDRESS);
      for (int i = 0; i < numberOfThreads; i++) {
         prod.send(createTextMessage(session, "Hello" + i));
      }
      session.commit();
      session.close();

      final CountDownLatch aligned = new CountDownLatch(numberOfThreads);
      final CountDownLatch startRollback = new CountDownLatch(1);

      class ConsumerThread extends Thread {

         ConsumerThread(int i) {
            super("RedeliveryCollisionAvoidance::" + i);
         }

         long delay = 0;
         int errors = 0;

         @Override
         public void run() {
            try (ServerLocator locator = createInVMNonHALocator()) {
               locator.setConsumerWindowSize(0);
               ClientSessionFactory factory = locator.createSessionFactory();
               ClientSession session = factory.createSession(false, false, false);
               session.start();
               ClientConsumer consumer = session.createConsumer(ADDRESS);
               ClientMessage msg = consumer.receive(5000);
               assertNotNull(msg);
               msg.acknowledge();
               aligned.countDown();
               startRollback.await();
               session.rollback();
               long start = System.currentTimeMillis();
               msg = consumer.receive(5000);
               delay = System.currentTimeMillis() - start;
               assertNotNull(msg);
               msg.acknowledge();
               session.commit();
            } catch (Exception e) {
               e.printStackTrace();
               errors++;
            }
         }
      }

      ConsumerThread[] threads = new ConsumerThread[numberOfThreads];

      for (int i = 0; i < numberOfThreads; i++) {
         threads[i] = new ConsumerThread(i);
         threads[i].start();
      }

      aligned.await();
      startRollback.countDown();

      try {
         for (ConsumerThread t : threads) {
            t.join(60000);
            assertFalse(t.isAlive());
            assertEquals("There are Errors on the test thread", 0, t.errors);
         }
      } finally {
         for (ConsumerThread t : threads) {
            if (t.isAlive()) {
               t.interrupt();
            }
            t.join(1000);
         }
      }

      long maxDelay = 0;
      long minDelay = Long.MAX_VALUE;

      for (ConsumerThread t : threads) {
         if (t.delay < minDelay) {
            minDelay = t.delay;
         }
         if (t.delay > maxDelay) {
            maxDelay = t.delay;
         }
      }

      // make sure the difference between the minimum redelivery delay and the maximum redelivery delay is larger that the expected nominal variance
      assertTrue((maxDelay - minDelay) > (redeliveryDelay * .05));

      factory.close();
   }



   /**
    * @param persistDeliveryCountBeforeDelivery
    * @throws Exception
    */
   private void setUp(final boolean persistDeliveryCountBeforeDelivery) throws Exception {
      Configuration config = createDefaultInVMConfig().setPersistDeliveryCountBeforeDelivery(persistDeliveryCountBeforeDelivery);

      server = createServer(true, config);

      server.start();
      locator = createInVMNonHALocator();
      factory = createSessionFactory(locator);

      ClientSession session = addClientSession(factory.createSession(false, false, false));
      try {
         session.createQueue(new QueueConfiguration(ADDRESS));
      } catch (ActiveMQException expected) {
         // in case of restart
      }

      session.close();
   }
}
