Add and use ObservableInputStream.ObservableInputStream(InputStream,
Observer...).
MODIFY
src/changes/changes.xml
src/changes/changes.xml
@@ -157,6 +157,9 @@ The <action> type attribute can be add,update,fix,remove.
<action dev="ggregory" type="add" due-to="Gary Gregory">
Add and use IOExceptionList(String, List).
</action>
+ <action dev="ggregory" type="add" due-to="Gary Gregory">
+ Add and use ObservableInputStream.ObservableInputStream(InputStream, Observer...).
+ </action>
<!-- UPDATES -->
<action dev="ggregory" type="update" due-to="Dependabot">
Update junit-jupiter from 5.6.2 to 5.7.0 #153.

MODIFY
src/main/java/org/apache/commons/io/input/MessageDigestCalculatingInputStream.java
src/main/java/org/apache/commons/io/input/MessageDigestCalculatingInputStream.java
@@ -61,12 +61,11 @@ public void data(final byte[] input, final int offset, final int length) throws
/** Creates a new instance, which calculates a signature on the given stream,
* using the given {@link MessageDigest}.
* @param inputStream the stream to calculate the message digest for
- * @param MessageDigest the message digest to use
+ * @param messageDigest the message digest to use
*/
- public MessageDigestCalculatingInputStream(final InputStream inputStream, final MessageDigest MessageDigest) {
- super(inputStream);
- this.messageDigest = MessageDigest;
- add(new MessageDigestMaintainingObserver(MessageDigest));
+ public MessageDigestCalculatingInputStream(final InputStream inputStream, final MessageDigest messageDigest) {
+ super(inputStream, new MessageDigestMaintainingObserver(messageDigest));
+ this.messageDigest = messageDigest;
}
/**

MODIFY
src/main/java/org/apache/commons/io/input/ObservableInputStream.java
src/main/java/org/apache/commons/io/input/ObservableInputStream.java
import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.List;
import org.apache.commons.io.IOUtils;
@@ -100,7 +101,7 @@ public void finished() throws IOException {
}
}
- private final List<Observer> observers = new ArrayList<>();
+ private final List<Observer> observers;
/**
* Creates a new ObservableInputStream for the given InputStream.
@@ -108,7 +109,29 @@ public void finished() throws IOException {
* @param inputStream the input stream to proxy.
*/
public ObservableInputStream(final InputStream inputStream) {
+ this(inputStream, new ArrayList<>());
+ }
+
+ /**
+ * Creates a new ObservableInputStream for the given InputStream.
+ *
+ * @param inputStream the input stream to proxy.
+ * @param observers List of callbacks.
+ */
+ private ObservableInputStream(final InputStream inputStream, final List<Observer> observers) {
super(inputStream);
+ this.observers = observers;
+ }
+
+ /**
+ * Creates a new ObservableInputStream for the given InputStream.
+ *
+ * @param inputStream the input stream to proxy.
+ * @param observers List of callbacks.
+ * @since 2.9.0
+ */
+ public ObservableInputStream(final InputStream inputStream, final Observer... observers) {
+ this(inputStream, Arrays.asList(observers));
}
/**

MODIFY
src/test/java/org/apache/commons/io/input/ObservableInputStreamTest.java
src/test/java/org/apache/commons/io/input/ObservableInputStreamTest.java
package org.apache.commons.io.input;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
import org.junit.jupiter.api.Test;
public class ObservableInputStreamTest {
+
private static class LastByteKeepingObserver extends Observer {
private int lastByteSeen = -1;
private boolean finished;
@@ -70,7 +72,7 @@ public void data(final byte[] pBuffer, final int pOffset, final int pLength) thr
* Tests that {@link Observer#data(int)} is called.
*/
@Test
- public void testDataByteCalled() throws Exception {
+ public void testDataByteCalled_add() throws Exception {
final byte[] buffer = MessageDigestCalculatingInputStreamTest
.generateRandomByteStream(IOUtils.DEFAULT_BUFFER_SIZE);
final LastByteKeepingObserver lko = new LastByteKeepingObserver();
@@ -98,6 +100,37 @@ public void testDataByteCalled() throws Exception {
}
}
+ /**
+ * Tests that {@link Observer#data(int)} is called.
+ */
+ @Test
+ public void testDataByteCalled_ctor() throws Exception {
+ final byte[] buffer = MessageDigestCalculatingInputStreamTest
+ .generateRandomByteStream(IOUtils.DEFAULT_BUFFER_SIZE);
+ final LastByteKeepingObserver lko = new LastByteKeepingObserver();
+ try (final ObservableInputStream ois = new ObservableInputStream(new ByteArrayInputStream(buffer), lko)) {
+ assertEquals(-1, lko.lastByteSeen);
+ ois.read();
+ assertNotEquals(-1, lko.lastByteSeen);
+ assertFalse(lko.finished);
+ assertFalse(lko.closed);
+ for (int i = 1; i < buffer.length; i++) {
+ final int result = ois.read();
+ assertEquals((byte) result, buffer[i]);
+ assertEquals(result, lko.lastByteSeen);
+ assertFalse(lko.finished);
+ assertFalse(lko.closed);
+ }
+ final int result = ois.read();
+ assertEquals(-1, result);
+ assertTrue(lko.finished);
+ assertFalse(lko.closed);
+ ois.close();
+ assertTrue(lko.finished);
+ assertTrue(lko.closed);
+ }
+ }
+
/**
* Tests that {@link Observer#data(byte[],int,int)} is called.
*/
@@ -137,4 +170,12 @@ public void testDataBytesCalled() throws Exception {
}
}
+ @Test
+ public void testTimestamped() {
+ final byte[] buffer = MessageDigestCalculatingInputStreamTest
+ .generateRandomByteStream(IOUtils.DEFAULT_BUFFER_SIZE);
+ final ByteArrayInputStream bais = new ByteArrayInputStream(buffer);
+ final ObservableInputStream ois = new ObservableInputStream(bais);
+
+ }
}

Add and use ObservableInputStream.ObservableInputStream(InputStream,
Observer...).