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() || !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}