Add ability to specify wait time to QueueInputStream (#452)

* Revert "Add PollingQueueInputStream and TakingQueueInputStream to replace"

This reverts commit b21c2ac.

* Add option to specify waitTime to QueueInputStream

Author: maxxedev

src/changes/changes.xml

@@ -474,9 +474,6 @@ The <action> type attribute can be add,update,fix,remove.
       <action dev="ggregory" type="add" due-to="Gary Gregory">
         Add builders and avoid creating more constructors for all permutations of current options.
       </action>
-      <action dev="ggregory" type="add" due-to="Gary Gregory, maxxedev">
-        Add PollingQueueInputStream and TakingQueueInputStream to replace QueueInputStream. See also #447.
-      </action>
       <action dev="ggregory" type="add" due-to="Gary Gregory, maxxedev">
         Refactor using new Supplier API IOUtils.toString(IOSupplier&lt;InputStream&gt;, Charset[, IOSupplier&lt;String&gt;]).
       </action>

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

@@ -19,10 +19,14 @@
 import static org.apache.commons.io.IOUtils.EOF;
 
 import java.io.IOException;
+import java.io.InputStream;
 import java.io.PipedInputStream;
 import java.io.PipedOutputStream;
+import java.time.Duration;
+import java.util.Objects;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
 
 import org.apache.commons.io.output.QueueOutputStream;
 
@@ -52,36 +56,71 @@
  *
  * @see QueueOutputStream
  * @since 2.9.0
- * @deprecated Use {@link AbstractBlockingQueueInputStream.PollingQueueInputStream}
  */
-@Deprecated
-public class QueueInputStream extends AbstractBlockingQueueInputStream {
+public class QueueInputStream extends InputStream {
+
+    private final BlockingQueue<Integer> blockingQueue;
+    private final long waitTimeMillis;
 
     /**
-     * Constructs a new instance with no limit to its internal buffer size.
+     * Constructs a new instance with no limit to its internal buffer size and zero wait time.
      */
     public QueueInputStream() {
         this(new LinkedBlockingQueue<>());
     }
 
     /**
-     * Constructs a new instance with given queue.
+     * Constructs a new instance with given buffer and zero wait time.
      *
-     * @param blockingQueue backing queue for the stream.
+     * @param blockingQueue backing queue for the stream
      */
     public QueueInputStream(final BlockingQueue<Integer> blockingQueue) {
-        super(blockingQueue);
+        this(blockingQueue, Duration.ofMillis(0));
+    }
+
+    /**
+     * Constructs a new instance with given buffer and wait time.
+     *
+     * @param blockingQueue backing queue for the stream
+     * @param waitTime how long to wait if necessary for a queue element is available
+     * @since 2.12.0
+     */
+    public QueueInputStream(final BlockingQueue<Integer> blockingQueue, final Duration waitTime) {
+        this.blockingQueue = Objects.requireNonNull(blockingQueue, "blockingQueue");
+        Objects.requireNonNull(waitTime, "waitTime");
+        if (waitTime.toMillis() < 0) {
+            throw new IllegalArgumentException("waitTime must not be negative");
+        }
+        this.waitTimeMillis = waitTime.toMillis();
+    }
+
+    /**
+     * 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
+     */
+    public QueueOutputStream newQueueOutputStream() {
+        return new QueueOutputStream(blockingQueue);
     }
 
     /**
      * Reads and returns a single byte.
      *
-     * @return either the byte read or {@code -1} if the end of the stream has been reached
+     * @return either the byte read or {@code -1} if the wait time elapses before a queue element is available
+     * @throws IllegalStateException if thread is interrupted while waiting
      */
     @Override
     public int read() {
-        final Integer value = getBlockingQueue().poll();
-        return value == null ? EOF : 0xFF & value;
+        try {
+            final Integer value = blockingQueue.poll(waitTimeMillis, TimeUnit.MILLISECONDS);
+            return value == null ? EOF : 0xFF & value;
+        } catch (final InterruptedException e) {
+            Thread.currentThread().interrupt();
+            // throw runtime unchecked exception to maintain signature backward-compatibilty of
+            // this read method, which does not declare IOException
+            throw new IllegalStateException(e);
+        }
     }
 
 }