Add support to ThrottledInputStream for setting a consumer for

ProxyInputStream.afterRead(int)

Author: garydgregory

src/changes/changes.xml

@@ -70,6 +70,7 @@ The <action> type attribute can be add,update,fix,remove.
       <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>
       <action dev="ggregory" type="add"                due-to="Gary Gregory">Add support to ChecksumInputStream for setting a consumer for ProxyInputStream.afterRead(int).</action>
+      <action dev="ggregory" type="add"                due-to="Gary Gregory">Add support to ThrottledInputStream 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/ThrottledInputStream.java

@@ -24,8 +24,6 @@
 import java.time.temporal.ChronoUnit;
 import java.util.concurrent.TimeUnit;
 
-import org.apache.commons.io.build.AbstractStreamBuilder;
-
 /**
  * Provides bandwidth throttling on a specified InputStream. It is implemented as a wrapper on top of another InputStream instance. The throttling works by
  * examining the number of bytes read from the underlying InputStream from the beginning, and sleep()ing for a time interval if the byte-transfer is found
@@ -74,7 +72,7 @@ public final class ThrottledInputStream extends CountingInputStream {
      * @see #get()
      */
     // @formatter:on
-    public static class Builder extends AbstractStreamBuilder<ThrottledInputStream, Builder> {
+    public static class Builder extends AbstractBuilder<ThrottledInputStream, Builder> {
 
         /**
          * Effectively not throttled.
@@ -103,7 +101,7 @@ public static class Builder extends AbstractStreamBuilder<ThrottledInputStream,
         @SuppressWarnings("resource")
         @Override
         public ThrottledInputStream get() throws IOException {
-            return new ThrottledInputStream(getInputStream(), maxBytesPerSecond);
+            return new ThrottledInputStream(this);
         }
 
         /**
@@ -147,12 +145,12 @@ static long toSleepMillis(final long bytesRead, final long maxBytesPerSec, final
     private final long startTime = System.currentTimeMillis();
     private Duration totalSleepDuration = Duration.ZERO;
 
-    private ThrottledInputStream(final InputStream proxy, final long maxBytesPerSecond) {
-        super(proxy);
-        if (maxBytesPerSecond <= 0) {
-            throw new IllegalArgumentException("Bandwidth " + maxBytesPerSecond + " is invalid.");
+    private ThrottledInputStream(final Builder builder) throws IOException {
+        super(builder);
+        if (builder.maxBytesPerSecond <= 0) {
+            throw new IllegalArgumentException("Bandwidth " + builder.maxBytesPerSecond + " is invalid.");
         }
-        this.maxBytesPerSecond = maxBytesPerSecond;
+        this.maxBytesPerSecond = builder.maxBytesPerSecond;
     }
 
     @Override

src/test/java/org/apache/commons/io/input/ThrottledInputStreamTest.java

@@ -18,10 +18,16 @@
 package org.apache.commons.io.input;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrowsExactly;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 
 import java.io.IOException;
+import java.io.InputStream;
 import java.time.Duration;
+import java.util.concurrent.atomic.AtomicBoolean;
 
+import org.apache.commons.io.IOUtils;
+import org.apache.commons.io.test.CustomIOException;
 import org.junit.jupiter.api.Test;
 
 /**
@@ -35,6 +41,32 @@ protected ThrottledInputStream createFixture() throws IOException {
         return ThrottledInputStream.builder().setInputStream(createOriginInputStream()).get();
     }
 
+    @Test
+    public void testAfterReadConsumer() throws Exception {
+        final AtomicBoolean boolRef = new AtomicBoolean();
+        // @formatter:off
+        try (InputStream bounded = ThrottledInputStream.builder()
+                .setCharSequence("Hi")
+                .setAfterRead(i -> boolRef.set(true))
+                .get()) {
+            IOUtils.consume(bounded);
+        }
+        // @formatter:on
+        assertTrue(boolRef.get());
+        // Throwing
+        final String message = "test exception message";
+        // @formatter:off
+        try (InputStream bounded = ThrottledInputStream.builder()
+                .setCharSequence("Hi")
+                .setAfterRead(i -> {
+                    throw new CustomIOException(message);
+                })
+                .get()) {
+            assertEquals(message, assertThrowsExactly(CustomIOException.class, () -> IOUtils.consume(bounded)).getMessage());
+        }
+        // @formatter:on
+    }
+
     @Test
     public void testCalSleepTimeMs() {
         // case 0: initial - no read, no sleep