Skip to content

Commit df770de

Browse files
garydgregorygarydgregory
committed
Use a builder instead of adding another constructor
Fix Javadoc Follow existing code conventions
1 parent d7a027c commit df770de

2 files changed

Lines changed: 138 additions & 83 deletions

File tree

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

Lines changed: 82 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -28,15 +28,16 @@
2828
import java.util.concurrent.LinkedBlockingQueue;
2929
import java.util.concurrent.TimeUnit;
3030

31+
import org.apache.commons.io.build.AbstractStreamBuilder;
3132
import org.apache.commons.io.output.QueueOutputStream;
3233

3334
/**
34-
* Simple alternative to JDK {@link java.io.PipedInputStream}; queue input stream provides what's written in queue
35-
* output stream.
35+
* Simple alternative to JDK {@link java.io.PipedInputStream}; queue input stream provides what's written in queue output stream.
3636
*
3737
* <p>
3838
* Example usage:
3939
* </p>
40+
*
4041
* <pre>
4142
* QueueInputStream inputStream = new QueueInputStream();
4243
* QueueOutputStream outputStream = inputStream.newQueueOutputStream();
@@ -45,22 +46,85 @@
4546
* inputStream.read();
4647
* </pre>
4748
* <p>
48-
* Unlike JDK {@link PipedInputStream} and {@link PipedOutputStream}, queue input/output streams may be used safely in a
49-
* single thread or multiple threads. Also, unlike JDK classes, no special meaning is attached to initial or current
50-
* thread. Instances can be used longer after initial threads exited.
49+
* Unlike JDK {@link PipedInputStream} and {@link PipedOutputStream}, queue input/output streams may be used safely in a single thread or multiple threads.
50+
* Also, unlike JDK classes, no special meaning is attached to initial or current thread. Instances can be used longer after initial threads exited.
5151
* </p>
5252
* <p>
53-
* Closing a {@link QueueInputStream} has no effect. The methods in this class can be called after the stream has been
54-
* closed without generating an {@link IOException}.
53+
* Closing a {@link QueueInputStream} has no effect. The methods in this class can be called after the stream has been closed without generating an
54+
* {@link IOException}.
5555
* </p>
5656
*
5757
* @see QueueOutputStream
5858
* @since 2.9.0
5959
*/
6060
public class QueueInputStream extends InputStream {
6161

62+
/**
63+
* Builds a new {@link QueueInputStream} instance.
64+
* <p>
65+
* For example:
66+
* </p>
67+
*
68+
* <pre>{@code
69+
* QueueInputStream s = QueueInputStream.builder()
70+
* .setBlockingQueue(new LinkedBlockingQueue<>())
71+
* .setTimeout(Duration.ZERO)
72+
* .get()}
73+
* </pre>
74+
* <p>
75+
*
76+
* @since 2.12.0
77+
*/
78+
public static class Builder extends AbstractStreamBuilder<QueueInputStream, Builder> {
79+
80+
private BlockingQueue<Integer> blockingQueue = new LinkedBlockingQueue<>();
81+
private Duration timeout = Duration.ZERO;
82+
83+
@Override
84+
public QueueInputStream get() throws IOException {
85+
return new QueueInputStream(blockingQueue, timeout);
86+
}
87+
88+
/**
89+
* Sets backing queue for the stream.
90+
*
91+
* @param blockingQueue backing queue for the stream.
92+
* @return this
93+
*/
94+
public Builder setBlockingQueue(final BlockingQueue<Integer> blockingQueue) {
95+
this.blockingQueue = blockingQueue != null ? blockingQueue : new LinkedBlockingQueue<>();
96+
return this;
97+
}
98+
99+
/**
100+
* Sets the polling timeout.
101+
*
102+
* @param timeout the polling timeout.
103+
* @return this.
104+
*/
105+
public Builder setTimeout(final Duration timeout) {
106+
if (timeout != null && timeout.toMillis() < 0) {
107+
throw new IllegalArgumentException("waitTime must not be negative");
108+
}
109+
this.timeout = timeout != null ? timeout : Duration.ZERO;
110+
return this;
111+
}
112+
113+
}
114+
115+
/**
116+
* Constructs a new {@link Builder}.
117+
*
118+
* @return a new {@link Builder}.
119+
* @since 2.12.0
120+
*/
121+
public static Builder builder() {
122+
return new Builder();
123+
}
124+
62125
private final BlockingQueue<Integer> blockingQueue;
63-
private final long waitTimeMillis;
126+
127+
private final long timeoutMillis;
64128

65129
/**
66130
* Constructs a new instance with no limit to its internal buffer size and zero wait time.
@@ -72,7 +136,7 @@ public QueueInputStream() {
72136
/**
73137
* Constructs a new instance with given buffer and zero wait time.
74138
*
75-
* @param blockingQueue backing queue for the stream
139+
* @param blockingQueue backing queue for the stream.
76140
*/
77141
public QueueInputStream(final BlockingQueue<Integer> blockingQueue) {
78142
this(blockingQueue, Duration.ZERO);
@@ -81,24 +145,18 @@ public QueueInputStream(final BlockingQueue<Integer> blockingQueue) {
81145
/**
82146
* Constructs a new instance with given buffer and wait time.
83147
*
84-
* @param blockingQueue backing queue for the stream
85-
* @param waitTime how long to wait if necessary for a queue element is available
86-
* @since 2.12.0
148+
* @param blockingQueue backing queue for the stream.
149+
* @param timeout how long to wait before giving up when polling the queue.
87150
*/
88-
public QueueInputStream(final BlockingQueue<Integer> blockingQueue, final Duration waitTime) {
151+
private QueueInputStream(final BlockingQueue<Integer> blockingQueue, final Duration timeout) {
89152
this.blockingQueue = Objects.requireNonNull(blockingQueue, "blockingQueue");
90-
Objects.requireNonNull(waitTime, "waitTime");
91-
if (waitTime.toMillis() < 0) {
92-
throw new IllegalArgumentException("waitTime must not be negative");
93-
}
94-
this.waitTimeMillis = waitTime.toMillis();
153+
this.timeoutMillis = Objects.requireNonNull(timeout, "timeout").toMillis();
95154
}
96155

97156
/**
98-
* Creates a new QueueOutputStream instance connected to this. Writes to the output stream will be visible to this
99-
* input stream.
157+
* Creates a new QueueOutputStream instance connected to this. Writes to the output stream will be visible to this input stream.
100158
*
101-
* @return QueueOutputStream connected to this stream
159+
* @return QueueOutputStream connected to this stream.
102160
*/
103161
public QueueOutputStream newQueueOutputStream() {
104162
return new QueueOutputStream(blockingQueue);
@@ -107,13 +165,13 @@ public QueueOutputStream newQueueOutputStream() {
107165
/**
108166
* Reads and returns a single byte.
109167
*
110-
* @return either the byte read or {@code -1} if the wait time elapses before a queue element is available
111-
* @throws IllegalStateException if thread is interrupted while waiting
168+
* @return either the byte read or {@code -1} if the wait time elapses before a queue element is available.
169+
* @throws IllegalStateException if thread is interrupted while waiting.
112170
*/
113171
@Override
114172
public int read() {
115173
try {
116-
final Integer value = blockingQueue.poll(waitTimeMillis, TimeUnit.MILLISECONDS);
174+
final Integer value = blockingQueue.poll(timeoutMillis, TimeUnit.MILLISECONDS);
117175
return value == null ? EOF : 0xFF & value;
118176
} catch (final InterruptedException e) {
119177
Thread.currentThread().interrupt();

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

Lines changed: 56 additions & 59 deletions
Original file line numberDiff line numberDiff line change
@@ -16,23 +16,11 @@
1616
*/
1717
package org.apache.commons.io.input;
1818

19-
import static java.nio.charset.StandardCharsets.UTF_8;
2019
import static org.junit.jupiter.api.Assertions.assertEquals;
2120
import static org.junit.jupiter.api.Assertions.assertThrows;
2221
import static org.junit.jupiter.api.Assertions.assertTimeout;
2322
import static org.junit.jupiter.api.Assertions.assertTrue;
2423

25-
import com.google.common.base.Stopwatch;
26-
import org.apache.commons.io.IOUtils;
27-
import org.apache.commons.io.output.QueueOutputStream;
28-
import org.apache.commons.io.output.QueueOutputStreamTest;
29-
import org.apache.commons.lang3.StringUtils;
30-
import org.junit.jupiter.api.DisplayName;
31-
import org.junit.jupiter.api.Test;
32-
import org.junit.jupiter.params.ParameterizedTest;
33-
import org.junit.jupiter.params.provider.Arguments;
34-
import org.junit.jupiter.params.provider.MethodSource;
35-
3624
import java.io.BufferedInputStream;
3725
import java.io.BufferedOutputStream;
3826
import java.io.ByteArrayOutputStream;
@@ -47,10 +35,22 @@
4735
import java.util.concurrent.atomic.AtomicReference;
4836
import java.util.stream.Stream;
4937

38+
import org.apache.commons.io.IOUtils;
39+
import org.apache.commons.io.output.QueueOutputStream;
40+
import org.apache.commons.io.output.QueueOutputStreamTest;
41+
import org.apache.commons.lang3.StringUtils;
42+
import org.junit.jupiter.api.DisplayName;
43+
import org.junit.jupiter.api.Test;
44+
import org.junit.jupiter.params.ParameterizedTest;
45+
import org.junit.jupiter.params.provider.Arguments;
46+
import org.junit.jupiter.params.provider.MethodSource;
47+
48+
import com.google.common.base.Stopwatch;
49+
5050
/**
5151
* Test {@link QueueInputStream}.
5252
*
53-
* @see {@link QueueOutputStreamTest}
53+
* @see QueueOutputStreamTest
5454
*/
5555
public class QueueInputStreamTest {
5656

@@ -71,80 +71,78 @@ public static Stream<Arguments> inputData() {
7171
// @formatter:on
7272
}
7373

74+
private int defaultBufferSize() {
75+
return 8192;
76+
}
77+
78+
private String readUnbuffered(final InputStream inputStream) throws IOException {
79+
return readUnbuffered(inputStream, Integer.MAX_VALUE);
80+
}
81+
82+
private String readUnbuffered(final InputStream inputStream, final int maxBytes) throws IOException {
83+
if (maxBytes == 0) {
84+
return "";
85+
}
86+
87+
final ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
88+
int n = -1;
89+
while ((n = inputStream.read()) != -1) {
90+
byteArrayOutputStream.write(n);
91+
if (byteArrayOutputStream.size() >= maxBytes) {
92+
break;
93+
}
94+
}
95+
return byteArrayOutputStream.toString(StandardCharsets.UTF_8.name());
96+
}
97+
7498
@ParameterizedTest(name = "inputData={0}")
7599
@MethodSource("inputData")
76-
public void bufferedReads(final String inputData) throws IOException {
100+
public void testBufferedReads(final String inputData) throws IOException {
77101
final BlockingQueue<Integer> queue = new LinkedBlockingQueue<>();
78102
try (BufferedInputStream inputStream = new BufferedInputStream(new QueueInputStream(queue));
79103
final QueueOutputStream outputStream = new QueueOutputStream(queue)) {
80-
outputStream.write(inputData.getBytes(UTF_8));
81-
final String actualData = IOUtils.toString(inputStream, UTF_8);
104+
outputStream.write(inputData.getBytes(StandardCharsets.UTF_8));
105+
final String actualData = IOUtils.toString(inputStream, StandardCharsets.UTF_8);
82106
assertEquals(inputData, actualData);
83107
}
84108
}
85109

86110
@ParameterizedTest(name = "inputData={0}")
87111
@MethodSource("inputData")
88-
public void bufferedReadWrite(final String inputData) throws IOException {
112+
public void testBufferedReadWrite(final String inputData) throws IOException {
89113
final BlockingQueue<Integer> queue = new LinkedBlockingQueue<>();
90114
try (BufferedInputStream inputStream = new BufferedInputStream(new QueueInputStream(queue));
91115
final BufferedOutputStream outputStream = new BufferedOutputStream(new QueueOutputStream(queue), defaultBufferSize())) {
92-
outputStream.write(inputData.getBytes(UTF_8));
116+
outputStream.write(inputData.getBytes(StandardCharsets.UTF_8));
93117
outputStream.flush();
94-
final String dataCopy = IOUtils.toString(inputStream, UTF_8);
118+
final String dataCopy = IOUtils.toString(inputStream, StandardCharsets.UTF_8);
95119
assertEquals(inputData, dataCopy);
96120
}
97121
}
98122

99123
@ParameterizedTest(name = "inputData={0}")
100124
@MethodSource("inputData")
101-
public void bufferedWrites(final String inputData) throws IOException {
125+
public void testBufferedWrites(final String inputData) throws IOException {
102126
final BlockingQueue<Integer> queue = new LinkedBlockingQueue<>();
103127
try (QueueInputStream inputStream = new QueueInputStream(queue);
104128
final BufferedOutputStream outputStream = new BufferedOutputStream(new QueueOutputStream(queue), defaultBufferSize())) {
105-
outputStream.write(inputData.getBytes(UTF_8));
129+
outputStream.write(inputData.getBytes(StandardCharsets.UTF_8));
106130
outputStream.flush();
107131
final String actualData = readUnbuffered(inputStream);
108132
assertEquals(inputData, actualData);
109133
}
110134
}
111135

112-
private int defaultBufferSize() {
113-
return 8192;
114-
}
115-
116136
@Test
117-
public void invalidArguments() {
137+
public void testInvalidArguments() {
118138
assertThrows(NullPointerException.class, () -> new QueueInputStream(null), "queue is required");
119-
assertThrows(NullPointerException.class, () -> new QueueInputStream(new LinkedBlockingQueue<>(), null), "waitTime is required");
120-
assertThrows(IllegalArgumentException.class, () -> new QueueInputStream(new LinkedBlockingQueue<>(), Duration.ofMillis(-1)),
121-
"waitTime must not be negative");
122-
}
123-
124-
private String readUnbuffered(final InputStream inputStream) throws IOException {
125-
return readUnbuffered(inputStream, Integer.MAX_VALUE);
126-
}
127-
128-
private String readUnbuffered(final InputStream inputStream, final int maxBytes) throws IOException {
129-
if (maxBytes == 0) {
130-
return "";
131-
}
132-
133-
final ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
134-
int n = -1;
135-
while ((n = inputStream.read()) != -1) {
136-
byteArrayOutputStream.write(n);
137-
if (byteArrayOutputStream.size() >= maxBytes) {
138-
break;
139-
}
140-
}
141-
return byteArrayOutputStream.toString(StandardCharsets.UTF_8.name());
139+
assertThrows(IllegalArgumentException.class, () -> QueueInputStream.builder().setTimeout(Duration.ofMillis(-1)).get(), "waitTime must not be negative");
142140
}
143141

144142
@Test
145143
@DisplayName("If read is interrupted while waiting, then exception is thrown")
146-
public void timeoutInterrupted() throws Exception {
147-
try (QueueInputStream inputStream = new QueueInputStream(new LinkedBlockingQueue<>(), Duration.ofMinutes(2));
144+
public void testTimeoutInterrupted() throws Exception {
145+
try (QueueInputStream inputStream = QueueInputStream.builder().setTimeout(Duration.ofMinutes(2)).get();
148146
final QueueOutputStream outputStream = inputStream.newQueueOutputStream()) {
149147

150148
// read in a background thread
@@ -169,22 +167,21 @@ public void timeoutInterrupted() throws Exception {
169167

170168
@Test
171169
@DisplayName("If data is not available in queue, then read will wait until wait time elapses")
172-
public void timeoutUnavailableData() throws IOException {
173-
try (QueueInputStream inputStream = new QueueInputStream(new LinkedBlockingQueue<>(), Duration.ofMillis(500));
170+
public void testTimeoutUnavailableData() throws IOException {
171+
try (QueueInputStream inputStream = QueueInputStream.builder().setTimeout(Duration.ofMillis(500)).get();
174172
final QueueOutputStream outputStream = inputStream.newQueueOutputStream()) {
175-
176173
final Stopwatch stopwatch = Stopwatch.createStarted();
177174
final String actualData = assertTimeout(Duration.ofSeconds(1), () -> readUnbuffered(inputStream, 3));
178175
stopwatch.stop();
179176
assertEquals("", actualData);
180177

181-
assertTrue(stopwatch.elapsed(TimeUnit.MILLISECONDS) >= 500);
178+
assertTrue(stopwatch.elapsed(TimeUnit.MILLISECONDS) >= 500, () -> stopwatch.toString());
182179
}
183180
}
184181

185182
@ParameterizedTest(name = "inputData={0}")
186183
@MethodSource("inputData")
187-
public void unbufferedReadWrite(final String inputData) throws IOException {
184+
public void testUnbufferedReadWrite(final String inputData) throws IOException {
188185
try (QueueInputStream inputStream = new QueueInputStream();
189186
final QueueOutputStream outputStream = inputStream.newQueueOutputStream()) {
190187
writeUnbuffered(outputStream, inputData);
@@ -195,8 +192,8 @@ public void unbufferedReadWrite(final String inputData) throws IOException {
195192

196193
@ParameterizedTest(name = "inputData={0}")
197194
@MethodSource("inputData")
198-
public void unbufferedReadWriteWithTimeout(final String inputData) throws IOException {
199-
try (QueueInputStream inputStream = new QueueInputStream(new LinkedBlockingQueue<>(), Duration.ofMinutes(2));
195+
public void testUnbufferedReadWriteWithTimeout(final String inputData) throws IOException {
196+
try (QueueInputStream inputStream = QueueInputStream.builder().setTimeout(Duration.ofMinutes(2)).get();
200197
final QueueOutputStream outputStream = inputStream.newQueueOutputStream()) {
201198
writeUnbuffered(outputStream, inputData);
202199
final String actualData = assertTimeout(Duration.ofSeconds(1), () -> readUnbuffered(inputStream, inputData.length()));
@@ -205,7 +202,7 @@ public void unbufferedReadWriteWithTimeout(final String inputData) throws IOExce
205202
}
206203

207204
private void writeUnbuffered(final QueueOutputStream outputStream, final String inputData) throws IOException {
208-
final byte[] bytes = inputData.getBytes(UTF_8);
205+
final byte[] bytes = inputData.getBytes(StandardCharsets.UTF_8);
209206
outputStream.write(bytes, 0, bytes.length);
210207
}
211208
}

0 commit comments

Comments
 (0)