Skip to content

Commit 1dd8108

Browse files
garydgregorygarydgregory
committed
Add ThrottledInputStream.Builder.setMaxBytes(int, ChronoUnit)
- Update Javadoc example
1 parent d538098 commit 1dd8108

3 files changed

Lines changed: 189 additions & 34 deletions

File tree

‎src/changes/changes.xml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -49,6 +49,7 @@ The <action> type attribute can be add,update,fix,remove.
4949
<release version="2.18.1" date="YYYY-MM-DD" description="Version 2.18.1: Java 8 is required.">
5050
<!-- FIX -->
5151
<!-- ADD -->
52+
<action dev="ggregory" type="add" issue="IO-860" due-to="Nico Strecker, Gary Gregory">Add ThrottledInputStream.Builder.setMaxBytes(int, ChronoUnit).</action>
5253
<!-- UPDATE -->
5354
</release>
5455
<release version="2.18.0" date="2024-11-16" description="Version 2.18.0: Java 8 is required.">
@@ -63,7 +64,6 @@ The <action> type attribute can be add,update,fix,remove.
6364
<action dev="ggregory" type="fix" due-to="Éamonn McManus">Use Unicode escapes for superscript characters. #701.</action>
6465
<action dev="ggregory" type="fix" issue="IO-863" due-to="Éamonn McManus, Gary Gregory">Recent incompatible change to FileUtils.listFiles re extensions, see also IO-856.</action>
6566
<action dev="ggregory" type="fix" issue="IO-857" due-to="Dmitry, Gary Gregory">Javadoc: Update details for PathUtils "clean" behavior.</action>
66-
6767
<!-- ADD -->
6868
<action dev="ggregory" type="add" due-to="Gary Gregory">Add @FunctionalInterface to ClassNameMatcher.</action>
6969
<action dev="ggregory" type="add" due-to="Gary Gregory">Add ValidatingObjectInputStream.Builder and ValidatingObjectInputStream.builder().</action>

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

Lines changed: 90 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -22,15 +22,15 @@
2222
import java.io.InterruptedIOException;
2323
import java.time.Duration;
2424
import java.time.temporal.ChronoUnit;
25+
import java.util.Objects;
2526
import java.util.concurrent.TimeUnit;
2627

2728
/**
28-
* Provides bandwidth throttling on a specified InputStream. It is implemented as a wrapper on top of another InputStream instance. The throttling works by
29-
* examining the number of bytes read from the underlying InputStream from the beginning, and sleep()ing for a time interval if the byte-transfer is found
30-
* exceed the specified tolerable maximum. (Thus, while the read-rate might exceed the maximum for a short interval, the average tends towards the
31-
* specified maximum, overall.)
29+
* Provides bandwidth throttling on an InputStream as a filter input stream. The throttling examines the number of bytes read from the underlying InputStream,
30+
* and sleeps for a time interval if the byte-transfer is found to exceed the specified maximum rate. Thus, while the read-rate might exceed the maximum for a
31+
* short interval, the average tends towards the specified maximum, overall.
3232
* <p>
33-
* To build an instance, see {@link Builder}
33+
* To build an instance, call {@link #builder()}.
3434
* </p>
3535
* <p>
3636
* Inspired by Apache HBase's class of the same name.
@@ -49,22 +49,22 @@ public final class ThrottledInputStream extends CountingInputStream {
4949
* <pre>{@code
5050
* ThrottledInputStream in = ThrottledInputStream.builder()
5151
* .setPath(Paths.get("MyFile.xml"))
52-
* .setMaxBytesPerSecond(100_000)
52+
* .setMaxBytes(100_000, ChronoUnit.SECONDS)
5353
* .get();
5454
* }
5555
* </pre>
5656
* <h2>Using IO</h2>
5757
* <pre>{@code
5858
* ThrottledInputStream in = ThrottledInputStream.builder()
5959
* .setFile(new File("MyFile.xml"))
60-
* .setMaxBytesPerSecond(100_000)
60+
* .setMaxBytes(100_000, ChronoUnit.SECONDS)
6161
* .get();
6262
* }
6363
* </pre>
6464
* <pre>{@code
6565
* ThrottledInputStream in = ThrottledInputStream.builder()
6666
* .setInputStream(inputStream)
67-
* .setMaxBytesPerSecond(100_000)
67+
* .setMaxBytes(100_000, ChronoUnit.SECONDS)
6868
* .get();
6969
* }
7070
* </pre>
@@ -77,7 +77,7 @@ public static class Builder extends AbstractBuilder<ThrottledInputStream, Builde
7777
/**
7878
* Effectively not throttled.
7979
*/
80-
private long maxBytesPerSecond = Long.MAX_VALUE;
80+
private double maxBytesPerSecond = Double.MAX_VALUE;
8181

