Add support to BoundedInputStream for setting a consumer for

ProxyInputStream.afterRead(int)

Author: garydgregory

src/changes/changes.xml

@@ -68,6 +68,7 @@ The <action> type attribute can be add,update,fix,remove.
       <action dev="ggregory" type="add" issue="IO-861" due-to="Gary Gregory">Add ProxyInputStream.AbstractBuilder. Supports setting a consumer for ProxyInputStream.afterRead(int).</action>
       <action dev="ggregory" type="add"                due-to="Gary Gregory">Add support to AutoCloseInputStream for setting a consumer for ProxyInputStream.afterRead(int).</action>
       <action dev="ggregory" type="add"                due-to="Gary Gregory">Add support to BOMInputStream for setting a consumer for ProxyInputStream.afterRead(int).</action>
+      <action dev="ggregory" type="add" issue="IO-861" due-to="Gary Gregory">Add support to BoundedInputStream for setting a consumer for ProxyInputStream.afterRead(int).</action>
       <!-- UPDATE -->
       <action dev="ggregory" type="update"             due-to="Gary Gregory">Bump org.apache.commons:commons-parent from 74 to 78 #670, #676, #679, #688.</action>
       <action dev="ggregory" type="update"             due-to="Gary Gregory">Bump commons.bytebuddy.version from 1.15.1 to 1.15.10 #672, #673, #685, #686, #694, #696, #698.</action>

src/main/java/org/apache/commons/io/input/BoundedInputStream.java

@@ -22,7 +22,7 @@
 import java.io.InputStream;
 
 import org.apache.commons.io.IOUtils;
-import org.apache.commons.io.build.AbstractStreamBuilder;
+import org.apache.commons.io.function.IOBiConsumer;
 
 //@formatter:off
 /**
@@ -73,6 +73,15 @@
  *   .get();
  * }
  * </pre>
+ * <h2>Listening for the max count reached</h2>
+ * <pre>{@code
+ * BoundedInputStream s = BoundedInputStream.builder()
+ *   .setPath(Paths.get("MyFile.xml"))
+ *   .setMaxCount(1024)
+ *   .setOnMaxCount((max, count) -> System.out.printf("Max count %,d reached with a last read count of %,d%n", max, count))
+ *   .get();
+ * }
+ * </pre>
  * @see Builder
  * @since 2.0
  */
