/*
 * 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.unit.core.journal.impl;

import java.io.File;
import java.lang.invoke.MethodHandles;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.activemq.artemis.core.io.SequentialFileFactory;
import org.apache.activemq.artemis.core.io.aio.AIOSequentialFileFactory;
import org.apache.activemq.artemis.core.io.mapped.MappedSequentialFileFactory;
import org.apache.activemq.artemis.core.io.nio.NIOSequentialFileFactory;
import org.apache.activemq.artemis.core.journal.IOCompletion;
import org.apache.activemq.artemis.core.journal.Journal;
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.server.JournalType;
import org.apache.activemq.artemis.nativo.jlibaio.LibaioContext;
import org.apache.activemq.artemis.tests.util.ActiveMQTestBase;
import org.apache.activemq.artemis.utils.ExecutorFactory;
import org.apache.activemq.artemis.utils.SimpleIDGenerator;
import org.apache.activemq.artemis.utils.actors.OrderedExecutorFactory;
import org.apache.activemq.artemis.utils.collections.ConcurrentHashSet;
import org.junit.Assert;
import org.junit.Assume;
import org.junit.Ignore;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class BatchCommitTest extends ActiveMQTestBase {

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

   private static final int FILE_SIZE = 10 * 1024 * 1024;
   private static final int MIN_FILES = 10;
   private static final int POOL_SIZE = 10;
   private static final String FILE_PREFIX = "journal-test";
   private static final String FILE_EXTENSION = "amq";
   private static final int BUFFER_SIZE = 100 * 1024;
   private static final int BUFFER_TIMEOUT = 10 * 1024;

   private static final int MAX_AIO = 255;

   private static final int OK = 100;
   private static final int ERROR = 101;

   private static final int RECORDS = 10000;

   JournalImpl journal;

   SequentialFileFactory journalFF;

   SimpleIDGenerator idGenerator = new SimpleIDGenerator(1);

   public Journal testRunNIO(String testFolder, boolean sync) throws Throwable {
      return testRun(testFolder, JournalType.NIO, sync);
   }

   public Journal testRunMapped(String testFolder, boolean sync) throws Throwable {
      return testRun(testFolder, JournalType.MAPPED, sync);
   }


   public Journal testRunAIO(String testFolder, boolean sync) throws Throwable {
      return testRun(testFolder, JournalType.ASYNCIO, sync);
   }


   public Journal testRun(String testFolder, JournalType journalType, boolean sync) throws Throwable {

      OrderedExecutorFactory orderedExecutorFactory = getExecutorFactory();
      setupJournal(journalType, testFolder, orderedExecutorFactory);
      journal.start();
      runAfter(journal::stop);
      journal.loadInternalOnly();

      CountDownLatch latch = new CountDownLatch(RECORDS);

      ConcurrentHashSet<Long> existingRecords = new ConcurrentHashSet<>();

      AtomicInteger errors = new AtomicInteger(0);


      for (int i = 0; i < RECORDS; i++) {
         long tx = idGenerator.generateID();
         long id = idGenerator.generateID();
         long upid = idGenerator.generateID();
         existingRecords.add(tx);
         IOCompletion completion = new IOCompletion() {
            @Override
            public void storeLineUp() {
            }

            @Override
            public void done() {
               if (!existingRecords.remove(tx)) {
                  errors.incrementAndGet();
                  logger.warn("Id {} was removed before", tx);
               }
               latch.countDown();
            }

            @Override
            public void onError(int errorCode, String errorMessage) {
            }
         };

         journal.appendAddRecordTransactional(tx, id, (byte) 1, ("add " + id).getBytes());
         journal.appendUpdateRecordTransactional(tx, upid, (byte) 1, ("up " + upid).getBytes());
         journal.appendCommitRecord(tx, sync, completion, true);
      }

      if (!latch.await(10, TimeUnit.SECONDS)) {
         logger.warn("latch didn't finish, count={}", latch.getCount());
         errors.incrementAndGet();
      }

      existingRecords.forEach(l -> logger.warn("id {} still in the list", l));

      Assert.assertEquals(0, errors.get());
      Assert.assertEquals(0, existingRecords.size());

      return journal;

   }

   private OrderedExecutorFactory getExecutorFactory() {
      ExecutorService service = Executors.newFixedThreadPool(10, new ThreadFactory() {
         int counter = 0;
         @Override
         public Thread newThread(Runnable r) {
            Thread t = new Thread(r);
            t.setName("AsyncCommitTest" + (counter++));
            return t;
         }
      });
      OrderedExecutorFactory orderedExecutorFactory = new OrderedExecutorFactory(service);
      runAfter(service::shutdownNow);
      return orderedExecutorFactory;
   }

   @Test
   public void testNIO() throws Exception {
      internalTest(JournalType.NIO, "testRunNIO", true);
   }

   @Test
   public void testNIONoSync() throws Exception {
      internalTest(JournalType.NIO, "testRunNIO", false);
   }

   @Ignore // TODO: We should fix Mapped eventually for this case
   public void testMapped() throws Exception {
      internalTest(JournalType.MAPPED, "testRunMapped", true);
   }

   @Test
   public void testMappedNoSync() throws Exception {
      internalTest(JournalType.MAPPED, "testRunMapped", false);
   }

   @Test
   public void testAIO() throws Exception {
      Assume.assumeTrue(LibaioContext.isLoaded());
      internalTest(JournalType.ASYNCIO, "testRunAIO", true);
   }

   @Test
   public void testAIONoSync() throws Exception {
      Assume.assumeTrue(LibaioContext.isLoaded());
      internalTest(JournalType.ASYNCIO, "testRunAIO", false);
   }

   private void proceedCall(String testName, String testFolder, boolean sync) throws Exception {
      Method method = getClass().getMethod(testName, String.class, Boolean.TYPE);
      Journal journal = (Journal)method.invoke(this, testFolder, sync);
      journal.stop();
   }

   private void internalTest(JournalType journalType, String testRunName, boolean sync) throws Exception {
      proceedCall(testRunName, getTestDir(), sync);

      OrderedExecutorFactory orderedExecutorFactory = getExecutorFactory();
      setupJournal(journalType, getTestDir(), orderedExecutorFactory);

      ArrayList<RecordInfo> commited = new ArrayList<>();
      ArrayList<PreparedTransactionInfo> prepared = new ArrayList<>();
      AtomicInteger failedTX = new AtomicInteger(0);

      journal.start();
      journal.load(commited, prepared, (id, records, toDelete) -> failedTX.incrementAndGet(), false);
      runAfter(journal::stop);

      commited.forEach(r -> {
         String dataAsString = new String(r.data);
         logger.debug("data={}, isUpdate={}, id={}", dataAsString, r.isUpdate, r.id);
         if (r.isUpdate) {
            Assert.assertEquals("up " + r.id, dataAsString);
         } else {
            Assert.assertEquals("add " + r.id, dataAsString);
         }
      });
      Assert.assertEquals(RECORDS * 2, commited.size());
      Assert.assertEquals(0, failedTX.get());



   }



   public void setupJournal(JournalType journalType, String location, ExecutorFactory executorFactory) {

      File locationFile = new File(location);

      switch (journalType) {

         case NIO:
            journalFF = new NIOSequentialFileFactory(locationFile, true, BUFFER_SIZE, BUFFER_TIMEOUT, 1, true, null, null);
            break;
         case ASYNCIO:
            journalFF = new AIOSequentialFileFactory(locationFile, BUFFER_SIZE, BUFFER_TIMEOUT, MAX_AIO, true, null, null);
            break;
         case MAPPED:
            journalFF = new MappedSequentialFileFactory(locationFile, FILE_SIZE, true, BUFFER_SIZE, BUFFER_TIMEOUT, null);
            break;
         default:
            throw new IllegalStateException("invalid journal type " + journalType);
      }

      journal = new JournalImpl(executorFactory, FILE_SIZE, MIN_FILES, POOL_SIZE, 0, 0, 30_000, journalFF, FILE_PREFIX, FILE_EXTENSION, MAX_AIO, 1, null, 10);
   }

}
