Add QueueInputStream and QueueOutputStream as simpler alternatives to
PipedInputStream and PipedOutputStream
Sort new methods.
Clean up Javadocs.
MODIFY
src/changes/changes.xml
src/changes/changes.xml
@@ -97,6 +97,9 @@ The <action> type attribute can be add,update,fix,remove.
<action dev="ggregory" type="add" due-to="Rob Spoor, Gary Gregory">
Add factory methods to CloseShieldInputStream, CloseShieldReader, CloseShieldOutputStream, CloseShieldWriter, #173.
</action>
+ <action dev="ggregory" type="add" due-to="Rob Spoor, Gary Gregory">
+ Add QueueInputStream and QueueOutputStream as simpler alternatives to PipedInputStream and PipedOutputStream #171.
+ </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/QueueInputStream.java
src/main/java/org/apache/commons/io/input/QueueInputStream.java
import java.util.concurrent.LinkedBlockingQueue;
/**
- * Simple alternative to JDK {@link java.io.PipedInputStream}; queue input stream provides what's written in queue output stream.
+ * Simple alternative to JDK {@link java.io.PipedInputStream}; queue input stream provides what's written in queue
+ * output stream.
*
+ * <p>
* Example usage:
+ * </p>
+ *
* <pre>
* QueueInputStream inputStream = new QueueInputStream();
* QueueOutputStream outputStream = inputStream.newQueueOutputStream();
* inputStream.read();
* </pre>
*
- * Unlike JDK {@link PipedInputStream} and {@link PipedOutputStream}, queue input/output streams may be
- * used safely in a single thread or multiple threads. Also, unlike JDK classes, no special meaning is
- * attached to initial or current thread. Instances can be used longer after initial threads exited.
+ * Unlike JDK {@link PipedInputStream} and {@link PipedOutputStream}, queue input/output streams may be used safely in a
+ * single thread or multiple threads. Also, unlike JDK classes, no special meaning is attached to initial or current
+ * thread. Instances can be used longer after initial threads exited.
*
- * Closing a {@code QueueInputStream} has no effect. The methods in this class can be called after
- * the stream has been closed without generating an {@code IOException}.
+ * <p>
+ * Closing a {@code QueueInputStream} has no effect. The methods in this class can be called after the stream has been
+ * closed without generating an {@code IOException}.
+ * </p>
*
* @see QueueOutputStream
* @since 2.9.0
private final BlockingQueue<Integer> queue;
/**
- * Constructs a QueueInputStream with no limit to internal buffer size
+ * Constructs a new instance with no limit to its internal buffer size.
*/
public QueueInputStream() {
this(new LinkedBlockingQueue<>());
}
/**
- * Constructs a QueueInputStream with given buffer
+ * Constructs a new instance with given buffer
*
* @param queue backing queue for the stream
*/
@@ -68,7 +74,8 @@ public QueueInputStream(final BlockingQueue<Integer> queue) {
}
/**
- * Creates a new QueueOutputStream instance connected to this. Writes to the output stream will be visible to this input stream.
+ * Creates a new QueueOutputStream instance connected to this. Writes to the output stream will be visible to this
+ * input stream.
*
* @return QueueOutputStream connected to this stream
*/
@@ -77,7 +84,7 @@ public QueueOutputStream newQueueOutputStream() {
}
/**
- * Reads a single byte.
+ * Reads and returns a single byte.
*
* @return either the byte read or {@code -1} if the end of the stream has been reached
*/

