spark/io/GenericFileInputStreamSuite.java

0001 /*
0002  * Licensed to the Apache Software Foundation (ASF) under one or more
0003  * contributor license agreements.  See the NOTICE file distributed with
0004  * this work for additional information regarding copyright ownership.
0005  * The ASF licenses this file to You under the Apache License, Version 2.0
0006  * (the "License"); you may not use this file except in compliance with
0007  * the License.  You may obtain a copy of the License at
0008  *
0009  *    http://www.apache.org/licenses/LICENSE-2.0
0010  *
0011  * Unless required by applicable law or agreed to in writing, software
0012  * distributed under the License is distributed on an "AS IS" BASIS,
0013  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
0014  * See the License for the specific language governing permissions and
0015  * limitations under the License.
0016  */
0017 package org.apache.spark.io;
0018
0019 import org.apache.commons.io.FileUtils;
0020 import org.apache.commons.lang3.RandomUtils;
0021 import org.junit.After;
0022 import org.junit.Before;
0023 import org.junit.Test;
0024
0025 import java.io.File;
0026 import java.io.IOException;
0027 import java.io.InputStream;
0028
0029 import static org.junit.Assert.assertEquals;
0030
0031 /**
0032  * Tests functionality of {@link NioBufferedFileInputStream}
0033  */
0034 public abstract class GenericFileInputStreamSuite {
0035
0036   private byte[] randomBytes;
0037
0038   protected File inputFile;
0039
0040   protected InputStream[] inputStreams;
0041
0042   @Before
0043   public void setUp() throws IOException {
0044     // Create a byte array of size 2 MB with random bytes
0045     randomBytes =  RandomUtils.nextBytes(2 * 1024 * 1024);
0046     inputFile = File.createTempFile("temp-file", ".tmp");
0047     FileUtils.writeByteArrayToFile(inputFile, randomBytes);
0048   }
0049
0050   @After
0051   public void tearDown() throws IOException {
0052     inputFile.delete();
0053
0054     for (InputStream is : inputStreams) {
0055       is.close();
0056     }
0057   }
0058
0059   @Test
0060   public void testReadOneByte() throws IOException {
0061     for (InputStream inputStream: inputStreams) {
0062       for (int i = 0; i < randomBytes.length; i++) {
0063         assertEquals(randomBytes[i], (byte) inputStream.read());
0064       }
0065     }
0066   }
0067
0068   @Test
0069   public void testReadMultipleBytes() throws IOException {
0070     for (InputStream inputStream: inputStreams) {
0071       byte[] readBytes = new byte[8 * 1024];
0072       int i = 0;
0073       while (i < randomBytes.length) {
0074         int read = inputStream.read(readBytes, 0, 8 * 1024);
0075         for (int j = 0; j < read; j++) {
0076           assertEquals(randomBytes[i], readBytes[j]);
0077           i++;
0078         }
0079       }
0080     }
0081   }
0082
0083   @Test
0084   public void testBytesSkipped() throws IOException {
0085     for (InputStream inputStream: inputStreams) {
0086       assertEquals(1024, inputStream.skip(1024));
0087       for (int i = 1024; i < randomBytes.length; i++) {
0088         assertEquals(randomBytes[i], (byte) inputStream.read());
0089       }
0090     }
0091   }
0092
0093   @Test
0094   public void testBytesSkippedAfterRead() throws IOException {
0095     for (InputStream inputStream: inputStreams) {
0096       for (int i = 0; i < 1024; i++) {
0097         assertEquals(randomBytes[i], (byte) inputStream.read());
0098       }
0099       assertEquals(1024, inputStream.skip(1024));
0100       for (int i = 2048; i < randomBytes.length; i++) {
0101         assertEquals(randomBytes[i], (byte) inputStream.read());
0102       }
0103     }
0104   }
0105
0106   @Test
0107   public void testNegativeBytesSkippedAfterRead() throws IOException {
0108     for (InputStream inputStream: inputStreams) {
0109       for (int i = 0; i < 1024; i++) {
0110         assertEquals(randomBytes[i], (byte) inputStream.read());
0111       }
0112       // Skipping negative bytes should essential be a no-op
0113       assertEquals(0, inputStream.skip(-1));
0114       assertEquals(0, inputStream.skip(-1024));
0115       assertEquals(0, inputStream.skip(Long.MIN_VALUE));
0116       assertEquals(1024, inputStream.skip(1024));
0117       for (int i = 2048; i < randomBytes.length; i++) {
0118         assertEquals(randomBytes[i], (byte) inputStream.read());
0119       }
0120     }
0121   }
0122
0123   @Test
0124   public void testSkipFromFileChannel() throws IOException {
0125     for (InputStream inputStream: inputStreams) {
0126       // Since the buffer is smaller than the skipped bytes, this will guarantee
0127       // we skip from underlying file channel.
0128       assertEquals(1024, inputStream.skip(1024));
0129       for (int i = 1024; i < 2048; i++) {
0130         assertEquals(randomBytes[i], (byte) inputStream.read());
0131       }
0132       assertEquals(256, inputStream.skip(256));
0133       assertEquals(256, inputStream.skip(256));
0134       assertEquals(512, inputStream.skip(512));
0135       for (int i = 3072; i < randomBytes.length; i++) {
0136         assertEquals(randomBytes[i], (byte) inputStream.read());
0137       }
0138     }
0139   }
0140
0141   @Test
0142   public void testBytesSkippedAfterEOF() throws IOException {
0143     for (InputStream inputStream: inputStreams) {
0144       assertEquals(randomBytes.length, inputStream.skip(randomBytes.length + 1));
0145       assertEquals(-1, inputStream.read());
0146     }
0147   }
0148
0149   @Test
0150   public void testReadPastEOF() throws IOException {
0151     InputStream is = inputStreams[0];
0152     byte[] buf = new byte[1024];
0153     int read;
0154     while ((read = is.read(buf, 0, buf.length)) != -1);
0155
0156     int readAfterEOF = is.read(buf, 0, buf.length);
0157     assertEquals(-1, readAfterEOF);
0158   }
0159 }