/**
 * 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
 * <p>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * 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.journal;

import java.io.FileInputStream;
import java.nio.ByteBuffer;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
import org.apache.activemq.artemis.api.core.ActiveMQBuffers;
import org.apache.activemq.artemis.core.io.IOCallback;
import org.apache.activemq.artemis.core.io.SequentialFile;
import org.apache.activemq.artemis.core.io.SequentialFileFactory;
import org.apache.activemq.artemis.core.io.aio.AIOSequentialFileFactory;
import org.apache.activemq.artemis.nativo.jlibaio.LibaioContext;
import org.apache.activemq.artemis.tests.util.ActiveMQTestBase;
import org.apache.activemq.artemis.tests.util.Wait;
import org.apache.activemq.artemis.utils.ReusableLatch;
import org.jboss.logging.Logger;
import org.junit.Assert;
import org.junit.Assume;
import org.junit.Test;

public class AsyncOpenCloseTest extends ActiveMQTestBase {

   private static final Logger logger = Logger.getLogger(AsyncOpenCloseTest.class);

   @Test
   public void testCloseOnSubmit() throws Exception {
      Assume.assumeTrue(LibaioContext.isLoaded());
      AtomicInteger errors = new AtomicInteger(0);

      SequentialFileFactory factory = new AIOSequentialFileFactory(temporaryFolder.getRoot(), (Throwable error, String message, SequentialFile file) -> errors.incrementAndGet(), 4 * 1024);
      factory.start();

      SequentialFile file = factory.createSequentialFile("fileAIO.bin");
      file.open(1024, true);

      final int WRITES = 100;
      final int RECORD_SIZE = 4 * 1024;
      final int OPEN_TIMES = 25;

      file.fill(WRITES * RECORD_SIZE);

      ByteBuffer buffer = factory.newBuffer(RECORD_SIZE);
      ActiveMQBuffer buffer2 = ActiveMQBuffers.wrappedBuffer(buffer);

      try {

         file.close(true, false);
         AtomicInteger submit = new AtomicInteger(0);

         ReusableLatch valve = new ReusableLatch(0);

         byte writtenByte = (byte) 'a';
         for (int nclose = 0; nclose < OPEN_TIMES; nclose++) {
            logger.debug("************************************************** test " + nclose);
            writtenByte++;
            if (writtenByte >= (byte) 'z') {
               writtenByte = (byte) 'a';
            }
            buffer2.setIndex(0, 0);
            for (int s = 0; s < RECORD_SIZE; s++) {
               buffer2.writeByte(writtenByte);
            }
            file.open(1024, true);
            CyclicBarrier blocked = new CyclicBarrier(2);
            for (int i = 0; i < WRITES; i++) {
               if (i == 10) {
                  valve.countUp();
               }
               file.position(i * RECORD_SIZE);
               submit.incrementAndGet();
               buffer2.setIndex(0, RECORD_SIZE);
               file.write(buffer2, true, new IOCallback() {
                  @Override
                  public void done() {
                     try {
                        if (!valve.await(1, TimeUnit.MILLISECONDS)) {
                           logger.debug("blocking");
                           blocked.await();
                           valve.await(10, TimeUnit.SECONDS);
                           logger.debug("unblocking");
                        }
                     } catch (Exception e) {
                        e.printStackTrace();
                        errors.incrementAndGet();
                     }
                     submit.decrementAndGet();
                  }

                  @Override
                  public void onError(int errorCode, String errorMessage) {
                     errors.incrementAndGet();
                  }
               });
            }
            blocked.await();
            logger.debug("Closing");
            file.close(false, false);
            // even though the callback is blocked, the content of the file should already be good as written
            validateFile(file, (byte) writtenByte);
            valve.countDown();
            Wait.assertEquals(0, submit::get, 5000, 10);

         }
         Wait.assertEquals(0, submit::get);
      } finally {
         factory.releaseBuffer(buffer);
         factory.stop();
      }

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

   }

   private void validateFile(SequentialFile file, byte writtenByte) throws Exception {
      FileInputStream fileInputStream = new FileInputStream(file.getJavaFile());
      int size = (int)file.size();
      byte[] wholeFile = new byte[size];
      fileInputStream.read(wholeFile);
      for (int i = 0; i < wholeFile.length; i++) {
         Assert.assertEquals(writtenByte, (byte) wholeFile[i]);
      }
      fileInputStream.close();
   }

}
