alluxio-master-2.x/core/client/fs/src/test/java/alluxio/client/file/ufs/UfsFileInStreamTest.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.client.file.ufs;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertThrows;
import static org.junit.Assert.assertTrue;

import alluxio.AlluxioTestDirectory;
import alluxio.conf.Configuration;
import alluxio.exception.AlluxioException;
import alluxio.exception.runtime.AlluxioRuntimeException;
import alluxio.underfs.UfsFileStatus;
import alluxio.underfs.UnderFileSystem;
import alluxio.underfs.UnderFileSystemFactoryRegistry;
import alluxio.underfs.local.LocalUnderFileSystemFactory;
import alluxio.underfs.options.DeleteOptions;
import alluxio.underfs.options.OpenOptions;
import alluxio.util.io.BufferUtils;

import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.io.OutputStream;
import java.nio.ByteBuffer;
import java.nio.file.Paths;
import java.util.Random;
import java.util.UUID;

/**
 * Add unit tests for {@link UfsFileInStream}.
 */
public class UfsFileInStreamTest {
  private static final int CHUNK_SIZE = 100;
  private String mRootUfs;
  private UnderFileSystem mUfs;

  /**
   * Sets up the file system and the context before a test runs.
   */
  @Before
  public void before() {
    mRootUfs = AlluxioTestDirectory.createTemporaryDirectory("ufs").toString();
    UnderFileSystemFactoryRegistry.register(new LocalUnderFileSystemFactory());
    mUfs = UnderFileSystem.Factory.create(mRootUfs, Configuration.global());
  }

  @After
  public void after() throws IOException, AlluxioException {
    mUfs.deleteDirectory(mRootUfs, DeleteOptions.defaults().setRecursive(true));
  }

  @Test
  public void readWithNullUfsStream() {
    assertThrows(NullPointerException.class,
        () -> new UfsFileInStream(null, 0L).close());
  }

  @Test
  public void openClose() throws IOException {
    String ufsPath = getUfsPath();
    createFile(ufsPath, 0);
    getStream(ufsPath).close();
  }

  @Test
  public void singleByteRead() throws IOException {
    String ufsPath = getUfsPath();
    createFile(ufsPath, 1);
    try (UfsFileInStream inStream = getStream(ufsPath)) {
      assertEquals(0, inStream.read());
    }
  }

  @Test
  public void twoBytesRead() throws IOException {
    String ufsPath = getUfsPath();
    createFile(ufsPath, 2);
    try (UfsFileInStream inStream = getStream(ufsPath)) {
      assertEquals(0, inStream.read());
      assertEquals(1, inStream.read());
    }
  }

  @Test
  public void manyBytesRead() throws IOException {
    String ufsPath = getUfsPath();
    createFile(ufsPath, CHUNK_SIZE);
    try (UfsFileInStream inStream = getStream(ufsPath)) {
      byte[] res = new byte[CHUNK_SIZE];
      assertEquals(CHUNK_SIZE, inStream.read(res));
      assertTrue(BufferUtils.equalIncreasingByteArray(CHUNK_SIZE, res));
    }
  }

  @Test
  public void manyBytesReadByteBuffer() throws IOException {
    String ufsPath = getUfsPath();
    createFile(ufsPath, CHUNK_SIZE);
    ByteBuffer buffer = ByteBuffer.allocate(CHUNK_SIZE);
    try (UfsFileInStream inStream = getStream(ufsPath)) {
      assertEquals(CHUNK_SIZE, inStream.read(buffer));
      assertTrue(BufferUtils.equalIncreasingByteBuffer(0, CHUNK_SIZE, buffer));
    }
  }

  @Test
  public void readAll() throws IOException {
    int len = CHUNK_SIZE * 5;
    int start = 0;
    String ufsPath = getUfsPath();
    createFile(ufsPath, CHUNK_SIZE * 5);
    byte[] res = new byte[CHUNK_SIZE];
    try (UfsFileInStream inStream = getStream(ufsPath)) {
      while (start < len) {
        assertEquals(CHUNK_SIZE, inStream.read(res));
        assertTrue(BufferUtils.equalIncreasingByteArray(start, CHUNK_SIZE, res));
        start += CHUNK_SIZE;
      }
    }
  }

  @Test
  public void readAllByteBuffer() throws IOException {
    int len = CHUNK_SIZE * 5;
    int start = 0;
    String ufsPath = getUfsPath();
    createFile(ufsPath, CHUNK_SIZE * 5);
    ByteBuffer buffer = ByteBuffer.allocate(CHUNK_SIZE);
    try (UfsFileInStream inStream = getStream(ufsPath)) {
      while (start < len) {
        assertEquals(CHUNK_SIZE, inStream.read(buffer));
        assertTrue(BufferUtils.equalIncreasingByteBuffer(start, CHUNK_SIZE, buffer));
        start += CHUNK_SIZE;
        buffer.clear();
      }
    }
  }

