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


import java.io.EOFException;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SeekableByteChannel;
import java.nio.channels.ClosedChannelException; // javadoc @link
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.StandardOpenOption;
import java.util.concurrent.Future;

A straightforward implementation of FSDirectory using Files.newByteChannel(Path, OpenOption...). However, this class has poor concurrent performance (multiple threads will bottleneck) as it synchronizes when multiple threads read from the same file. It's usually better to use NIOFSDirectory or MMapDirectory instead. 

NOTE: Accessing this class either directly or indirectly from a thread while it's interrupted can close the underlying file descriptor immediately if at the same time the thread is blocked on IO. The file descriptor will remain closed and subsequent access to SimpleFSDirectory will throw a ClosedChannelException. If your application uses either Thread.interrupt() or Future.cancel(boolean) you should use the legacy RAFDirectory from the Lucene misc module in favor of SimpleFSDirectory. 
/** A straightforward implementation of {@link FSDirectory}
 *  using {@link Files#newByteChannel(Path, java.nio.file.OpenOption...)}.  
 *  However, this class has
 *  poor concurrent performance (multiple threads will
 *  bottleneck) as it synchronizes when multiple threads
 *  read from the same file.  It's usually better to use
 *  {@link NIOFSDirectory} or {@link MMapDirectory} instead.
 * <p>
 * <b>NOTE:</b> Accessing this class either directly or
 * indirectly from a thread while it's interrupted can close the
 * underlying file descriptor immediately if at the same time the thread is
 * blocked on IO. The file descriptor will remain closed and subsequent access
 * to {@link SimpleFSDirectory} will throw a {@link ClosedChannelException}. If
 * your application uses either {@link Thread#interrupt()} or
 * {@link Future#cancel(boolean)} you should use the legacy {@code RAFDirectory}
 * from the Lucene {@code misc} module in favor of {@link SimpleFSDirectory}.
 * </p>
 */
public class SimpleFSDirectory extends FSDirectory {
    
  
Create a new SimpleFSDirectory for the named location.
 The directory is created at the named location if it does not yet exist.
Params:
path – the path of the directory
lockFactory – the lock factory to use
Throws:
IOException – if there is a low-level I/O error/** Create a new SimpleFSDirectory for the named location.
   *  The directory is created at the named location if it does not yet exist.
   *
   * @param path the path of the directory
   * @param lockFactory the lock factory to use
   * @throws IOException if there is a low-level I/O error
   */
  public SimpleFSDirectory(Path path, LockFactory lockFactory) throws IOException {
    super(path, lockFactory);
  }
  
  
Create a new SimpleFSDirectory for the named location and FSLockFactory.getDefault(). The directory is created at the named location if it does not yet exist. 
Params:
path – the path of the directory
Throws:
IOException – if there is a low-level I/O error/** Create a new SimpleFSDirectory for the named location and {@link FSLockFactory#getDefault()}.
   *  The directory is created at the named location if it does not yet exist.
   *
   * @param path the path of the directory
   * @throws IOException if there is a low-level I/O error
   */
  public SimpleFSDirectory(Path path) throws IOException {
    this(path, FSLockFactory.getDefault());
  }

  
Creates an IndexInput for the file with the given name. /** Creates an IndexInput for the file with the given name. */
  @Override
  public IndexInput openInput(String name, IOContext context) throws IOException {
    ensureOpen();
    ensureCanRead(name);
    Path path = directory.resolve(name);
    SeekableByteChannel channel = Files.newByteChannel(path, StandardOpenOption.READ);
    return new SimpleFSIndexInput("SimpleFSIndexInput(path=\"" + path + "\")", channel, context);
  }

  
Reads bytes with SeekableByteChannel.read(ByteBuffer) /**
   * Reads bytes with {@link SeekableByteChannel#read(ByteBuffer)}
   */
  static final class SimpleFSIndexInput extends BufferedIndexInput {
    
The maximum chunk size for reads of 16384 bytes.
/**
     * The maximum chunk size for reads of 16384 bytes.
     */
    private static final int CHUNK_SIZE = 16384;
    
    
the channel we will read from /** the channel we will read from */
    protected final SeekableByteChannel channel;
    
is this instance a clone and hence does not own the file to close it /** is this instance a clone and hence does not own the file to close it */
    boolean isClone = false;
    
start offset: non-zero in the slice case /** start offset: non-zero in the slice case */
    protected final long off;
    
end offset (start+length) /** end offset (start+length) */
    protected final long end;
    
