/*
 * 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.core.io.nio;

import java.io.File;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.FileChannel;
import java.nio.file.StandardOpenOption;
import java.util.List;
import java.util.concurrent.Executor;

import io.netty.buffer.ByteBuf;
import org.apache.activemq.artemis.api.core.ActiveMQException;
import org.apache.activemq.artemis.api.core.ActiveMQExceptionType;
import org.apache.activemq.artemis.api.core.ActiveMQIOErrorException;
import org.apache.activemq.artemis.api.core.ActiveMQIllegalStateException;
import org.apache.activemq.artemis.core.io.AbstractSequentialFile;
import org.apache.activemq.artemis.core.io.DelegateCallback;
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.buffer.TimedBufferObserver;
import org.apache.activemq.artemis.journal.ActiveMQJournalBundle;
import org.apache.activemq.artemis.journal.ActiveMQJournalLogger;
import org.apache.activemq.artemis.utils.Env;

public class NIOSequentialFile extends AbstractSequentialFile {

   private FileChannel channel;

   private final int maxIO;

   public NIOSequentialFile(final SequentialFileFactory factory,
                            final File directory,
                            final String file,
                            final int maxIO,
                            final Executor writerExecutor) {
      super(directory, file, factory, writerExecutor);
      this.maxIO = maxIO;
   }

   @Override
   protected TimedBufferObserver createTimedBufferObserver() {
      return new SyncLocalBufferObserver();
   }

   @Override
   public int calculateBlockStart(final int position) {
      return position;
   }

   @Override
   public synchronized boolean isOpen() {
      return channel != null;
   }

   /**
    * this.maxIO represents the default maxIO.
    * Some operations while initializing files on the journal may require a different maxIO
    */
   @Override
   public synchronized void open() throws IOException {
      open(maxIO, true);
   }

   @Override
   public void open(final int maxIO, final boolean useExecutor) throws IOException {
      try {
         channel = FileChannel.open(getFile().toPath(), StandardOpenOption.CREATE, StandardOpenOption.WRITE, StandardOpenOption.READ);

         fileSize = channel.size();
      } catch (ClosedChannelException e) {
         throw e;
      } catch (IOException e) {
         factory.onIOError(new ActiveMQIOErrorException(e.getMessage(), e), e.getMessage(), this);
         throw e;
      }
   }

   @Override
   public void fill(final int size) throws IOException {
      try {
         //uses the most common OS page size to match the Page Cache entry size and reduce JVM memory footprint
         final int zeroPageCapacity = Env.osPageSize();
         final ByteBuffer zeroPage = this.factory.newBuffer(zeroPageCapacity);
         try {
            int bytesToWrite = size;
            long writePosition = 0;
            while (bytesToWrite > 0) {
               zeroPage.clear();
               final int zeroPageLimit = Math.min(bytesToWrite, zeroPageCapacity);
               zeroPage.limit(zeroPageLimit);
               //use the cheaper pwrite instead of fseek + fwrite
               final int writtenBytes = channel.write(zeroPage, writePosition);
               bytesToWrite -= writtenBytes;
               writePosition += writtenBytes;
            }
            if (factory.isDatasync()) {
               channel.force(true);
            }
            //set the position to 0 to match the fill contract
            channel.position(0);
            fileSize = channel.size();
         } finally {
            //return it to the factory
            this.factory.releaseBuffer(zeroPage);
         }
      } catch (ClosedChannelException e) {
         throw e;
      } catch (IOException e) {
         factory.onIOError(new ActiveMQIOErrorException(e.getMessage(), e), e.getMessage(), this);
         throw e;
      }
   }

   @Override
   public synchronized void close() throws IOException, InterruptedException, ActiveMQException {
      close(true);
   }

   @Override
   public synchronized void close(boolean waitSync) throws IOException, InterruptedException, ActiveMQException {
      super.close();

      try {
         if (channel != null) {
            if (waitSync && factory.isDatasync())
               channel.force(false);
            channel.close();
         }
      } catch (ClosedChannelException e) {
         throw e;
      } catch (IOException e) {
         factory.onIOError(new ActiveMQIOErrorException(e.getMessage(), e), e.getMessage(), this);
         throw e;
      }
      channel = null;

      notifyAll();
   }

   @Override
   public int read(final ByteBuffer bytes) throws Exception {
      return read(bytes, null);
   }

   @Override
   public synchronized int read(final ByteBuffer bytes,
                                final IOCallback callback) throws IOException, ActiveMQIllegalStateException {
      try {
         if (channel == null) {
            throw new ActiveMQIllegalStateException("File " + this.getFileName() + " has a null channel");
         }
         int bytesRead = channel.read(bytes);

         if (callback != null) {
            callback.done();
         }

         bytes.flip();

         return bytesRead;
      } catch (ClosedChannelException e) {
         throw e;
      } catch (IOException e) {
         if (callback != null) {
            callback.onError(ActiveMQExceptionType.IO_ERROR.getCode(), e.getLocalizedMessage());
         }

         factory.onIOError(new ActiveMQIOErrorException(e.getMessage(), e), e.getMessage(), this);

         throw e;
      }
   }

   @Override
   public void sync() throws IOException {
      if (factory.isDatasync() && channel != null) {
         try {
            channel.force(false);
         } catch (ClosedChannelException e) {
            throw e;
         } catch (IOException e) {
            factory.onIOError(new ActiveMQIOErrorException(e.getMessage(), e), e.getMessage(), this);
            throw e;
         }
      }
   }

   @Override
   public long size() throws IOException {
      if (channel == null) {
         return getFile().length();
      }

      try {
         return channel.size();
      } catch (ClosedChannelException e) {
         throw e;
      } catch (IOException e) {
         factory.onIOError(new ActiveMQIOErrorException(e.getMessage(), e), e.getMessage(), this);
         throw e;
      }
   }

   @Override
   public void position(final long pos) throws IOException {
      try {
         super.position(pos);
         channel.position(pos);
      } catch (ClosedChannelException e) {
         throw e;
      } catch (IOException e) {
         factory.onIOError(new ActiveMQIOErrorException(e.getMessage(), e), e.getMessage(), this);
         throw e;
      }
   }

   @Override
   public String toString() {
      return "NIOSequentialFile " + getFile();
   }

   @Override
   public SequentialFile cloneFile() {
      return new NIOSequentialFile(factory, directory, getFileName(), maxIO, null);
   }

   @Override
   public void writeDirect(final ByteBuffer bytes, final boolean sync, final IOCallback callback) {
      if (callback == null) {
         throw new NullPointerException("callback parameter need to be set");
      }

      try {
         internalWrite(bytes, sync, callback, true);
      } catch (Exception e) {
         callback.onError(ActiveMQExceptionType.GENERIC_EXCEPTION.getCode(), e.getMessage());
      }
   }

   @Override
   public void writeDirect(final ByteBuffer bytes, final boolean sync) throws Exception {
      internalWrite(bytes, sync, null, true);
   }

   @Override
   public void blockingWriteDirect(ByteBuffer bytes,boolean sync, boolean releaseBuffer) throws Exception {
      internalWrite(bytes, sync, null, releaseBuffer);
   }

   @Override
   protected ByteBuffer newBuffer(int size, final int limit) {
      // For NIO, we don't need to allocate a buffer the entire size of the timed buffer, unlike AIO

      size = limit;

      return super.newBuffer(size, limit);
   }

   private void internalWrite(final ByteBuffer bytes,
                              final boolean sync,
                              final IOCallback callback,
                              boolean releaseBuffer) throws IOException, ActiveMQIOErrorException, InterruptedException {
      if (!isOpen()) {
         if (callback != null) {
            callback.onError(ActiveMQExceptionType.IO_ERROR.getCode(), "File not opened");
         } else {
            throw ActiveMQJournalBundle.BUNDLE.fileNotOpened();
         }
         return;
      }

      position.addAndGet(bytes.limit());

      try {
         doInternalWrite(bytes, sync, callback, releaseBuffer);
      } catch (ClosedChannelException e) {
         throw e;
      } catch (IOException e) {
         factory.onIOError(new ActiveMQIOErrorException(e.getMessage(), e), e.getMessage(), this);
      }
   }

   /**
    * @param bytes
    * @param sync
    * @param callback
    * @throws IOException
    * @throws Exception
    */
   private void doInternalWrite(final ByteBuffer bytes,
                                final boolean sync,
                                final IOCallback callback,
                                boolean releaseBuffer) throws IOException {
      try {
         channel.write(bytes);

         if (sync) {
            sync();
         }

         if (callback != null) {
            callback.done();
         }
      } finally {
         if (releaseBuffer) {
            //release it to recycle the write buffer if big enough
            this.factory.releaseBuffer(bytes);
         }
      }
   }

   @Override
   public void copyTo(SequentialFile dstFile) throws IOException {
      if (ActiveMQJournalLogger.LOGGER.isDebugEnabled()) {
         ActiveMQJournalLogger.LOGGER.debug("Copying " + this + " as " + dstFile);
      }
      if (isOpen()) {
         throw new IllegalStateException("File opened!");
      }
      if (dstFile.isOpen()) {
         throw new IllegalArgumentException("dstFile must be closed too");
      }
      SequentialFile.appendTo(getFile().toPath(), dstFile.getJavaFile().toPath());
   }

   private class SyncLocalBufferObserver extends LocalBufferObserver {

      @Override
      public void flushBuffer(ByteBuf byteBuf, boolean requestedSync, List<IOCallback> callbacks) {
         //maybe no need to perform any copy
         final int bytes = byteBuf.readableBytes();
         if (bytes == 0) {
            IOCallback.done(callbacks);
         } else {
            //enable zero copy case
            if (byteBuf.nioBufferCount() == 1 && byteBuf.isDirect()) {
               final ByteBuffer buffer = byteBuf.internalNioBuffer(byteBuf.readerIndex(), bytes);
               final IOCallback callback = new DelegateCallback(callbacks);
               try {
                  //no need to pool the buffer and don't care if the NIO buffer got modified
                  internalWrite(buffer, requestedSync, callback, false);
               } catch (Exception e) {
                  callback.onError(ActiveMQExceptionType.GENERIC_EXCEPTION.getCode(), e.getMessage());
               }
            } else {
               super.flushBuffer(byteBuf, requestedSync, callbacks);
            }
         }
      }
   }
}