  @Test
  public void readOffset() throws IOException {
    String ufsPath = getUfsPath();
    createFile(ufsPath, CHUNK_SIZE);
    int start = CHUNK_SIZE / 4;
    int len = CHUNK_SIZE / 2;
    try (UfsFileInStream inStream = getStream(ufsPath)) {
      byte[] res = new byte[CHUNK_SIZE];
      assertEquals(CHUNK_SIZE / 2, inStream.read(res, start, len));
      for (int i = start; i < start + len; i++) {
        assertEquals(i - start, res[i]);
      }
    }
  }

  @Test
  public void readOffsetByteBuffer() throws IOException {
    String ufsPath = getUfsPath();
    createFile(ufsPath, CHUNK_SIZE);
    int start = CHUNK_SIZE / 4;
    int len = CHUNK_SIZE / 2;
    try (UfsFileInStream inStream = getStream(ufsPath)) {
      ByteBuffer buffer = ByteBuffer.allocate(CHUNK_SIZE);
      assertEquals(CHUNK_SIZE / 2, inStream.read(buffer, start, len));
      for (int i = start; i < start + len; i++) {
        assertEquals(i - start, buffer.get(i));
      }
    }
  }

  @Test
  public void readOutOfBound() throws IOException {
    String ufsPath = getUfsPath();
    createFile(ufsPath, CHUNK_SIZE);
    try (UfsFileInStream inStream = getStream(ufsPath)) {
      byte[] res = new byte[CHUNK_SIZE * 2];
      assertEquals(CHUNK_SIZE, inStream.read(res));
      assertTrue(BufferUtils.matchIncreasingByteArray(0, CHUNK_SIZE, res));
      assertEquals(-1, inStream.read(res));
    }
  }

  @Test
  public void readOutOfBoundByteBuffer() throws IOException {
    String ufsPath = getUfsPath();
    createFile(ufsPath, CHUNK_SIZE);
    ByteBuffer buffer = ByteBuffer.allocate(CHUNK_SIZE * 2);
    try (UfsFileInStream inStream = getStream(ufsPath)) {
      assertEquals(CHUNK_SIZE, inStream.read(buffer));
      assertTrue(BufferUtils.matchIncreasingByteBuffer(0, CHUNK_SIZE, buffer));
      assertEquals(-1, inStream.read(buffer));
    }
  }

  @Test
  public void readOverflowOffLen() throws IOException {
    String ufsPath = getUfsPath();
    createFile(ufsPath, CHUNK_SIZE);
    try (UfsFileInStream inStream = getStream(ufsPath)) {
      assertThrows(IllegalArgumentException.class,
          () -> inStream.read(new byte[CHUNK_SIZE], 0, CHUNK_SIZE * 2));
    }
  }

  @Test
  public void readOverflowOffLenByteBuffer() throws IOException {
    String ufsPath = getUfsPath();
    createFile(ufsPath, CHUNK_SIZE);
    try (UfsFileInStream inStream = getStream(ufsPath)) {
      assertThrows(IllegalArgumentException.class,
          () -> inStream.read(ByteBuffer.allocate(CHUNK_SIZE), 0, CHUNK_SIZE * 2));
    }
  }

  @Test
  public void readNullArray() throws IOException {
    String ufsPath = getUfsPath();
    createFile(ufsPath, CHUNK_SIZE);
    try (UfsFileInStream inStream = getStream(ufsPath)) {
      assertThrows(NullPointerException.class,
          () -> inStream.read((byte[]) null));
    }
  }

  @Test
  public void readNullBuffer() throws IOException {
    String ufsPath = getUfsPath();
    createFile(ufsPath, CHUNK_SIZE);
    try (UfsFileInStream inStream = getStream(ufsPath)) {
      assertThrows(NullPointerException.class,
          () -> inStream.read((ByteBuffer) null, 0, CHUNK_SIZE));
    }
  }

  @Test
  public void readNullArrayOffset() throws IOException {
    String ufsPath = getUfsPath();
    createFile(ufsPath, CHUNK_SIZE);
    try (UfsFileInStream inStream = getStream(ufsPath)) {
      assertThrows(NullPointerException.class,
          () -> inStream.read((byte[]) null, 0, CHUNK_SIZE));
    }
  }