    private ByteBuffer byteBuf; // wraps the buffer for NIO

    public SimpleFSIndexInput(String resourceDesc, SeekableByteChannel channel, IOContext context) throws IOException {
      super(resourceDesc, context);
      this.channel = channel; 
      this.off = 0L;
      this.end = channel.size();
    }
    
    public SimpleFSIndexInput(String resourceDesc, SeekableByteChannel channel, long off, long length, int bufferSize) {
      super(resourceDesc, bufferSize);
      this.channel = channel;
      this.off = off;
      this.end = off + length;
      this.isClone = true;
    }
    
    @Override
    public void close() throws IOException {
      if (!isClone) {
        channel.close();
      }
    }
    
    @Override
    public SimpleFSIndexInput clone() {
      SimpleFSIndexInput clone = (SimpleFSIndexInput)super.clone();
      clone.isClone = true;
      return clone;
    }
    
    @Override
    public IndexInput slice(String sliceDescription, long offset, long length) throws IOException {
      if (offset < 0 || length < 0 || offset + length > this.length()) {
        throw new IllegalArgumentException("slice() " + sliceDescription + " out of bounds: offset=" + offset + ",length=" + length + ",fileLength="  + this.length() + ": "  + this);
      }
      return new SimpleFSIndexInput(getFullSliceDescription(sliceDescription), channel, off + offset, length, getBufferSize());
    }

    @Override
    public final long length() {
      return end - off;
    }

    @Override
    protected void newBuffer(byte[] newBuffer) {
      super.newBuffer(newBuffer);
      byteBuf = ByteBuffer.wrap(newBuffer);
    }

    @Override
    protected void readInternal(byte[] b, int offset, int len) throws IOException {
      final ByteBuffer bb;

      // Determine the ByteBuffer we should use
      if (b == buffer) {
        // Use our own pre-wrapped byteBuf:
        assert byteBuf != null;
        bb = byteBuf;
        byteBuf.clear().position(offset);
      } else {
        bb = ByteBuffer.wrap(b, offset, len);
      }

      synchronized(channel) {
        long pos = getFilePointer() + off;
        
        if (pos + len > end) {
          throw new EOFException("read past EOF: " + this);
        }
               
        try {
          channel.position(pos);

          int readLength = len;
          while (readLength > 0) {
            final int toRead = Math.min(CHUNK_SIZE, readLength);
            bb.limit(bb.position() + toRead);
            assert bb.remaining() == toRead;
            final int i = channel.read(bb);
            if (i < 0) { // be defensive here, even though we checked before hand, something could have changed
              throw new EOFException("read past EOF: " + this + " off: " + offset + " len: " + len + " pos: " + pos + " chunkLen: " + toRead + " end: " + end);
            }
            assert i > 0 : "SeekableByteChannel.read with non zero-length bb.remaining() must always read at least one byte (Channel is in blocking mode, see spec of ReadableByteChannel)";
            pos += i;
            readLength -= i;
          }
          assert readLength == 0;
        } catch (IOException ioe) {
          throw new IOException(ioe.getMessage() + ": " + this, ioe);
        }
      }
    }

    @Override
    protected void seekInternal(long pos) throws IOException {
      if (pos > length()) {
        throw new EOFException("read past EOF: pos=" + pos + " vs length=" + length() + ": " + this);
      }
    }
  }
}