8282
/**
8383
* Builds a new {@link ThrottledInputStream}.
@@ -98,19 +98,87 @@ public static class Builder extends AbstractBuilder<ThrottledInputStream, Builde
9898
* @throws IOException if an I/O error occurs.
9999
* @see #getInputStream()
100100
*/
101-
@SuppressWarnings("resource")
102101
@Override
103102
public ThrottledInputStream get() throws IOException {
104103
return new ThrottledInputStream(this);
105104
}
106105

106+
// package private for testing.
107+
double getMaxBytesPerSecond() {
108+
return maxBytesPerSecond;
109+
}
110+
111+
/**
112+
* Sets the maximum bytes per time period unit.
113+
* <p>
114+
* For example, to throttle reading to 100K per second, use:
115+
* </p>
116+
* <pre>
117+
* builder.setMaxBytes(100_000, ChronoUnit.SECONDS)
118+
* </pre>
119+
* <p>
120+
* To test idle timeouts for example, use 1 byte per minute, 1 byte per 30 seconds, and so on.
121+
* </p>
122+
*
123+
* @param value the maximum bytes
124+
* @param chronoUnit a duration scale goal.
125+
* @return this instance.
126+
* @throws IllegalArgumentException Thrown if maxBytesPerSecond &lt;= 0.
127+
* @since 2.19.0
128+
*/
129+
public Builder setMaxBytes(final int value, final ChronoUnit chronoUnit) {
130+
setMaxBytes(value, chronoUnit.getDuration());
131+
return asThis();
132+
}
133+
134+
/**
135+
* Sets the maximum bytes per duration.
136+
* <p>
137+
* For example, to throttle reading to 100K per second, use:
138+
* </p>
139+
* <pre>
140+
* builder.setMaxBytes(100_000, Duration.ofSeconds(1))
141+
* </pre>
142+
* <p>
143+
* To test idle timeouts for example, use 1 byte per minute, 1 byte per 30 seconds, and so on.
144+
* </p>
145+
*
146+
* @param value the maximum bytes
147+
* @param duration a duration goal.
148+
* @return this instance.
149+
* @throws IllegalArgumentException Thrown if maxBytesPerSecond &lt;= 0.
150+
*/
151+
// Consider making public in the future
152+
Builder setMaxBytes(final int value, final Duration duration) {
153+
setMaxBytesPerSecond((double) Objects.requireNonNull(duration, "duration").toMillis() / 1_000 * value);
154+
return asThis();
155+
}
156+
107157
/**
108158
* Sets the maximum bytes per second.
109159
*
110160
* @param maxBytesPerSecond the maximum bytes per second.
161+
* @return this instance.
162+
* @throws IllegalArgumentException Thrown if maxBytesPerSecond &lt;= 0.
111163
*/
112-
public void setMaxBytesPerSecond(final long maxBytesPerSecond) {
164+
private Builder setMaxBytesPerSecond(final double maxBytesPerSecond) {
165+
if (maxBytesPerSecond <= 0) {
166+
throw new IllegalArgumentException("Bandwidth " + maxBytesPerSecond + " must be > 0.");
167+
}
113168
this.maxBytesPerSecond = maxBytesPerSecond;
169+
return asThis();
170+
}
171+
172+
/**
173+
* Sets the maximum bytes per second.
174+
*
175+
* @param maxBytesPerSecond the maximum bytes per second.
176+
* @throws IllegalArgumentException Thrown if maxBytesPerSecond &lt;= 0.
177+
*/
178+
public void setMaxBytesPerSecond(final long maxBytesPerSecond) {
179+
setMaxBytesPerSecond((double) maxBytesPerSecond);
180+
// TODO 3.0
181+
// return asThis();
114182
}
115183

116184
}
@@ -124,24 +192,22 @@ public static Builder builder() {
124192
return new Builder();
125193
}
126194

