    001/*
    002 * Licensed under the Apache License, Version 2.0 (the "License");
    003 * you may not use this file except in compliance with the License.
    004 * You may obtain a copy of the License at
    005 *
    006 *     http://www.apache.org/licenses/LICENSE-2.0
    007 *
    008 * Unless required by applicable law or agreed to in writing, software
    009 * distributed under the License is distributed on an "AS IS" BASIS,
    010 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    011 * See the License for the specific language governing permissions and
    012 * limitations under the License.
    013 */
    014package org.apache.commons.io.input;
    015
    016import static org.apache.commons.io.IOUtils.EOF;
    017
    018import java.io.BufferedInputStream;
    019import java.io.File;
    020import java.io.IOException;
    021import java.io.InputStream;
    022import java.nio.ByteBuffer;
    023import java.nio.channels.FileChannel;
    024import java.nio.file.Path;
    025import java.nio.file.StandardOpenOption;
    026import java.util.Objects;
    027
    028import org.apache.commons.io.IOUtils;
    029import org.apache.commons.io.build.AbstractStreamBuilder;
    030
    031/**
    032 * {@link InputStream} implementation which uses direct buffer to read a file to avoid extra copy of data between Java and native memory which happens when
    033 * using {@link BufferedInputStream}. Unfortunately, this is not something already available in JDK, {@code sun.nio.ch.ChannelInputStream} supports
    034 * reading a file using NIO, but does not support buffering.
    035 * <p>
    036 * To build an instance, use {@link Builder}.
    037 * </p>
    038 * <p>
    039 * This class was ported and adapted from Apache Spark commit 933dc6cb7b3de1d8ccaf73d124d6eb95b947ed19 where it was called {@code NioBufferedFileInputStream}.
    040 * </p>
    041 *
    042 * @see Builder
    043 * @since 2.9.0
    044 */
    045public final class BufferedFileChannelInputStream extends InputStream {
    046
    047    // @formatter:off
    048    /**
    049     * Builds a new {@link BufferedFileChannelInputStream}.
    050     *
    051     * <p>
    052     * Using File IO:
    053     * </p>
    054     * <pre>{@code
    055     * BufferedFileChannelInputStream s = BufferedFileChannelInputStream.builder()
    056     *   .setFile(file)
    057     *   .setBufferSize(4096)
    058     *   .get();}
    059     * </pre>
    060     * <p>
    061     * Using NIO Path:
    062     * </p>
    063     * <pre>{@code
    064     * BufferedFileChannelInputStream s = BufferedFileChannelInputStream.builder()
    065     *   .setPath(path)
    066     *   .setBufferSize(4096)
    067     *   .get();}
    068     * </pre>
    069     *
    070     * @see #get()
    071     * @since 2.12.0
    072     */
    073    // @formatter:on
    074    public static class Builder extends AbstractStreamBuilder<BufferedFileChannelInputStream, Builder> {
    075
    076        private FileChannel fileChannel;
    077
    078        /**
    079         * Builds a new {@link BufferedFileChannelInputStream}.
    080         * <p>
    081         * You must set input that supports {@link #getInputStream()}, otherwise, this method throws an exception.
    082         * </p>
    083         * <p>
    084         * This builder use the following aspects:
    085         * </p>
    086         * <ul>
    087         * <li>{@link #getInputStream()}</li>
    088         * <li>{@link #getBufferSize()}</li>
    089         * </ul>
    090         *
    091         * @return a new instance.
    092         * @throws IllegalStateException         if the {@code origin} is {@code null}.
    093         * @throws UnsupportedOperationException if the origin cannot be converted to a {@link Path}.
    094         * @throws IOException If an I/O error occurs
    095         * @see #getPath()
    096         * @see #getBufferSize()
    097         */
    098        @Override
    099        public BufferedFileChannelInputStream get() throws IOException {
    100            return fileChannel != null ? new BufferedFileChannelInputStream(fileChannel, getBufferSize())
    101                    : new BufferedFileChannelInputStream(getPath(), getBufferSize());
    102        }
    103
    104        /**
    105         * Sets the file channel.
    106         * <p>
    107         * This setting takes precedence over all others.
    108         * </p>
    109         *
    110         * @param fileChannel the file channel.
    111         * @return this instance.
    112         * @since 2.18.0
    113         */
    114        public Builder setFileChannel(final FileChannel fileChannel) {
    115            this.fileChannel = fileChannel;
    116            return this;
    117        }
    118
    119    }
    120
    121    /**
    122     * Constructs a new {@link Builder}.
    123     *
    124     * @return a new {@link Builder}.
    125     * @since 2.12.0
    126     */
    127    public static Builder builder() {
    128        return new Builder();
    129    }
    130
    131    private final ByteBuffer byteBuffer;
    132
    133    private final FileChannel fileChannel;
    134
    135    /**
    136     * Constructs a new instance for the given File.
    137     *
    138     * @param file The file to stream.
    139     * @throws IOException If an I/O error occurs
    140     * @deprecated Use {@link #builder()}, {@link Builder}, and {@link Builder#get()}
    141     */
    142    @Deprecated
    143    public BufferedFileChannelInputStream(final File file) throws IOException {
    144        this(file, IOUtils.DEFAULT_BUFFER_SIZE);
    145    }
    146
    147    /**
    148     * Constructs a new instance for the given File and buffer size.
    149     *
    150     * @param file       The file to stream.
    151     * @param bufferSize buffer size.
    152     * @throws IOException If an I/O error occurs
    153     * @deprecated Use {@link #builder()}, {@link Builder}, and {@link Builder#get()}
    154     */
    155    @Deprecated
    156    public BufferedFileChannelInputStream(final File file, final int bufferSize) throws IOException {
    157        this(file.toPath(), bufferSize);
    158    }
    159
    160    private BufferedFileChannelInputStream(final FileChannel fileChannel, final int bufferSize) {
    161        this.fileChannel = Objects.requireNonNull(fileChannel, "path");
    162        byteBuffer = ByteBuffer.allocateDirect(bufferSize);
    163        byteBuffer.flip();
    164    }
    165
    166    /**
    167     * Constructs a new instance for the given Path.
    168     *
    169     * @param path The path to stream.
    170     * @throws IOException If an I/O error occurs
    171     * @deprecated Use {@link #builder()}, {@link Builder}, and {@link Builder#get()}
    172     */
    173    @Deprecated
    174    public BufferedFileChannelInputStream(final Path path) throws IOException {
    175        this(path, IOUtils.DEFAULT_BUFFER_SIZE);
    176    }
    177
    178    /**
    179     * Constructs a new instance for the given Path and buffer size.
    180     *
    181     * @param path       The path to stream.
    182     * @param bufferSize buffer size.
    183     * @throws IOException If an I/O error occurs
    184     * @deprecated Use {@link #builder()}, {@link Builder}, and {@link Builder#get()}
    185     */
    186    @SuppressWarnings("resource")
    187    @Deprecated
    188    public BufferedFileChannelInputStream(final Path path, final int bufferSize) throws IOException {
    189        this(FileChannel.open(path, StandardOpenOption.READ), bufferSize);
    190    }
    191
    192    @Override
    193    public synchronized int available() throws IOException {
    194        if (!fileChannel.isOpen()) {
    195            return 0;
    196        }
    197        if (!refill()) {
    198            return 0;
    199        }
    200        return byteBuffer.remaining();
    201    }
    202
    203    /**
    204     * Attempts to clean up a ByteBuffer if it is direct or memory-mapped. This uses an *unsafe* Sun API that will cause errors if one attempts to read from the
    205     * disposed buffer. However, neither the bytes allocated to direct buffers nor file descriptors opened for memory-mapped buffers put pressure on the garbage
    206     * collector. Waiting for garbage collection may lead to the depletion of off-heap memory or huge numbers of open files. There's unfortunately no standard
    207     * API to manually dispose of these kinds of buffers.
    208     *
    209     * @param buffer the buffer to clean.
    210     */
    211    private void clean(final ByteBuffer buffer) {
    212        if (buffer.isDirect()) {
    213            cleanDirectBuffer(buffer);
    214        }
    215    }
    216
    217    /**
    218     * In Java 8, the type of {@code sun.nio.ch.DirectBuffer.cleaner()} was {@code sun.misc.Cleaner}, and it was possible to access the method
    219     * {@code sun.misc.Cleaner.clean()} to invoke it. The type changed to {@code jdk.internal.ref.Cleaner} in later JDKs, and the {@code clean()} method is not
    220     * accessible even with reflection. However {@code sun.misc.Unsafe} added an {@code invokeCleaner()} method in JDK 9+ and this is still accessible with
    221     * reflection.
    222     *
    223     * @param buffer the buffer to clean. must be a DirectBuffer.
    224     */
    225    private void cleanDirectBuffer(final ByteBuffer buffer) {
    226        if (ByteBufferCleaner.isSupported()) {
    227            ByteBufferCleaner.clean(buffer);
    228        }
    229    }
    230
    231    @Override
    232    public synchronized void close() throws IOException {
    233        try {
    234            fileChannel.close();
    235        } finally {
    236            clean(byteBuffer);
    237        }
    238    }
    239
    240    @Override
    241    public synchronized int read() throws IOException {
    242        if (!refill()) {
    243            return EOF;
    244        }
    245        return byteBuffer.get() & 0xFF;
    246    }
    247
    248    @Override
    249    public synchronized int read(final byte[] b, final int offset, int len) throws IOException {
    250        if (offset < 0 || len < 0 || offset + len < 0 || offset + len > b.length) {
    251            throw new IndexOutOfBoundsException();
    252        }
    253        if (!refill()) {
    254            return EOF;
    255        }
    256        len = Math.min(len, byteBuffer.remaining());
    257        byteBuffer.get(b, offset, len);
    258        return len;
    259    }
    260
    261    /**
    262     * Checks whether data is left to be read from the input stream.
    263     *
    264     * @return true if data is left, false otherwise
    265     * @throws IOException if an I/O error occurs.
    266     */
    267    private boolean refill() throws IOException {
    268        Input.checkOpen(fileChannel.isOpen());
    269        if (!byteBuffer.hasRemaining()) {
    270            byteBuffer.clear();
    271            int nRead = 0;
    272            while (nRead == 0) {
    273                nRead = fileChannel.read(byteBuffer);
    274            }
    275            byteBuffer.flip();
    276            return nRead >= 0;
    277        }
    278        return true;
    279    }
    280
    281    @Override
    282    public synchronized long skip(final long n) throws IOException {
    283        if (n <= 0L) {
    284            return 0L;
    285        }
    286        if (byteBuffer.remaining() >= n) {
    287            // The buffered content is enough to skip
    288            byteBuffer.position(byteBuffer.position() + (int) n);
    289            return n;
    290        }
    291        final long skippedFromBuffer = byteBuffer.remaining();
    292        final long toSkipFromFileChannel = n - skippedFromBuffer;
    293        // Discard everything we have read in the buffer.
    294        byteBuffer.position(0);
    295        byteBuffer.flip();
    296        return skippedFromBuffer + skipFromFileChannel(toSkipFromFileChannel);
    297    }
    298
    299    private long skipFromFileChannel(final long n) throws IOException {
    300        final long currentFilePosition = fileChannel.position();
    301        final long size = fileChannel.size();
    302        if (n > size - currentFilePosition) {
    303            fileChannel.position(size);
    304            return size - currentFilePosition;
    305        }
    306        fileChannel.position(currentFilePosition + n);
    307        return n;
    308    }
    309
    310}