001/*
002 * Licensed to the Apache Software Foundation (ASF) under one or more
003 * contributor license agreements.  See the NOTICE file distributed with
004 * this work for additional information regarding copyright ownership.
005 * The ASF licenses this file to You under the Apache License, Version 2.0
006 * (the "License"); you may not use this file except in compliance with
007 * the License.  You may obtain a copy of the License at
008 *
009 *      http://www.apache.org/licenses/LICENSE-2.0
010 *
011 * Unless required by applicable law or agreed to in writing, software
012 * distributed under the License is distributed on an "AS IS" BASIS,
013 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
014 * See the License for the specific language governing permissions and
015 * limitations under the License.
016 */
017package org.apache.commons.io.input;
018
019import static org.apache.commons.io.IOUtils.EOF;
020
021import java.io.IOException;
022import java.io.InputStream;
023
024import org.apache.commons.io.IOUtils;
025import org.apache.commons.io.build.AbstractStreamBuilder;
026
027//@formatter:off
028/**
029 * Reads bytes up to a maximum count and stops once reached.
030 * <p>
031 * To build an instance, see {@link AbstractBuilder}.
032 * </p>
033 * <p>
034 * By default, a {@link BoundedInputStream} is <em>unbound</em>; so make sure to call {@link AbstractBuilder#setMaxCount(long)}.
035 * </p>
036 * <p>
037 * You can find out how many bytes this stream has seen so far by calling {@link BoundedInputStream#getCount()}. This value reflects bytes read and skipped.
038 * </p>
039 * <h2>Using a ServletInputStream</h2>
040 * <p>
041 * A {@code ServletInputStream} can block if you try to read content that isn't there
042 * because it doesn't know whether the content hasn't arrived yet or whether the content has finished. Initialize an {@link BoundedInputStream} with the
043 * {@code Content-Length} sent in the {@code ServletInputStream}'s header, this stop it from blocking, providing it's been sent with a correct content
044 * length in the first place.
045 * </p>
046 * <h2>Using NIO</h2>
047 * <pre>{@code
048 * BoundedInputStream s = BoundedInputStream.builder()
049 *   .setPath(Paths.get("MyFile.xml"))
050 *   .setMaxCount(1024)
051 *   .setPropagateClose(false)
052 *   .get();
053 * }
054 * </pre>
055 * <h2>Using IO</h2>
056 * <pre>{@code
057 * BoundedInputStream s = BoundedInputStream.builder()
058 *   .setFile(new File("MyFile.xml"))
059 *   .setMaxCount(1024)
060 *   .setPropagateClose(false)
061 *   .get();
062 * }
063 * </pre>
064 * <h2>Counting Bytes</h2>
065 * <p>You can set the running count when building, which is most useful when starting from another stream:
066 * <pre>{@code
067 * InputStream in = ...;
068 * BoundedInputStream s = BoundedInputStream.builder()
069 *   .setInputStream(in)
070 *   .setCount(12)
071 *   .setMaxCount(1024)
072 *   .setPropagateClose(false)
073 *   .get();
074 * }
075 * </pre>
076 * @see Builder
077 * @since 2.0
078 */
079//@formatter:on
080public class BoundedInputStream extends ProxyInputStream {
081
082    /**
083     * For subclassing builders from {@link BoundedInputStream} subclassses.
084     *
085     * @param <T> The subclass.
086     */
087    static abstract class AbstractBuilder<T extends AbstractBuilder<T>> extends AbstractStreamBuilder<BoundedInputStream, T> {
088
089        /** The current count of bytes counted. */
090        private long count;
091
092        /** The max count of bytes to read. */
093        private long maxCount = EOF;
094
095        /** Flag if {@link #close()} should be propagated, {@code true} by default. */
096        private boolean propagateClose = true;
097
098        long getCount() {
099            return count;
100        }
101
102        long getMaxCount() {
103            return maxCount;
104        }
105
106        boolean isPropagateClose() {
107            return propagateClose;
108        }
109
110        /**
111         * Sets the current number of bytes counted.
112         * <p>
113         * Useful when building from another stream to carry forward a read count.
114         * </p>
115         * <p>
116         * Default is {@code 0}, negative means 0.
117         * </p>
118         *
119         * @param count The current number of bytes counted.
120         * @return this.
121         */
122        public T setCount(final long count) {
123            this.count = Math.max(0, count);
124            return asThis();
125        }
126
127        /**
128         * Sets the maximum number of bytes to return.
129         * <p>
130         * Default is {@value IOUtils#EOF}, negative means unbound.
131         * </p>
132         *
133         * @param maxCount The maximum number of bytes to return.
134         * @return this.
135         */
136        public T setMaxCount(final long maxCount) {
137            this.maxCount = Math.max(EOF, maxCount);
138            return asThis();
139        }
140
141        /**
142         * Sets whether the {@link #close()} method should propagate to the underling {@link InputStream}.
143         * <p>
144         * Default is {@code true}.
145         * </p>
146         *
147         * @param propagateClose {@code true} if calling {@link #close()} propagates to the {@code close()} method of the underlying stream or {@code false} if
148         *                       it does not.
149         * @return this.
150         */
151        public T setPropagateClose(final boolean propagateClose) {
152            this.propagateClose = propagateClose;
153            return asThis();
154        }
155
156    }
157
158    //@formatter:off
159    /**
160     * Builds a new {@link BoundedInputStream}.
161     * <p>
162     * By default, a {@link BoundedInputStream} is <em>unbound</em>; so make sure to call {@link AbstractBuilder#setMaxCount(long)}.
163     * </p>
164     * <p>
165     * You can find out how many bytes this stream has seen so far by calling {@link BoundedInputStream#getCount()}. This value reflects bytes read and skipped.
166     * </p>
167     * <h2>Using a ServletInputStream</h2>
168     * <p>
169     * A {@code ServletInputStream} can block if you try to read content that isn't there
170     * because it doesn't know whether the content hasn't arrived yet or whether the content has finished. Initialize an {@link BoundedInputStream} with the
171     * {@code Content-Length} sent in the {@code ServletInputStream}'s header, this stop it from blocking, providing it's been sent with a correct content
172     * length in the first place.
173     * </p>
174     * <h2>Using NIO</h2>
175     * <pre>{@code
176     * BoundedInputStream s = BoundedInputStream.builder()
177     *   .setPath(Paths.get("MyFile.xml"))
178     *   .setMaxCount(1024)
179     *   .setPropagateClose(false)
180     *   .get();
181     * }
182     * </pre>
183     * <h2>Using IO</h2>
184     * <pre>{@code
185     * BoundedInputStream s = BoundedInputStream.builder()
186     *   .setFile(new File("MyFile.xml"))
187     *   .setMaxCount(1024)
188     *   .setPropagateClose(false)
189     *   .get();
190     * }
191     * </pre>
192     * <h2>Counting Bytes</h2>
193     * <p>You can set the running count when building, which is most useful when starting from another stream:
194     * <pre>{@code
195     * InputStream in = ...;
196     * BoundedInputStream s = BoundedInputStream.builder()
197     *   .setInputStream(in)
198     *   .setCount(12)
199     *   .setMaxCount(1024)
200     *   .setPropagateClose(false)
201     *   .get();
202     * }
203     * </pre>
204     *
205     * @see #get()
206     * @since 2.16.0
207     */
208    //@formatter:on
209    public static class Builder extends AbstractBuilder<Builder> {
210
211        /**
212         * Builds a new {@link BoundedInputStream}.
213         * <p>
214         * You must set input that supports {@link #getInputStream()}, otherwise, this method throws an exception.
215         * </p>
216         * <p>
217         * This builder use the following aspects:
218         * </p>
219         * <ul>
220         * <li>{@link #getInputStream()}</li>
221         * <li>maxCount</li>
222         * <li>propagateClose</li>
223         * </ul>
224         *
225         * @return a new instance.
226         * @throws IllegalStateException         if the {@code origin} is {@code null}.
227         * @throws UnsupportedOperationException if the origin cannot be converted to an {@link InputStream}.
228         * @throws IOException                   if an I/O error occurs.
229         * @see #getInputStream()
230         */
231        @SuppressWarnings("resource")
232        @Override
233        public BoundedInputStream get() throws IOException {
234            return new BoundedInputStream(getInputStream(), getCount(), getMaxCount(), isPropagateClose());
235        }
236
237    }
238
239    /**
240     * Constructs a new {@link AbstractBuilder}.
241     *
242     * @return a new {@link AbstractBuilder}.
243     * @since 2.16.0
244     */
245    public static Builder builder() {
246        return new Builder();
247    }
248
249    /** The current count of bytes counted. */
250    private long count;
251
252    /** The max count of bytes to read. */
253    private final long maxCount;
254
255    /**
256     * Flag if close should be propagated.
257     *
258     * TODO Make final in 3.0.
259     */
260    private boolean propagateClose = true;
261
262    /**
263     * Constructs a new {@link BoundedInputStream} that wraps the given input stream and is unlimited.
264     *
265     * @param in The wrapped input stream.
266     * @deprecated Use {@link AbstractBuilder#get()}.
267     */
268    @Deprecated
269    public BoundedInputStream(final InputStream in) {
270        this(in, EOF);
271    }
272
273    /**
274     * Constructs a new {@link BoundedInputStream} that wraps the given input stream and limits it to a certain size.
275     *
276     * @param inputStream The wrapped input stream.
277     * @param maxCount    The maximum number of bytes to return.
278     * @deprecated Use {@link AbstractBuilder#get()}.
279     */
280    @Deprecated
281    public BoundedInputStream(final InputStream inputStream, final long maxCount) {
282        // Some badly designed methods - e.g. the Servlet API - overload length
283        // such that "-1" means stream finished
284        this(inputStream, 0, maxCount, true);
285    }
286
287    /**
288     * Constructs a new {@link BoundedInputStream} that wraps the given input stream and limits it to a certain size.
289     *
290     * @param inputStream    The wrapped input stream.
291     * @param count          The current number of bytes read.
292     * @param maxCount       The maximum number of bytes to return.
293     * @param propagateClose {@code true} if calling {@link #close()} propagates to the {@code close()} method of the underlying stream or {@code false} if it
294     *                       does not.
295     */
296    BoundedInputStream(final InputStream inputStream, final long count, final long maxCount, final boolean propagateClose) {
297        // Some badly designed methods - e.g. the Servlet API - overload length
298        // such that "-1" means stream finished
299        // Can't throw because we start from an InputStream.
300        super(inputStream);
301        this.count = count;
302        this.maxCount = maxCount;
303        this.propagateClose = propagateClose;
304    }
305
306    /**
307     * Adds the number of read bytes to the count.
308     *
309     * @param n number of bytes read, or -1 if no more bytes are available
310     * @throws IOException Not thrown here but subclasses may throw.
311     * @since 2.0
312     */
313    @Override
314    protected synchronized void afterRead(final int n) throws IOException {
315        if (n != EOF) {
316            count += n;
317        }
318    }
319
320    /**
321     * {@inheritDoc}
322     */
323    @Override
324    public int available() throws IOException {
325        if (isMaxCount()) {
326            onMaxLength(maxCount, getCount());
327            return 0;
328        }
329        return in.available();
330    }
331
332    /**
333     * Invokes the delegate's {@link InputStream#close()} method if {@link #isPropagateClose()} is {@code true}.
334     *
335     * @throws IOException if an I/O error occurs.
336     */
337    @Override
338    public void close() throws IOException {
339        if (propagateClose) {
340            in.close();
341        }
342    }
343
344    /**
345     * Gets the count of bytes read.
346     *
347     * @return The count of bytes read.
348     * @since 2.12.0
349     */
350    public long getCount() {
351        return count;
352    }
353
354    /**
355     * Gets the max count of bytes to read.
356     *
357     * @return The max count of bytes to read.
358     * @since 2.16.0
359     */
360    public long getMaxCount() {
361        return maxCount;
362    }
363
364    /**
365     * Gets the max count of bytes to read.
366     *
367     * @return The max count of bytes to read.
368     * @since 2.12.0
369     * @deprecated Use {@link #getMaxCount()}.
370     */
371    @Deprecated
372    public long getMaxLength() {
373        return maxCount;
374    }
375
376    /**
377     * Gets how many bytes remain to read.
378     *
379     * @return bytes how many bytes remain to read.
380     * @since 2.16.0
381     */
382    public long getRemaining() {
383        return Math.max(0, getMaxCount() - getCount());
384    }
385
386    private boolean isMaxCount() {
387        return maxCount >= 0 && getCount() >= maxCount;
388    }
389
390    /**
391     * Tests whether the {@link #close()} method should propagate to the underling {@link InputStream}.
392     *
393     * @return {@code true} if calling {@link #close()} propagates to the {@code close()} method of the underlying stream or {@code false} if it does not.
394     */
395    public boolean isPropagateClose() {
396        return propagateClose;
397    }
398
399    /**
400     * Invokes the delegate's {@link InputStream#mark(int)} method.
401     *
402     * @param readLimit read ahead limit
403     */
404    @Override
405    public synchronized void mark(final int readLimit) {
406        in.mark(readLimit);
407    }
408
409    /**
410     * Invokes the delegate's {@link InputStream#markSupported()} method.
411     *
412     * @return true if mark is supported, otherwise false
413     */
414    @Override
415    public boolean markSupported() {
416        return in.markSupported();
417    }
418
419    /**
420     * A caller has caused a request that would cross the {@code maxLength} boundary.
421     *
422     * @param maxLength The max count of bytes to read.
423     * @param count     The count of bytes read.
424     * @throws IOException Subclasses may throw.
425     * @since 2.12.0
426     */
427    @SuppressWarnings("unused")
428    protected void onMaxLength(final long maxLength, final long count) throws IOException {
429        // for subclasses
430    }
431
432    /**
433     * Invokes the delegate's {@link InputStream#read()} method if the current position is less than the limit.
434     *
435     * @return the byte read or -1 if the end of stream or the limit has been reached.
436     * @throws IOException if an I/O error occurs.
437     */
438    @Override
439    public int read() throws IOException {
440        if (isMaxCount()) {
441            onMaxLength(maxCount, getCount());
442            return EOF;
443        }
444        return super.read();
445    }
446
447    /**
448     * Invokes the delegate's {@link InputStream#read(byte[])} method.
449     *
450     * @param b the buffer to read the bytes into
451     * @return the number of bytes read or -1 if the end of stream or the limit has been reached.
452     * @throws IOException if an I/O error occurs.
453     */
454    @Override
455    public int read(final byte[] b) throws IOException {
456        return read(b, 0, b.length);
457    }
458
459    /**
460     * Invokes the delegate's {@link InputStream#read(byte[], int, int)} method.
461     *
462     * @param b   the buffer to read the bytes into
463     * @param off The start offset
464     * @param len The number of bytes to read
465     * @return the number of bytes read or -1 if the end of stream or the limit has been reached.
466     * @throws IOException if an I/O error occurs.
467     */
468    @Override
469    public int read(final byte[] b, final int off, final int len) throws IOException {
470        if (isMaxCount()) {
471            onMaxLength(maxCount, getCount());
472            return EOF;
473        }
474        return super.read(b, off, (int) toReadLen(len));
475    }
476
477    /**
478     * Invokes the delegate's {@link InputStream#reset()} method.
479     *
480     * @throws IOException if an I/O error occurs.
481     */
482    @Override
483    public synchronized void reset() throws IOException {
484        in.reset();
485    }
486
487    /**
488     * Sets whether the {@link #close()} method should propagate to the underling {@link InputStream}.
489     *
490     * @param propagateClose {@code true} if calling {@link #close()} propagates to the {@code close()} method of the underlying stream or {@code false} if it
491     *                       does not.
492     * @deprecated Use {@link AbstractBuilder#setPropagateClose(boolean)}.
493     */
494    @Deprecated
495    public void setPropagateClose(final boolean propagateClose) {
496        this.propagateClose = propagateClose;
497    }
498
499    /**
500     * Invokes the delegate's {@link InputStream#skip(long)} method.
501     *
502     * @param n the number of bytes to skip
503     * @return the actual number of bytes skipped
504     * @throws IOException if an I/O error occurs.
505     */
506    @Override
507    public long skip(final long n) throws IOException {
508        final long skip = super.skip(toReadLen(n));
509        count += skip;
510        return skip;
511    }
512
513    private long toReadLen(final long len) {
514        return maxCount >= 0 ? Math.min(len, maxCount - getCount()) : len;
515    }
516
517    /**
518     * Invokes the delegate's {@link InputStream#toString()} method.
519     *
520     * @return the delegate's {@link InputStream#toString()}
521     */
522    @Override
523    public String toString() {
524        return in.toString();
525    }
526}