  @Test
  public void readNullBufferOffset() throws IOException {
    String ufsPath = getUfsPath();
    createFile(ufsPath, CHUNK_SIZE);
    try (UfsFileInStream inStream = getStream(ufsPath)) {
      assertThrows(NullPointerException.class,
          () -> inStream.read((ByteBuffer) null, 0, CHUNK_SIZE));
    }
  }

  @Test
  public void positionedRead() throws IOException {
    String ufsPath = getUfsPath();
    createFile(ufsPath, CHUNK_SIZE);
    try (UfsFileInStream inStream = getStream(ufsPath)) {
      byte[] res = new byte[CHUNK_SIZE / 2];
      assertEquals(CHUNK_SIZE / 2,
          inStream.positionedRead(CHUNK_SIZE / 2, res, 0, CHUNK_SIZE / 2));
      assertTrue(BufferUtils.equalIncreasingByteArray(CHUNK_SIZE / 2, CHUNK_SIZE / 2, res));
    }
  }

  @Test
  public void positionedReadMulti() throws IOException {
    String ufsPath = getUfsPath();
    createFile(ufsPath, CHUNK_SIZE);
    Random random = new Random();
    try (UfsFileInStream inStream = getStream(ufsPath)) {
      for (int i = 0; i < 10; i++) {
        int pos = random.nextInt(CHUNK_SIZE);
        int len = CHUNK_SIZE - pos;
        byte[] res = new byte[len];
        assertEquals(len,
            inStream.positionedRead(pos, res, 0, len));
        assertTrue(BufferUtils.equalIncreasingByteArray(pos, len, res));
      }
    }
  }

  @Test
  public void seekForward() throws IOException {
    String ufsPath = getUfsPath();
    createFile(ufsPath, CHUNK_SIZE);
    Random random = new Random();
    int pos = 0;
    try (UfsFileInStream inStream = getStream(ufsPath)) {
      for (int i = 0; i < 10; i++) {
        pos += random.nextInt(CHUNK_SIZE - pos);
        inStream.seek(pos);
        assertEquals(pos, inStream.getPos());
        int len = CHUNK_SIZE - pos;
        byte[] res = new byte[len];
        assertEquals(len,
            inStream.read(res, 0, len));
        assertTrue(BufferUtils.equalIncreasingByteArray(pos, len, res));
        if (CHUNK_SIZE == pos) {
          break;
        }
      }
    }
  }

  @Test
  public void seekBackward() throws IOException {
    String ufsPath = getUfsPath();
    createFile(ufsPath, CHUNK_SIZE);
    Random random = new Random();
    int pos = CHUNK_SIZE - 1;
    try (UfsFileInStream inStream = getStream(ufsPath)) {
      for (int i = 0; i < 10; i++) {
        pos -= random.nextInt(pos);
        inStream.seek(pos);
        assertEquals(pos, inStream.getPos());
        int len = CHUNK_SIZE - pos;
        byte[] res = new byte[len];
        assertEquals(len,
            inStream.read(res, 0, len));
        assertTrue(BufferUtils.equalIncreasingByteArray(pos, len, res));
        if (pos <= 0) {
          break;
        }
      }
    }
  }

  @Test
  public void seekToBeginning() throws IOException {
    String ufsPath = getUfsPath();
    createFile(ufsPath, CHUNK_SIZE);
    try (UfsFileInStream inStream = getStream(ufsPath)) {
      byte[] res = new byte[CHUNK_SIZE];
      assertEquals(CHUNK_SIZE, inStream.read(res));
      assertTrue(BufferUtils.equalIncreasingByteArray(CHUNK_SIZE, res));
      inStream.seek(0);
      assertEquals(0, inStream.getPos());
      assertEquals(CHUNK_SIZE, inStream.read(res));
      assertTrue(BufferUtils.equalIncreasingByteArray(CHUNK_SIZE, res));
    }
  }

  @Test
  public void seekForwardAndBackward() throws IOException {
    String ufsPath = getUfsPath();
    createFile(ufsPath, CHUNK_SIZE);
    Random random = new Random();
    try (UfsFileInStream inStream = getStream(ufsPath)) {
      for (int i = 0; i < 10; i++) {
        int pos = random.nextInt(CHUNK_SIZE);
        inStream.seek(pos);
        assertEquals(pos, inStream.getPos());
        int len = CHUNK_SIZE - pos;
        byte[] res = new byte[len];
        assertEquals(len,
            inStream.read(res, 0, len));
        assertTrue(BufferUtils.equalIncreasingByteArray(pos, len, res));
      }
    }
  }

  @Test
  public void seekPassEnd() throws IOException {
    String ufsPath = getUfsPath();
    createFile(ufsPath, CHUNK_SIZE);
    try (UfsFileInStream inStream = getStream(ufsPath)) {
      assertThrows(IllegalArgumentException.class, () -> inStream.seek(CHUNK_SIZE + 1));
    }
  }

