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