127-
static long toSleepMillis(final long bytesRead, final long maxBytesPerSec, final long elapsedMillis) {
128-
if (elapsedMillis < 0) {
129-
throw new IllegalArgumentException("The elapsed time should be greater or equal to zero");
130-
}
195+
// package private for testing
196+
static long toSleepMillis(final long bytesRead, final long elapsedMillis, final double maxBytesPerSec) {
131197
if (bytesRead <= 0 || maxBytesPerSec <= 0 || elapsedMillis == 0) {
132198
return 0;
133199
}
134200
// We use this class to load the single source file, so the bytesRead
135201
// and maxBytesPerSec aren't greater than Double.MAX_VALUE.
136202
// We can get the precise sleep time by using the double value.
137-
final long millis = (long) ((double) bytesRead / (double) maxBytesPerSec * 1000 - elapsedMillis);
203+
final long millis = (long) (bytesRead / maxBytesPerSec * 1000 - elapsedMillis);
138204
if (millis <= 0) {
139205
return 0;
140206
}
141207
return millis;
142208
}
143209

144-
private final long maxBytesPerSecond;
210+
private final double maxBytesPerSecond;
145211
private final long startTime = System.currentTimeMillis();
146212
private Duration totalSleepDuration = Duration.ZERO;
147213

@@ -171,15 +237,21 @@ private long getBytesPerSecond() {
171237
return getByteCount() / elapsedSeconds;
172238
}
173239

240+
// package private for testing.
241+
double getMaxBytesPerSecond() {
242+
return maxBytesPerSecond;
243+
}
244+
174245
private long getSleepMillis() {
175-
return toSleepMillis(getByteCount(), maxBytesPerSecond, System.currentTimeMillis() - startTime);
246+
return toSleepMillis(getByteCount(), System.currentTimeMillis() - startTime, maxBytesPerSecond);
176247
}
177248

178249
/**
179250
* Gets the total duration spent in sleep.
180251
*
181252
* @return Duration spent in sleep.
182253
*/
254+
// package private for testing
183255
Duration getTotalSleepDuration() {
184256
return totalSleepDuration;
185257
}

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

Lines changed: 98 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -18,15 +18,18 @@
1818
package org.apache.commons.io.input;
1919

2020
import static org.junit.jupiter.api.Assertions.assertEquals;
21+
import static org.junit.jupiter.api.Assertions.assertThrows;
2122
import static org.junit.jupiter.api.Assertions.assertThrowsExactly;
2223
import static org.junit.jupiter.api.Assertions.assertTrue;
2324

2425
import java.io.IOException;
2526
import java.io.InputStream;
2627
import java.time.Duration;
28+
import java.time.temporal.ChronoUnit;
2729
import java.util.concurrent.atomic.AtomicBoolean;
2830

2931
import org.apache.commons.io.IOUtils;
32+
import org.apache.commons.io.input.ThrottledInputStream.Builder;
3033
import org.apache.commons.io.test.CustomIOException;
3134
import org.junit.jupiter.api.Test;
3235

@@ -56,35 +59,115 @@ public void testAfterReadConsumer() throws Exception {
5659
// Throwing
5760
final String message = "test exception message";
5861
// @formatter:off
59-
try (InputStream bounded = ThrottledInputStream.builder()
62+
try (InputStream inputStream = ThrottledInputStream.builder()
6063
.setCharSequence("Hi")
6164
.setAfterRead(i -> {
6265
throw new CustomIOException(message);
6366
})
6467
.get()) {
65-
assertEquals(message, assertThrowsExactly(CustomIOException.class, () -> IOUtils.consume(bounded)).getMessage());
68+
assertEquals(message, assertThrowsExactly(CustomIOException.class, () -> IOUtils.consume(inputStream)).getMessage());
69+
}
70+
// @formatter:on
71+
}
72+
73+
@Test
74+
public void testBuilder() throws IOException {
75+
final Builder builder = ThrottledInputStream.builder();
76+
assertThrows(IllegalArgumentException.class, () -> builder.setMaxBytesPerSecond(-1));
77+
assertThrows(IllegalArgumentException.class, () -> builder.setMaxBytesPerSecond(0));
78+
assertThrows(IllegalArgumentException.class, () -> builder.setMaxBytes(1, Duration.ZERO.minusMillis(1)));
79+
assertThrows(IllegalArgumentException.class, () -> builder.setMaxBytes(1, Duration.ZERO));
80+
assertThrows(NullPointerException.class, () -> builder.setMaxBytes(1, (Duration) null));
81+
assertThrows(NullPointerException.class, () -> builder.setMaxBytes(1, (ChronoUnit) null));
82+
//
83+
// 2 bytes per second
84+
builder.setMaxBytesPerSecond(2);
85+
assertEquals(2.0, builder.getMaxBytesPerSecond());
86+
// @formatter:off
87+
try (ThrottledInputStream inputStream = builder
88+
.setInputStream(createOriginInputStream())
89+
.get()) {
90+
assertEquals(2.0, builder.getMaxBytesPerSecond());
91+
assertEquals(2.0, inputStream.getMaxBytesPerSecond());
92+
}
93+
try (ThrottledInputStream inputStream = builder
94+
.setInputStream(createOriginInputStream())
95+
.setMaxBytes(2, ChronoUnit.SECONDS)
96+
.get()) {
97+
assertEquals(2.0, builder.getMaxBytesPerSecond());
98+
assertEquals(2.0, inputStream.getMaxBytesPerSecond());
99+
}
100+
// @formatter:on
101+
Duration maxBytesPer = Duration.ofSeconds(1);
102+
// @formatter:off
103+
try (ThrottledInputStream inputStream = builder
104+
.setInputStream(createOriginInputStream())
105+
.setMaxBytes(2, maxBytesPer)
106+
.get()) {
107+
assertEquals(2.0, builder.getMaxBytesPerSecond());
108+
assertEquals(2.0, inputStream.getMaxBytesPerSecond());
109+
}
110+
//
111+
// 1 bytes per 1/2 second (30_000 millis)
112+
// @formatter:on
113+
maxBytesPer = maxBytesPer.dividedBy(2);
114+
// @formatter:off
115+
try (ThrottledInputStream inputStream = builder
116+
.setInputStream(createOriginInputStream())
117+
.setMaxBytes(1, maxBytesPer)
118+
.get()) {
119+
assertEquals(0.5, inputStream.getMaxBytesPerSecond());
120+
}
121+
// 1 byte/millis
122+
try (ThrottledInputStream inputStream = builder
123+
.setInputStream(createOriginInputStream())
124+
.setMaxBytes(1, ChronoUnit.MILLIS)
125+
.get()) {
126+
assertEquals(0.001, inputStream.getMaxBytesPerSecond());
127+
}
128+
// @formatter:on
129+
// 1 byte per 10_0011 millis.
130+
maxBytesPer = Duration.ofSeconds(20).plusMillis(11);
131+
// @formatter:off
132+
try (ThrottledInputStream inputStream = builder
133+
.setInputStream(createOriginInputStream())
134+
.setMaxBytes(1, maxBytesPer)
135+
.get()) {
136+
assertEquals(20.011, inputStream.getMaxBytesPerSecond());
137+
}
138+
// @formatter:on
139+
// Javadoc example
140+
// @formatter:off
141+
try (ThrottledInputStream inputStream = builder
142+
.setInputStream(createOriginInputStream())
143+
.setMaxBytes(100_000, ChronoUnit.SECONDS)
144+
.get()) {
145+
assertEquals(100_000.0, inputStream.getMaxBytesPerSecond());
66146
}
67147
// @formatter:on
68148
}
69149

70150
@Test
71151
public void testCalSleepTimeMs() {
72152
// case 0: initial - no read, no sleep
73-
assertEquals(0, ThrottledInputStream.toSleepMillis(0, 10_000, 1_000));
74-
153+
assertEquals(0, ThrottledInputStream.toSleepMillis(0, 1_000, 10_000));
75154
// case 1: no threshold
76-
assertEquals(0, ThrottledInputStream.toSleepMillis(Long.MAX_VALUE, 0, 1_000));
77-
assertEquals(0, ThrottledInputStream.toSleepMillis(Long.MAX_VALUE, -1, 1_000));
78-
155+
assertEquals(0, ThrottledInputStream.toSleepMillis(Long.MAX_VALUE, 1_000, 0));
156+
assertEquals(0, ThrottledInputStream.toSleepMillis(Long.MAX_VALUE, 1_000, -1));
79157
// case 2: too fast
80-
assertEquals(1500, ThrottledInputStream.toSleepMillis(5, 2, 1_000));
81-
assertEquals(500, ThrottledInputStream.toSleepMillis(5, 2, 2_000));
82-
assertEquals(6500, ThrottledInputStream.toSleepMillis(15, 2, 1_000));
83-
84-
// case 3: too slow
85-
assertEquals(0, ThrottledInputStream.toSleepMillis(1, 2, 1_000));
86-
assertEquals(0, ThrottledInputStream.toSleepMillis(2, 2, 2_000));
87-
assertEquals(0, ThrottledInputStream.toSleepMillis(1, 2, 1_000));
158+
assertEquals(1500, ThrottledInputStream.toSleepMillis(5, 1_000, 2));
159+
assertEquals(500, ThrottledInputStream.toSleepMillis(5, 2_000, 2));
160+
assertEquals(6500, ThrottledInputStream.toSleepMillis(15, 1_000, 2));
161+
assertEquals(4000, ThrottledInputStream.toSleepMillis(5, 1_000, 1));
162+
assertEquals(9000, ThrottledInputStream.toSleepMillis(5, 1_000, 0.5));
163+
assertEquals(99000, ThrottledInputStream.toSleepMillis(5, 1_000, 0.05));
164+
// case 3: too slow, no sleep needed
165+
assertEquals(0, ThrottledInputStream.toSleepMillis(1, 1_000, 2));
166+
assertEquals(0, ThrottledInputStream.toSleepMillis(2, 2_000, 2));
167+
assertEquals(0, ThrottledInputStream.toSleepMillis(1, 1_000, 2));
168+
assertEquals(0, ThrottledInputStream.toSleepMillis(1, 1_000, 2.0));
169+
assertEquals(0, ThrottledInputStream.toSleepMillis(1, 1_000, 1));
170+
assertEquals(0, ThrottledInputStream.toSleepMillis(1, 1_000, 1.0));
88171
}
89172

90173
@Test

0 commit comments

Comments
 (0)