  @Test
  public void seekNegative() throws IOException {
    String ufsPath = getUfsPath();
    createFile(ufsPath, CHUNK_SIZE);
    try (UfsFileInStream inStream = getStream(ufsPath)) {
      assertThrows(IllegalArgumentException.class, () -> inStream.seek(-1));
    }
  }

  @Test
  public void skip() throws IOException {
    String ufsPath = getUfsPath();
    createFile(ufsPath, CHUNK_SIZE);
    Random random = new Random();
    try (UfsFileInStream inStream = getStream(ufsPath)) {
      for (int i = 0; i < 10; i++) {
        if (inStream.remaining() <= 0) {
          break;
        }
        int skip = random.nextInt((int) inStream.remaining());
        assertEquals(skip, inStream.skip(skip));
        assertEquals(skip, inStream.getPos());
        int len = CHUNK_SIZE - skip;
        byte[] res = new byte[len];
        assertEquals(len,
            inStream.read(res, 0, len));
        assertTrue(BufferUtils.equalIncreasingByteArray(skip, len, res));
      }
    }
  }

  @Test
  public void skipToEnd() throws IOException {
    String ufsPath = getUfsPath();
    createFile(ufsPath, CHUNK_SIZE);
    try (UfsFileInStream inStream = getStream(ufsPath)) {
      assertEquals(CHUNK_SIZE, inStream.skip(CHUNK_SIZE));
      assertEquals(-1, inStream.read());
    }
  }

  @Test
  public void skipPassEnd() throws IOException {
    String ufsPath = getUfsPath();
    createFile(ufsPath, CHUNK_SIZE);
    try (UfsFileInStream inStream = getStream(ufsPath)) {
      assertEquals(CHUNK_SIZE, inStream.skip(CHUNK_SIZE + 1));
      Assert.assertEquals(-1, inStream.read());
    }
  }

  @Test
  public void skipNegative() throws IOException {
    String ufsPath = getUfsPath();
    createFile(ufsPath, CHUNK_SIZE);
    try (UfsFileInStream inStream = getStream(ufsPath)) {
      assertEquals(0, inStream.skip(-1));
      Assert.assertEquals(0, inStream.read());
    }
  }

  @Test
  public void getPosition() throws IOException {
    String ufsPath = getUfsPath();
    createFile(ufsPath, CHUNK_SIZE);
    try (UfsFileInStream inStream = getStream(ufsPath)) {
      assertEquals(0, inStream.getPos());
      inStream.read();
      assertEquals(1, inStream.getPos());
      int len = CHUNK_SIZE / 2;
      inStream.read(new byte[len], 0, len);
      assertEquals(1 + len, inStream.getPos());
      len = CHUNK_SIZE / 4;
      inStream.read(ByteBuffer.allocate(len), 0, len);
      assertEquals(1 + CHUNK_SIZE / 4 * 3, inStream.getPos());
    }
  }

  @Test
  public void remaining() throws IOException {
    String ufsPath = getUfsPath();
    createFile(ufsPath, CHUNK_SIZE);
    try (UfsFileInStream inStream = getStream(ufsPath)) {
      assertEquals(CHUNK_SIZE, inStream.remaining());
      inStream.read();
      assertEquals(CHUNK_SIZE - 1, inStream.remaining());
      int len = CHUNK_SIZE / 2;
      inStream.read(new byte[len], 0, len);
      assertEquals(CHUNK_SIZE - len - 1, inStream.remaining());
      len = CHUNK_SIZE / 4;
      inStream.read(ByteBuffer.allocate(len), 0, len);
      assertEquals(CHUNK_SIZE / 4 - 1, inStream.remaining());
    }
  }

  private String getUfsPath() {
    return Paths.get(mRootUfs, String.valueOf(UUID.randomUUID())).toString();
  }

  private UfsFileInStream getStream(String ufsPath) throws IOException {
    return new UfsFileInStream(offset -> {
      try {
        return mUfs.open(ufsPath, OpenOptions.defaults().setOffset(offset));
      } catch (IOException e) {
        throw AlluxioRuntimeException.from(e);
      }
    }, ((UfsFileStatus) mUfs.getStatus(ufsPath)).getContentLength());
  }

  private void createFile(String ufsPath, int len) throws IOException {
    createFile(ufsPath, 0, len);
  }

  private void createFile(String ufsPath, int start, int len) throws IOException {
    try (OutputStream outStream = mUfs.create(ufsPath)) {
      outStream.write(BufferUtils.getIncreasingByteArray(start, len));
    }
  }
}
