alluxio-master-2.x/integration/fuse/src/main/java/alluxio/fuse/file/FuseFileInStream.java
/*
 * The Alluxio Open Foundation licenses this work under the Apache License, version 2.0
 * (the "License"). You may not use this work except in compliance with the License, which is
 * available at www.apache.org/licenses/LICENSE-2.0
 *
 * This software is distributed on an "AS IS" basis, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
 * either express or implied, as more fully set forth in the License.
 *
 * See the NOTICE file distributed with this work for information regarding copyright ownership.
 */

package alluxio.fuse.file;

import alluxio.AlluxioURI;
import alluxio.client.file.FileInStream;
import alluxio.client.file.FileSystem;
import alluxio.client.file.URIStatus;
import alluxio.concurrent.LockMode;
import alluxio.exception.AlluxioException;
import alluxio.exception.PreconditionMessage;
import alluxio.exception.runtime.AlluxioRuntimeException;
import alluxio.exception.runtime.FailedPreconditionRuntimeException;
import alluxio.exception.runtime.NotFoundRuntimeException;
import alluxio.exception.runtime.UnimplementedRuntimeException;
import alluxio.fuse.AlluxioFuseUtils;
import alluxio.fuse.lock.FuseReadWriteLockManager;
import alluxio.resource.CloseableResource;

import com.google.common.base.Preconditions;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Optional;
import java.util.concurrent.locks.Lock;
import javax.annotation.concurrent.ThreadSafe;

/**
 * An implementation for {@link FuseFileStream} for read only operations against an Alluxio uri.
 */
@ThreadSafe
public class FuseFileInStream implements FuseFileStream {
  private final FileInStream mInStream;
  private final FileStatus mFileStatus;
  private final AlluxioURI mURI;
  private final CloseableResource<Lock> mLockResource;
  private volatile boolean mClosed = false;

  /**
   * Creates a {@link FuseFileInStream}.
   *
   * @param fileSystem the file system
   * @param lockManager the lock manager
   * @param uri the alluxio uri
   * @return a {@link FuseFileInStream}
   */
  public static FuseFileInStream create(FileSystem fileSystem, FuseReadWriteLockManager lockManager,
      AlluxioURI uri) {
    Preconditions.checkNotNull(fileSystem);
    Preconditions.checkNotNull(uri);
    // Make sure file is not being written by current FUSE
    // deal with the async Fuse.release issue by waiting for write lock to be released
    CloseableResource<Lock> lockResource = lockManager.tryLock(uri.toString(), LockMode.READ);

    try {
      // Make sure file is not being written by other clients outside current FUSE
      Optional<URIStatus> status = AlluxioFuseUtils.getPathStatus(fileSystem, uri);
      if (status.isPresent() && !status.get().isCompleted()) {
        status = AlluxioFuseUtils.waitForFileCompleted(fileSystem, uri);
        if (!status.isPresent()) {
          throw new UnimplementedRuntimeException(String.format(
              "Failed to create fuse file in stream for %s: file is being written", uri));
        }
      }

      if (!status.isPresent()) {
        throw new NotFoundRuntimeException(String.format(
            "Failed to create read-only stream for %s: file does not exist", uri));
      }

      try {
        FileInStream is = fileSystem.openFile(uri);
        return new FuseFileInStream(is, lockResource,
            new FileStatus(status.get().getLength()), uri);
      } catch (IOException | AlluxioException e) {
        throw new RuntimeException(e);
      }
    } catch (Throwable t) {
      lockResource.close();
      throw t;
    }
  }

  private FuseFileInStream(FileInStream inStream, CloseableResource<Lock> lockResource,
      FileStatus fileStatus, AlluxioURI uri) {
    mInStream = Preconditions.checkNotNull(inStream);
    mLockResource = Preconditions.checkNotNull(lockResource);
    mFileStatus = Preconditions.checkNotNull(fileStatus);
    mURI = Preconditions.checkNotNull(uri);
  }

  @Override
  public synchronized int read(ByteBuffer buf, long size, long offset) {
    Preconditions.checkArgument(size >= 0 && offset >= 0 && size <= buf.capacity(),
        PreconditionMessage.ERR_BUFFER_STATE.toString(), buf.capacity(), offset, size);
    if (size == 0) {
      return 0;
    }
    if (offset >= mFileStatus.getFileLength()) {
      return 0;
    }
    final int sz = (int) size;
    int totalRead = 0;
    int currentRead = 0;
    try {
      mInStream.seek(offset);
      while (currentRead >= 0 && totalRead < sz) {
        currentRead = mInStream.read(buf, totalRead, sz - totalRead);
        if (currentRead > 0) {
          totalRead += currentRead;
        }
      }
    } catch (IOException e) {
      throw AlluxioRuntimeException.from(e);
    }
    return totalRead;
  }

  @Override
  public void write(ByteBuffer buf, long size, long offset) {
    throw new FailedPreconditionRuntimeException(String
        .format("Cannot write to read-only stream of path %s", mURI));
  }

  @Override
  public FileStatus getFileStatus() {
    return mFileStatus;
  }

  @Override
  public void flush() {}

  @Override
  public void truncate(long size) {
    throw new UnimplementedRuntimeException(String
        .format("Cannot truncate read-only stream of path %s", mURI));
  }

  @Override
  public synchronized void close() {
    if (mClosed) {
      return;
    }
    mClosed = true;
    try {
      mInStream.close();
    } catch (IOException e) {
      throw AlluxioRuntimeException.from(e);
    } finally {
      mLockResource.close();
    }
  }
}