@@ -84,14 +93,16 @@ public class BoundedInputStream extends ProxyInputStream {
      *
      * @param <T> The subclass.
      */
-    static abstract class AbstractBuilder<T extends AbstractBuilder<T>> extends AbstractStreamBuilder<BoundedInputStream, T> {
+    static abstract class AbstractBuilder<T extends AbstractBuilder<T>> extends ProxyInputStream.AbstractBuilder<BoundedInputStream, T> {
 
         /** The current count of bytes counted. */
         private long count;
 
         /** The max count of bytes to read. */
         private long maxCount = EOF;
 
+        private IOBiConsumer<Long, Long> onMaxCount = IOBiConsumer.noop();
+
         /** Flag if {@link #close()} should be propagated, {@code true} by default. */
         private boolean propagateClose = true;
 
@@ -103,6 +114,10 @@ long getMaxCount() {
             return maxCount;
         }
 
+        IOBiConsumer<Long, Long> getOnMaxCount() {
+            return onMaxCount;
+        }
+
         boolean isPropagateClose() {
             return propagateClose;
         }
@@ -138,6 +153,25 @@ public T setMaxCount(final long maxCount) {
             return asThis();
         }
 
+        /**
+         * Sets the default {@link BoundedInputStream#onMaxLength(long, long)} behavior, {@code null} resets to a NOOP.
+         * <p>
+         * The first Long is the max count of bytes to read. The second Long is the count of bytes read.
+         * </p>
+         * <p>
+         * This does <em>not</em> override a {@code BoundedInputStream} subclass' implementation of the {@link BoundedInputStream#onMaxLength(long, long)}
+         * method.
+         * </p>
+         *
+         * @param onMaxCount the {@link ProxyInputStream#afterRead(int)} behavior.
+         * @return this instance.
+         * @since 2.18.0
+         */
+        public T setOnMaxCount(final IOBiConsumer<Long, Long> onMaxCount) {
+            this.onMaxCount = onMaxCount != null ? onMaxCount : IOBiConsumer.noop();
+            return asThis();
+        }
+
         /**
          * Sets whether the {@link #close()} method should propagate to the underling {@link InputStream}.
          * <p>
@@ -218,8 +252,11 @@ public static class Builder extends AbstractBuilder<Builder> {
          * </p>
          * <ul>
          * <li>{@link #getInputStream()}</li>
-         * <li>maxCount</li>
-         * <li>propagateClose</li>
+         * <li>{@link #getAfterRead()}</li>
+         * <li>{@link #getCount()}</li>
+         * <li>{@link #getMaxCount()}</li>
+         * <li>{@link #isPropagateClose()}</li>
+         * <li>{@link #getOnMaxCount()}</li>
          * </ul>
          *
          * @return a new instance.
@@ -228,10 +265,9 @@ public static class Builder extends AbstractBuilder<Builder> {
          * @throws IOException                   if an I/O error occurs.
          * @see #getInputStream()
          */
-        @SuppressWarnings("resource")
         @Override
         public BoundedInputStream get() throws IOException {
-            return new BoundedInputStream(getInputStream(), getCount(), getMaxCount(), isPropagateClose());
+            return new BoundedInputStream(this);
         }
 
     }
@@ -255,13 +291,23 @@ public static Builder builder() {
     /** The max count of bytes to read. */
     private final long maxCount;
 
+    private final IOBiConsumer<Long, Long> onMaxCount;
+
     /**
      * Flag if close should be propagated.
      *
      * TODO Make final in 3.0.
      */
     private boolean propagateClose = true;
 
+    BoundedInputStream(final Builder builder) throws IOException {
+        super(builder);
+        this.count = builder.getCount();
+        this.maxCount = builder.getMaxCount();
+        this.propagateClose = builder.isPropagateClose();
+        this.onMaxCount = builder.getOnMaxCount();
+    }
+
     /**
      * Constructs a new {@link BoundedInputStream} that wraps the given input stream and is unlimited.
      *
@@ -273,6 +319,14 @@ public BoundedInputStream(final InputStream in) {
         this(in, EOF);
     }
 
+    BoundedInputStream(final InputStream inputStream, final Builder builder) {
+        super(inputStream, builder);
+        this.count = builder.getCount();
+        this.maxCount = builder.getMaxCount();
+        this.propagateClose = builder.isPropagateClose();
+        this.onMaxCount = builder.getOnMaxCount();
+    }
+
     /**
      * Constructs a new {@link BoundedInputStream} that wraps the given input stream and limits it to a certain size.
      *
@@ -284,26 +338,7 @@ public BoundedInputStream(final InputStream in) {
     public BoundedInputStream(final InputStream inputStream, final long maxCount) {
         // Some badly designed methods - e.g. the Servlet API - overload length
         // such that "-1" means stream finished
-        this(inputStream, 0, maxCount, true);
-    }
-
-    /**
-     * Constructs a new {@link BoundedInputStream} that wraps the given input stream and limits it to a certain size.
-     *
-     * @param inputStream    The wrapped input stream.
-     * @param count          The current number of bytes read.
-     * @param maxCount       The maximum number of bytes to return.
-     * @param propagateClose {@code true} if calling {@link #close()} propagates to the {@code close()} method of the underlying stream or {@code false} if it
-     *                       does not.
-     */
-    BoundedInputStream(final InputStream inputStream, final long count, final long maxCount, final boolean propagateClose) {
-        // Some badly designed methods - e.g. the Servlet API - overload length
-        // such that "-1" means stream finished
-        // Can't throw because we start from an InputStream.
-        super(inputStream);
-        this.count = count;
-        this.maxCount = maxCount;
-        this.propagateClose = propagateClose;
+        this(inputStream, builder().setMaxCount(maxCount));
     }
 
     /**
@@ -318,6 +353,7 @@ protected synchronized void afterRead(final int n) throws IOException {
         if (n != EOF) {
             count += n;
         }
+        super.afterRead(n);
     }
 
     /**
@@ -422,15 +458,19 @@ public boolean markSupported() {
 
     /**
      * A caller has caused a request that would cross the {@code maxLength} boundary.
+     * <p>
+     * Delegates to the consumer set in {@link Builder#setOnMaxCount(IOBiConsumer)}.
+     * </p>
      *
-     * @param maxLength The max count of bytes to read.
+     * @param max The max count of bytes to read.
      * @param count     The count of bytes read.
      * @throws IOException Subclasses may throw.
      * @since 2.12.0
      */
     @SuppressWarnings("unused")
-    protected void onMaxLength(final long maxLength, final long count) throws IOException {
-        // for subclasses
+    // TODO Rename to onMaxCount for 3.0
+    protected void onMaxLength(final long max, final long count) throws IOException {
+        onMaxCount.accept(max, count);
     }
 
     /**