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