    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 *     https://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;
    026
    027import org.apache.commons.io.IOUtils;
    028import org.apache.commons.io.build.AbstractStreamBuilder;
    029
    030/**
    031 * {@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
    032 * using {@link BufferedInputStream}. Unfortunately, this is not something already available in JDK, {@code sun.nio.ch.ChannelInputStream} supports
    033 * reading a file using NIO, but does not support buffering.
    034 * <p>
    035 * To build an instance, use {@link Builder}.
    036 * </p>
    037 * <p>
    038 * This class was ported and adapted from Apache Spark commit 933dc6cb7b3de1d8ccaf73d124d6eb95b947ed19 where it was called {@code NioBufferedFileInputStream}.
    039 * </p>
    040 *
    041 * @see Builder
    042 * @since 2.9.0
    043 */
    044public final class BufferedFileChannelInputStream extends InputStream {
    045
    046    // @formatter:off
    047    /**
    048     * Builds a new {@link BufferedFileChannelInputStream}.
    049     *
    050     * <p>
    051     * Using File IO:
    052     * </p>
    053     * <pre>{@code
    054     * BufferedFileChannelInputStream s = BufferedFileChannelInputStream.builder()
    055     *   .setFile(file)
    056     *   .setBufferSize(4096)
    057     *   .get();}
    058     * </pre>
    059     * <p>
    060     * Using NIO Path:
    061     * </p>
    062     * <pre>{@code
    063     * BufferedFileChannelInputStream s = BufferedFileChannelInputStream.builder()
    064     *   .setPath(path)
    065     *   .setBufferSize(4096)
    066     *   .get();}
    067     * </pre>
    068     *
    069     * @see #get()
    070     * @since 2.12.0
    071     */
    072    // @formatter:on
    073    public static class Builder extends AbstractStreamBuilder<BufferedFileChannelInputStream, Builder> {
    074
    075        private FileChannel fileChannel;
    076
    077        /**
    078         * Constructs a new builder of {@link BufferedFileChannelInputStream}.
    079         */
    080        public Builder() {
    081            // empty
    082        }
    083
    084        /**
    085         * Builds a new {@link BufferedFileChannelInputStream}.
    086         * <p>
    087         * You must set an aspect that supports {@link #getInputStream()}, otherwise, this method throws an exception.
    088         * </p>
    089         * <p>
    090         * This builder uses the following aspects:
    091         * </p>
    092         * <ul>
    093         * <li>{@link FileChannel} takes precedence is set. </li>
    094         * <li>{@link #getPath()} if the file channel is not set.</li>
    095         * <li>{@link #getBufferSize()}</li>
    096         * </ul>
    097         *
    098         * @return a new instance.
    099         * @throws IllegalStateException         if the {@code origin} is {@code null}.
    100         * @throws UnsupportedOperationException if the origin cannot be converted to a {@link Path}.
    101         * @throws IOException                   if an I/O error occurs converting to an {@link Path} using {@link #getPath()}.
    102         * @see #getPath()
    103         * @see #getBufferSize()
    104         * @see #getUnchecked()
    105         */
    106        @Override
    107        public BufferedFileChannelInputStream get() throws IOException {
    108            return new BufferedFileChannelInputStream(this);
    109        }
    110
    111        /**
    112         * Sets the file channel.
    113         * <p>
    114         * This setting takes precedence over all others.
    115         * </p>
    116         *
    117         * @param fileChannel the file channel.
    118         * @return {@code this} instance.
    119         * @since 2.18.0
    120         */
    121        public Builder setFileChannel(final FileChannel fileChannel) {
    122            this.fileChannel = fileChannel;
    123            return this;
    124        }
    125
    126    }
    127
    128    /**
    129     * Constructs a new {@link Builder}.
    130     *
    131     * @return a new {@link Builder}.
    132     * @since 2.12.0
    133     */
    134    public static Builder builder() {
    135        return new Builder();
    136    }
    137
    138    private final ByteBuffer byteBuffer;
    139
    140    private final FileChannel fileChannel;
    141
    142    @SuppressWarnings("resource")
    143    private BufferedFileChannelInputStream(final Builder builder) throws IOException {
    144        this.fileChannel = builder.fileChannel != null ? builder.fileChannel : FileChannel.open(builder.getPath(), StandardOpenOption.READ);
    145        this.byteBuffer = ByteBuffer.allocateDirect(builder.getBufferSize());
    146        this.byteBuffer.flip();
    147    }
    148
    149    /**
    150     * Constructs a new instance for the given File.
    151     *
    152     * @param file The file to stream.
    153     * @throws IOException If an I/O error occurs.
    154     * @deprecated Use {@link #builder()}, {@link Builder}, and {@link Builder#get()}
    155     */
    156    @Deprecated
    157    public BufferedFileChannelInputStream(final File file) throws IOException {
    158        this(file, IOUtils.DEFAULT_BUFFER_SIZE);
    159    }
    160
    161    /**
    162     * Constructs a new instance for the given File and buffer size.
    163     *
    164     * @param file       The file to stream.
    165     * @param bufferSize buffer size.
    166     * @throws IOException If an I/O error occurs.
    167     * @deprecated Use {@link #builder()}, {@link Builder}, and {@link Builder#get()}
    168     */
    169    @Deprecated
    170    public BufferedFileChannelInputStream(final File file, final int bufferSize) throws IOException {
    171        this(file.toPath(), bufferSize);
    172    }
    173
    174    /**
    175     * Constructs a new instance for the given Path.
    176     *
    177     * @param path The path to stream.
    178     * @throws IOException If an I/O error occurs.
    179     * @deprecated Use {@link #builder()}, {@link Builder}, and {@link Builder#get()}
    180     */
    181    @Deprecated
    182    public BufferedFileChannelInputStream(final Path path) throws IOException {
    183        this(path, IOUtils.DEFAULT_BUFFER_SIZE);
    184    }
    185
    186    /**
    187     * Constructs a new instance for the given Path and buffer size.
    188     *
    189     * @param path       The path to stream.
    190     * @param bufferSize buffer size.
    191     * @throws IOException If an I/O error occurs.
    192     * @deprecated Use {@link #builder()}, {@link Builder}, and {@link Builder#get()}
    193     */
    194    @Deprecated
    195    public BufferedFileChannelInputStream(final Path path, final int bufferSize) throws IOException {
    196        this(builder().setPath(path).setBufferSize(bufferSize));
    197    }
    198
    199    @Override
    200    public synchronized int available() throws IOException {
    201        if (!fileChannel.isOpen() || !refill()) {
    202            return 0;
    203        }
    204        return byteBuffer.remaining();
    205    }
    206
    207    /**
    208     * 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
    209     * disposed buffer. However, neither the bytes allocated to direct buffers nor file descriptors opened for memory-mapped buffers put pressure on the garbage
    210     * 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
    211     * API to manually dispose of these kinds of buffers.
    212     *
    213     * @param buffer the buffer to clean.
    214     */
    215    private void clean(final ByteBuffer buffer) {
    216        if (buffer.isDirect()) {
    217            cleanDirectBuffer(buffer);
    218        }
    219    }
    220
    221    /**
    222     * 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
    223     * {@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
    224     * accessible even with reflection. However {@code sun.misc.Unsafe} added an {@code invokeCleaner()} method in JDK 9+ and this is still accessible with
    225     * reflection.
    226     *
    227     * @param buffer the buffer to clean. must be a DirectBuffer.
    228     */
    229    private void cleanDirectBuffer(final ByteBuffer buffer) {
    230        if (ByteBufferCleaner.isSupported()) {
    231            ByteBufferCleaner.clean(buffer);
    232        }
    233    }
    234
    235    @Override
    236    public synchronized void close() throws IOException {
    237        try {
    238            fileChannel.close();
    239        } finally {
    240            clean(byteBuffer);
    241        }
    242    }
    243
    244    @Override
    245    public synchronized int read() throws IOException {
    246        if (!refill()) {
    247            return EOF;
    248        }
    249        return byteBuffer.get() & 0xFF;
    250    }
    251
    252    @Override
    253    public synchronized int read(final byte[] b, final int offset, int len) throws IOException {
    254        IOUtils.checkFromIndexSize(b, offset, len);
    255        if (len == 0) {
    256            return 0;
    257        }
    258        if (!refill()) {
    259            return EOF;
    260        }
    261        len = Math.min(len, byteBuffer.remaining());
    262        byteBuffer.get(b, offset, len);
    263        return len;
    264    }
    265
    266    /**
    267     * Checks whether data is left to be read from the input stream.
    268     *
    269     * @return true if data is left, false otherwise.
    270     * @throws IOException if an I/O error occurs.
    271     */
    272    private boolean refill() throws IOException {
    273        Input.checkOpen(fileChannel.isOpen());
    274        if (!byteBuffer.hasRemaining()) {
    275            byteBuffer.clear();
    276            int nRead = 0;
    277            while (nRead == 0) {
    278                nRead = fileChannel.read(byteBuffer);
    279            }
    280            byteBuffer.flip();
    281            return nRead >= 0;
    282        }
    283        return true;
    284    }
    285
    286    @Override
    287    public synchronized long skip(final long n) throws IOException {
    288        if (n <= 0L) {
    289            return 0L;
    290        }
    291        if (byteBuffer.remaining() >= n) {
    292            // The buffered content is enough to skip
    293            byteBuffer.position(byteBuffer.position() + (int) n);
    294            return n;
    295        }
    296        final long skippedFromBuffer = byteBuffer.remaining();
    297        final long toSkipFromFileChannel = n - skippedFromBuffer;
    298        // Discard everything we have read in the buffer.
    299        byteBuffer.position(0);
    300        byteBuffer.flip();
    301        return skippedFromBuffer + skipFromFileChannel(toSkipFromFileChannel);
    302    }
    303
    304    private long skipFromFileChannel(final long n) throws IOException {
    305        final long currentFilePosition = fileChannel.position();
    306        final long size = fileChannel.size();
    307        if (n > size - currentFilePosition) {
    308            fileChannel.position(size);
    309            return size - currentFilePosition;
    310        }
    311        fileChannel.position(currentFilePosition + n);
    312        return n;
    313    }
    314
    315}