MODIFY
src/main/java/org/apache/commons/io/output/QueueOutputStream.java
src/main/java/org/apache/commons/io/output/QueueOutputStream.java
import java.util.concurrent.LinkedBlockingQueue;
/**
- * Simple alternative to JDK {@link java.io.PipedOutputStream}; queue input stream provides what's
- * written in queue output stream.
- *
+ * Simple alternative to JDK {@link java.io.PipedOutputStream}; queue input stream provides what's written in queue
+ * output stream.
+ * <p>
* Example usage:
+ * </p>
+ *
* <pre>
* QueueOutputStream outputStream = new QueueOutputStream();
* QueueInputStream inputStream = outputStream.newPipeInputStream();
* inputStream.read();
* </pre>
*
- * Unlike JDK {@link PipedInputStream} and {@link PipedOutputStream}, queue input/output streams may be
- * used safely in a single thread or multiple threads. Also, unlike JDK classes, no special meaning is
- * attached to initial or current thread. Instances can be used longer after initial threads exited.
- *
- * Closing a {@code QueueOutputStream} has no effect. The methods in this class can be called after
- * the stream has been closed without generating an {@code IOException}.
+ * Unlike JDK {@link PipedInputStream} and {@link PipedOutputStream}, queue input/output streams may be used safely in a
+ * single thread or multiple threads. Also, unlike JDK classes, no special meaning is attached to initial or current
+ * thread. Instances can be used longer after initial threads exited.
+ * <p>
+ * Closing a {@code QueueOutputStream} has no effect. The methods in this class can be called after the stream has been
+ * closed without generating an {@code IOException}.
+ * </p>
*
* @see QueueInputStream
* @since 2.9.0
private final BlockingQueue<Integer> queue;
/**
- * Constructs a QueueOutputStream with no limit to internal buffer size
+ * Constructs a new instance with no limit to internal buffer size.
*/
public QueueOutputStream() {
this(new LinkedBlockingQueue<>());
}
/**
- * Constructs a QueueOutputStream with given buffer
+ * Constructs a new instance with given buffer.
*
* @param queue backing queue for the stream
*/
@@ -70,7 +73,8 @@ public QueueOutputStream(final BlockingQueue<Integer> queue) {
}
/**
- * Creates a new QueueInputStream instance connected to this. Writes to this output stream will be visible to the input stream.
+ * Creates a new QueueInputStream instance connected to this. Writes to this output stream will be visible to the
+ * input stream.
*
* @return QueueInputStream connected to this stream
*/
@@ -95,4 +99,3 @@ public void write(final int b) throws InterruptedIOException {
}
}
}
-

MODIFY
src/test/java/org/apache/commons/io/output/QueueOutputStreamTest.java
src/test/java/org/apache/commons/io/output/QueueOutputStreamTest.java
@@ -50,28 +50,19 @@ public static void afterAll() {
executorService.shutdown();
}
- @Test
- public void writeString() throws Exception {
- try (final QueueOutputStream outputStream = new QueueOutputStream();
- final QueueInputStream inputStream = outputStream.newQueueInputStream()) {
- outputStream.write("ABC".getBytes(UTF_8));
- final String value = IOUtils.toString(inputStream, UTF_8);
- assertEquals("ABC", value);
- }
- }
-
- @Test
- public void writeStringMultiThread() throws Exception {
- try (final QueueOutputStream outputStream = callInThrowAwayThread(QueueOutputStream::new);
- final QueueInputStream inputStream = callInThrowAwayThread(outputStream::newQueueInputStream)) {
- callInThrowAwayThread(() -> {
- outputStream.write("ABC".getBytes(UTF_8));
+ private static <T> T callInThrowAwayThread(final Callable<T> callable) throws Exception {
+ final Exchanger<T> exchanger = new Exchanger<>();
+ executorService.submit(() -> {
+ final T value = callable.call();
+ exchanger.exchange(value);
return null;
});
-
- final String value = callInThrowAwayThread(() -> IOUtils.toString(inputStream, UTF_8));
- assertEquals("ABC", value);
+ return exchanger.exchange(null);
}
+
+ @Test
+ public void testNullArgument() {
+ assertThrows(NullPointerException.class, () -> new QueueOutputStream(null), "queue is required");
}
@Test
@@ -106,17 +97,26 @@ public void writeInterrupted() throws Exception {
}
@Test
- public void testNullArgument() {
- assertThrows(NullPointerException.class, () -> new QueueOutputStream(null), "queue is required");
+ public void writeString() throws Exception {
+ try (final QueueOutputStream outputStream = new QueueOutputStream();
+ final QueueInputStream inputStream = outputStream.newQueueInputStream()) {
+ outputStream.write("ABC".getBytes(UTF_8));
+ final String value = IOUtils.toString(inputStream, UTF_8);
+ assertEquals("ABC", value);
+ }
}
- private static <T> T callInThrowAwayThread(final Callable<T> callable) throws Exception {
- final Exchanger<T> exchanger = new Exchanger<>();
- executorService.submit(() -> {
- final T value = callable.call();
- exchanger.exchange(value);
+ @Test
+ public void writeStringMultiThread() throws Exception {
+ try (final QueueOutputStream outputStream = callInThrowAwayThread(QueueOutputStream::new);
+ final QueueInputStream inputStream = callInThrowAwayThread(outputStream::newQueueInputStream)) {
+ callInThrowAwayThread(() -> {
+ outputStream.write("ABC".getBytes(UTF_8));
return null;
});
- return exchanger.exchange(null);
+
+ final String value = callInThrowAwayThread(() -> IOUtils.toString(inputStream, UTF_8));
+ assertEquals("ABC", value);
+ }
}
}

Add QueueInputStream and QueueOutputStream as simpler alternatives to
PipedInputStream and PipedOutputStream #171.

Sort new methods.
Clean up Javadocs.