Skip to content

Commit 0a339fe

Browse files
committed
Apply nanoseconds precision for timeout duration.
Also improve javadocs
1 parent 7be5b59 commit 0a339fe

2 files changed

Lines changed: 15 additions & 13 deletions

File tree

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

Lines changed: 10 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -103,8 +103,8 @@ public Builder setBlockingQueue(final BlockingQueue<Integer> blockingQueue) {
103103
* @return this.
104104
*/
105105
public Builder setTimeout(final Duration timeout) {
106-
if (timeout != null && timeout.toMillis() < 0) {
107-
throw new IllegalArgumentException("waitTime must not be negative");
106+
if (timeout != null && timeout.toNanos() < 0) {
107+
throw new IllegalArgumentException("timeout must not be negative");
108108
}
109109
this.timeout = timeout != null ? timeout : Duration.ZERO;
110110
return this;
@@ -124,17 +124,17 @@ public static Builder builder() {
124124

125125
private final BlockingQueue<Integer> blockingQueue;
126126

127-
private final long timeoutMillis;
127+
private final long timeoutNanos;
128128

129129
/**
130-
* Constructs a new instance with no limit to its internal buffer size and zero wait time.
130+
* Constructs a new instance with no limit to its internal queue size and zero timeout.
131131
*/
132132
public QueueInputStream() {
133133
this(new LinkedBlockingQueue<>());
134134
}
135135

136136
/**
137-
* Constructs a new instance with given buffer and zero wait time.
137+
* Constructs a new instance with given queue and zero timeout.
138138
*
139139
* @param blockingQueue backing queue for the stream.
140140
*/
@@ -143,14 +143,14 @@ public QueueInputStream(final BlockingQueue<Integer> blockingQueue) {
143143
}
144144

145145
/**
146-
* Constructs a new instance with given buffer and wait time.
146+
* Constructs a new instance with given queue and timeout.
147147
*
148148
* @param blockingQueue backing queue for the stream.
149149
* @param timeout how long to wait before giving up when polling the queue.
150150
*/
151151
private QueueInputStream(final BlockingQueue<Integer> blockingQueue, final Duration timeout) {
152152
this.blockingQueue = Objects.requireNonNull(blockingQueue, "blockingQueue");
153-
this.timeoutMillis = Objects.requireNonNull(timeout, "timeout").toMillis();
153+
this.timeoutNanos = Objects.requireNonNull(timeout, "timeout").toNanos();
154154
}
155155

156156
/**
@@ -168,7 +168,7 @@ BlockingQueue<Integer> getBlockingQueue() {
168168
* @return the timeout duration.
169169
*/
170170
Duration getTimeout() {
171-
return Duration.ofMillis(timeoutMillis);
171+
return Duration.ofNanos(timeoutNanos);
172172
}
173173

174174
/**
@@ -183,13 +183,13 @@ public QueueOutputStream newQueueOutputStream() {
183183
/**
184184
* Reads and returns a single byte.
185185
*
186-
* @return either the byte read or {@code -1} if the wait time elapses before a queue element is available.
186+
* @return the byte read, or {@code -1} if a timeout occurs before a queue element is available.
187187
* @throws IllegalStateException if thread is interrupted while waiting.
188188
*/
189189
@Override
190190
public int read() {
191191
try {
192-
final Integer value = blockingQueue.poll(timeoutMillis, TimeUnit.MILLISECONDS);
192+
final Integer value = blockingQueue.poll(timeoutNanos, TimeUnit.NANOSECONDS);
193193
return value == null ? EOF : 0xFF & value;
194194
} catch (final InterruptedException e) {
195195
Thread.currentThread().interrupt();

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

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -32,7 +32,7 @@
3232
import java.util.concurrent.CountDownLatch;
3333
import java.util.concurrent.LinkedBlockingQueue;
3434
import java.util.concurrent.TimeUnit;
35-
import java.util.concurrent.atomic.AtomicReference;
35+
import java.util.concurrent.atomic.AtomicBoolean;
3636
import java.util.stream.Stream;
3737

3838
import org.apache.commons.io.IOUtils;
@@ -158,7 +158,7 @@ public void testTimeoutInterrupted() throws Exception {
158158
final QueueOutputStream outputStream = inputStream.newQueueOutputStream()) {
159159

160160
// read in a background thread
161-
final AtomicReference<Boolean> result = new AtomicReference<>();
161+
final AtomicBoolean result = new AtomicBoolean();
162162
final CountDownLatch latch = new CountDownLatch(1);
163163
final Thread thread = new Thread(() -> {
164164
// when thread is interrupted, verify ...
@@ -205,8 +205,10 @@ public void testUnbufferedReadWrite(final String inputData) throws IOException {
205205
@ParameterizedTest(name = "inputData={0}")
206206
@MethodSource("inputData")
207207
public void testUnbufferedReadWriteWithTimeout(final String inputData) throws IOException {
208-
try (QueueInputStream inputStream = QueueInputStream.builder().setTimeout(Duration.ofMinutes(2)).get();
208+
final Duration timeout = Duration.ofMinutes(2);
209+
try (QueueInputStream inputStream = QueueInputStream.builder().setTimeout(timeout).get();
209210
final QueueOutputStream outputStream = inputStream.newQueueOutputStream()) {
211+
assertEquals(timeout, inputStream.getTimeout());
210212
writeUnbuffered(outputStream, inputData);
211213
final String actualData = assertTimeout(Duration.ofSeconds(1), () -> readUnbuffered(inputStream, inputData.length()));
212214
assertEquals(inputData, actualData);

0 commit comments

Comments
 (0)