Add constructor ThresholdingOutputStream(int, IOConsumer, IOFunction)

and make the class concrete.

Author: Gary Gregory

src/changes/changes.xml

@@ -210,6 +210,9 @@ The <action> type attribute can be add,update,fix,remove.
       <action dev="ggregory" type="add" due-to="Gary Gregory">
         Add IOConsumer.noop().
       </action>
+      <action dev="ggregory" type="add" due-to="Gary Gregory">
+        Add constructor ThresholdingOutputStream(int, IOConsumer, IOFunction) and make the class concrete.
+      </action>
       <!-- UPDATES -->
       <action dev="ggregory" type="update" due-to="Dependabot">
         Update junit-jupiter from 5.6.2 to 5.7.0 #153.

src/main/java/org/apache/commons/io/output/ThresholdingOutputStream.java

@@ -19,6 +19,9 @@
 import java.io.IOException;
 import java.io.OutputStream;
 
+import org.apache.commons.io.function.IOConsumer;
+import org.apache.commons.io.function.IOFunction;
+
 /**
  * An output stream which triggers an event when a specified number of bytes of data have been written to it. The event
  * can be used, for example, to throw an exception if a maximum has been reached, or to switch the underlying stream
@@ -32,13 +35,28 @@
  * when a pending write operation would cause the threshold to be exceeded.
  * </p>
  */
-public abstract class ThresholdingOutputStream extends OutputStream {
+public class ThresholdingOutputStream extends OutputStream {
+
+    /**
+     * Noop output stream getter function.
+     */
+    private static IOFunction<ThresholdingOutputStream, OutputStream> NOOP_OS_GETTER = os -> NullOutputStream.NULL_OUTPUT_STREAM;
 
     /**
      * The threshold at which the event will be triggered.
      */
     private final int threshold;
 
+    /**
+     * Accepts reaching the threshold.
+     */
+    private final IOConsumer<ThresholdingOutputStream> thresholdConsumer;
+
+    /**
+     * Gets the output stream.
+     */
+    private final IOFunction<ThresholdingOutputStream, OutputStream> outputStreamGetter;
+
     /**
      * The number of bytes written to the output stream.
      */
@@ -55,7 +73,22 @@ public abstract class ThresholdingOutputStream extends OutputStream {
      * @param threshold The number of bytes at which to trigger an event.
      */
     public ThresholdingOutputStream(final int threshold) {
+        this(threshold, IOConsumer.noop(), NOOP_OS_GETTER);
+    }
+
+    /**
+     * Constructs an instance of this class which will trigger an event at the specified threshold.
+     *
+     * @param threshold The number of bytes at which to trigger an event.
+     * @param thresholdConsumer Accepts reaching the threshold.
+     * @param outputStreamGetter Gets the output stream.
+     * @since 2.9.0
+     */
+    public ThresholdingOutputStream(final int threshold, final IOConsumer<ThresholdingOutputStream> thresholdConsumer,
+        final IOFunction<ThresholdingOutputStream, OutputStream> outputStreamGetter) {
         this.threshold = threshold;
+        this.thresholdConsumer = thresholdConsumer == null ? IOConsumer.noop() : thresholdConsumer;
+        this.outputStreamGetter = outputStreamGetter == null ? NOOP_OS_GETTER : outputStreamGetter;
     }
 
     /**
@@ -116,7 +149,9 @@ public long getByteCount() {
      *
      * @throws IOException if an error occurs.
      */
-    protected abstract OutputStream getStream() throws IOException;
+    protected OutputStream getStream() throws IOException {
+        return outputStreamGetter.apply(this);
+    }
 
     /**
      * Returns the threshold, in bytes, at which an event will be triggered.
@@ -162,7 +197,9 @@ protected void setByteCount(final long count) {
      *
      * @throws IOException if an error occurs.
      */
-    protected abstract void thresholdReached() throws IOException;
+    protected void thresholdReached() throws IOException {
+        thresholdConsumer.accept(this);
+    }
 
     /**
      * Writes {@code b.length} bytes from the specified byte array to this output stream.

src/test/java/org/apache/commons/io/output/ThresholdingOutputStreamTest.java

@@ -18,6 +18,7 @@
 package org.apache.commons.io.output;
 
 import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 import java.io.IOException;
@@ -46,10 +47,61 @@ protected void thresholdReached() throws IOException {
                 reached.set(true);
             }
         }) {
-            tos.write(12);
+            tos.write('a');
             assertFalse(reached.get());
-            tos.write(12);
+            tos.write('a');
             assertTrue(reached.get());
         }
     }
+
+    @Test
+    public void testThresholdIOConsumer() throws Exception {
+        final AtomicBoolean reached = new AtomicBoolean();
+        // Null threshold consumer
+        reached.set(false);
+        try (final ThresholdingOutputStream tos = new ThresholdingOutputStream(1, null,
+            os -> new ByteArrayOutputStream(4))) {
+            tos.write('a');
+            assertFalse(reached.get());
+            tos.write('a');
+            assertFalse(reached.get());
+        }
+        // Null output stream function
+        reached.set(false);
+        try (final ThresholdingOutputStream tos = new ThresholdingOutputStream(1, os -> reached.set(true), null)) {
+            tos.write('a');
+            assertFalse(reached.get());
+            tos.write('a');
+            assertTrue(reached.get());
+        }
+        // non-null inputs.
+        reached.set(false);
+        try (final ThresholdingOutputStream tos = new ThresholdingOutputStream(1, os -> reached.set(true),
+            os -> new ByteArrayOutputStream(4))) {
+            tos.write('a');
+            assertFalse(reached.get());
+            tos.write('a');
+            assertTrue(reached.get());
+        }
+    }
+
+    @Test
+    public void testThresholdIOConsumerIOException() throws Exception {
+        try (final ThresholdingOutputStream tos = new ThresholdingOutputStream(1, os -> {
+            throw new IOException("Threshold reached.");
+        }, os -> new ByteArrayOutputStream(4))) {
+            tos.write('a');
+            assertThrows(IOException.class, () -> tos.write('a'));
+        }
+    }
+
+    @Test
+    public void testThresholdIOConsumerUncheckedException() throws Exception {
+        try (final ThresholdingOutputStream tos = new ThresholdingOutputStream(1, os -> {
+            throw new IllegalStateException("Threshold reached.");
+        }, os -> new ByteArrayOutputStream(4))) {
+            tos.write('a');
+            assertThrows(IllegalStateException.class, () -> tos.write('a'));
+        }
+    }
 }