Source code

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()) {
202            return 0;
203        }
204        if (!refill()) {
205            return 0;
206        }
207        return byteBuffer.remaining();
208    }
209
210    /**
211     * 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
212     * disposed buffer. However, neither the bytes allocated to direct buffers nor file descriptors opened for memory-mapped buffers put pressure on the garbage
213     * 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
214     * API to manually dispose of these kinds of buffers.
215     *
216     * @param buffer the buffer to clean.
217     */
218    private void clean(final ByteBuffer buffer) {
219        if (buffer.isDirect()) {
220            cleanDirectBuffer(buffer);
221        }
222    }
223
224    /**
225     * 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
226     * {@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
227     * accessible even with reflection. However {@code sun.misc.Unsafe} added an {@code invokeCleaner()} method in JDK 9+ and this is still accessible with
228     * reflection.
229     *
230     * @param buffer the buffer to clean. must be a DirectBuffer.
231     */
232    private void cleanDirectBuffer(final ByteBuffer buffer) {
233        if (ByteBufferCleaner.isSupported()) {
234            ByteBufferCleaner.clean(buffer);
235        }
236    }
237
238    @Override
239    public synchronized void close() throws IOException {
240        try {
241            fileChannel.close();
242        } finally {
243            clean(byteBuffer);
244        }
245    }
246
247    @Override
248    public synchronized int read() throws IOException {
249        if (!refill()) {
250            return EOF;
251        }
252        return byteBuffer.get() & 0xFF;
253    }
254
255    @Override
256    public synchronized int read(final byte[] b, final int offset, int len) throws IOException {
257        IOUtils.checkFromIndexSize(b, offset, len);
258        if (len == 0) {
259            return 0;
260        }
261        if (!refill()) {
262            return EOF;
263        }
264        len = Math.min(len, byteBuffer.remaining());
265        byteBuffer.get(b, offset, len);
266        return len;
267    }
268
269    /**
270     * Checks whether data is left to be read from the input stream.
271     *
272     * @return true if data is left, false otherwise
273     * @throws IOException if an I/O error occurs.
274     */
275    private boolean refill() throws IOException {
276        Input.checkOpen(fileChannel.isOpen());
277        if (!byteBuffer.hasRemaining()) {
278            byteBuffer.clear();
279            int nRead = 0;
280            while (nRead == 0) {
281                nRead = fileChannel.read(byteBuffer);
282            }
283            byteBuffer.flip();
284            return nRead >= 0;
285        }
286        return true;
287    }
288
289    @Override
290    public synchronized long skip(final long n) throws IOException {
291        if (n <= 0L) {
292            return 0L;
293        }
294        if (byteBuffer.remaining() >= n) {
295            // The buffered content is enough to skip
296            byteBuffer.position(byteBuffer.position() + (int) n);
297            return n;
298        }
299        final long skippedFromBuffer = byteBuffer.remaining();
300        final long toSkipFromFileChannel = n - skippedFromBuffer;
301        // Discard everything we have read in the buffer.
302        byteBuffer.position(0);
303        byteBuffer.flip();
304        return skippedFromBuffer + skipFromFileChannel(toSkipFromFileChannel);
305    }
306
307    private long skipFromFileChannel(final long n) throws IOException {
308        final long currentFilePosition = fileChannel.position();
309        final long size = fileChannel.size();
310        if (n > size - currentFilePosition) {
311            fileChannel.position(size);
312            return size - currentFilePosition;
313        }
314        fileChannel.position(currentFilePosition + n);
315        return n;
316    }
317
318}