Skip to content

Commit fcb1ce5

Browse files
garydgregorygarydgregory
committed
Add IOUtils.skip[Fully](InputStream, long, Supplier<byte>)
1 parent 283c50b commit fcb1ce5

5 files changed

Lines changed: 268 additions & 16 deletions

File tree

‎src/changes/changes.xml‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -55,6 +55,9 @@ The <action> type attribute can be add,update,fix,remove.
5555
<action dev="ggregory" type="add" due-to="Gary Gregory">
5656
Add FileCleaningTracker.track(Path, Object[, FileDeleteStrategy]).
5757
</action>
58+
<action dev="ggregory" type="add" due-to="Gary Gregory">
59+
Add IOUtils.skip[Fully](InputStream, long, Supplier&lt;byte[]&gt;).
60+
</action>
5861
<!-- FIX -->
5962
<action dev="ggregory" type="fix" issue="IO-799" due-to="Jeroen van der Vegt, Gary Gregory">
6063
ReaderInputStream.read() throws an exception instead of returning -1 when called again after returning -1.

‎src/main/java/org/apache/commons/io/IOUtils.java‎

Lines changed: 65 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -52,8 +52,10 @@
5252
import java.util.List;
5353
import java.util.Objects;
5454
import java.util.function.Consumer;
55+
import java.util.function.Supplier;
5556
import java.util.stream.Collectors;
5657
import java.util.stream.Stream;
58+
import java.util.zip.InflaterInputStream;
5759

5860
import org.apache.commons.io.function.IOConsumer;
5961
import org.apache.commons.io.function.IOSupplier;
@@ -2375,6 +2377,36 @@ public static URL resourceToURL(final String name, final ClassLoader classLoader
23752377
* @since 2.0
23762378
*/
23772379
public static long skip(final InputStream input, final long toSkip) throws IOException {
2380+
return skip(input, toSkip, IOUtils::getScratchByteArrayWriteOnly);
2381+
}
2382+
2383+
/**
2384+
* Skips bytes from an input byte stream.
2385+
* <p>
2386+
* Intended for special cases when customization of the temporary buffer is needed because, for example, a nested input stream has requirements for the
2387+
* bytes read. For example, when using {@link InflaterInputStream}s from multiple threads.
2388+
* </p>
2389+
* <p>
2390+
* This implementation guarantees that it will read as many bytes as possible before giving up; this may not always be the case for skip() implementations
2391+
* in subclasses of {@link InputStream}.
2392+
* </p>
2393+
* <p>
2394+
* Note that the implementation uses {@link InputStream#read(byte[], int, int)} rather than delegating to {@link InputStream#skip(long)}. This means that
2395+
* the method may be considerably less efficient than using the actual skip implementation, this is done to guarantee that the correct number of bytes are
2396+
* skipped.
2397+
* </p>
2398+
*
2399+
* @param input byte stream to skip
2400+
* @param toSkip number of bytes to skip.
2401+
* @param skipBufferSupplier Supplies the buffer to use for reading.
2402+
* @return number of bytes actually skipped.
2403+
* @throws IOException if there is a problem reading the file
2404+
* @throws IllegalArgumentException if toSkip is negative
2405+
* @see InputStream#skip(long)
2406+
* @see <a href="https://issues.apache.org/jira/browse/IO-203">IO-203 - Add skipFully() method for InputStreams</a>
2407+
* @since 2.14.0
2408+
*/
2409+
public static long skip(final InputStream input, final long toSkip, final Supplier<byte[]> skipBufferSupplier) throws IOException {
23782410
if (toSkip < 0) {
23792411
throw new IllegalArgumentException("Skip count must be non-negative, actual: " + toSkip);
23802412
}
@@ -2385,9 +2417,9 @@ public static long skip(final InputStream input, final long toSkip) throws IOExc
23852417
//
23862418
long remain = toSkip;
23872419
while (remain > 0) {
2420+
final byte[] skipBuffer = skipBufferSupplier.get();
23882421
// See https://issues.apache.org/jira/browse/IO-203 for why we use read() rather than delegating to skip()
2389-
final byte[] byteArray = getScratchByteArrayWriteOnly();
2390-
final long n = input.read(byteArray, 0, (int) Math.min(remain, byteArray.length));
2422+
final long n = input.read(skipBuffer, 0, (int) Math.min(remain, skipBuffer.length));
23912423
if (n < 0) { // EOF
23922424
break;
23932425
}
@@ -2485,10 +2517,40 @@ public static long skip(final Reader reader, final long toSkip) throws IOExcepti
24852517
* @since 2.0
24862518
*/
24872519
public static void skipFully(final InputStream input, final long toSkip) throws IOException {
2520+
final long skipped = skip(input, toSkip, IOUtils::getScratchByteArrayWriteOnly);
2521+
if (skipped != toSkip) {
2522+
throw new EOFException("Bytes to skip: " + toSkip + " actual: " + skipped);
2523+
}
2524+
}
2525+
2526+
/**
2527+
* Skips the requested number of bytes or fail if there are not enough left.
2528+
* <p>
2529+
* Intended for special cases when customization of the temporary buffer is needed because, for example, a nested input stream has requirements for the
2530+
* bytes read. For example, when using {@link InflaterInputStream}s from multiple threads.
2531+
* </p>
2532+
* <p>
2533+
* This allows for the possibility that {@link InputStream#skip(long)} may not skip as many bytes as requested (most likely because of reaching EOF).
2534+
* </p>
2535+
* <p>
2536+
* Note that the implementation uses {@link #skip(InputStream, long)}. This means that the method may be considerably less efficient than using the actual
2537+
* skip implementation, this is done to guarantee that the correct number of characters are skipped.
2538+
* </p>
2539+
*
2540+
* @param input stream to skip
2541+
* @param toSkip the number of bytes to skip
2542+
* @param skipBufferSupplier Supplies the buffer to use for reading.
2543+
* @throws IOException if there is a problem reading the file
2544+
* @throws IllegalArgumentException if toSkip is negative
2545+
* @throws EOFException if the number of bytes skipped was incorrect
2546+
* @see InputStream#skip(long)
2547+
* @since 2.14.0
2548+
*/
2549+
public static void skipFully(final InputStream input, final long toSkip, final Supplier<byte[]> skipBufferSupplier) throws IOException {
24882550
if (toSkip < 0) {
24892551
throw new IllegalArgumentException("Bytes to skip must not be negative: " + toSkip);
24902552
}
2491-
final long skipped = skip(input, toSkip);
2553+
final long skipped = skip(input, toSkip, skipBufferSupplier);
24922554
if (skipped != toSkip) {
24932555
throw new EOFException("Bytes to skip: " + toSkip + " actual: " + skipped);
24942556
}

‎src/test/java/org/apache/commons/io/IOUtilsMultithreadedSkipTest.java‎

Lines changed: 147 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,147 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package org.apache.commons.io;
19+
20+
import static org.junit.jupiter.api.Assertions.assertEquals;
21+
import static org.junit.jupiter.api.Assertions.fail;
22+
23+
import java.io.ByteArrayInputStream;
24+
import java.io.ByteArrayOutputStream;
25+
import java.io.EOFException;
26+
import java.io.IOException;
27+
import java.io.InputStream;
28+
import java.util.Random;
29+
import java.util.concurrent.ExecutorCompletionService;
30+
import java.util.concurrent.ExecutorService;
31+
import java.util.concurrent.Executors;
32+
import java.util.concurrent.Future;
33+
import java.util.function.Supplier;
34+
import java.util.zip.Inflater;
35+
import java.util.zip.InflaterInputStream;
36+
37+
import org.junit.jupiter.api.BeforeEach;
38+
import org.junit.jupiter.api.Test;
39+
40+
/**
41+
* See Jira ticket IO-802.
42+
*/
43+
public class IOUtilsMultithreadedSkipTest {
44+
45+
private static final String FIXTURE = "TIKA-4065.bin";
46+
long seed = 1;
47+
private final ThreadLocal<byte[]> threadLocal = ThreadLocal.withInitial(() -> new byte[4096]);
48+
49+
private int[] generateExpected(final InputStream is, final int[] skips) throws IOException {
50+
final int[] testBytes = new int[skips.length];
51+
for (int i = 0; i < skips.length; i++) {
52+
try {
53+
IOUtils.skipFully(is, skips[i]);
54+
testBytes[i] = is.read();
55+
} catch (final EOFException e) {
56+
testBytes[i] = -1;
57+
}
58+
}
59+
return testBytes;
60+
}
61+
62+
private int[] generateSkips(final byte[] bytes, final int numSkips, final Random random) {
63+
final int[] skips = new int[numSkips];
64+
for (int i = 0; i < skips.length; i++) {
65+
skips[i] = random.nextInt(bytes.length / numSkips) + bytes.length / 10;
66+
}
67+
return skips;
68+
}
69+
70+
private InputStream inflate(final byte[] deflated) throws IOException {
71+
final ByteArrayOutputStream bos = new ByteArrayOutputStream();
72+
IOUtils.copy(new InflaterInputStream(new ByteArrayInputStream(deflated), new Inflater(true)), bos);
73+
return new ByteArrayInputStream(bos.toByteArray());
74+
}
75+
76+
@BeforeEach
77+
public void setUp() {
78+
// Not the best random we can use but good enough here.
79+
seed = new Random().nextLong();
80+
}
81+
82+
private void testSkipFullyOnInflaterInputStream(final Supplier<byte[]> baSupplier) throws Exception {
83+
final long thisSeed = seed;
84+
// thisSeed = -727624427837034313l;
85+
final Random random = new Random(thisSeed);
86+
final byte[] bytes;
87+
try (final InputStream inputStream = getClass().getResourceAsStream(FIXTURE)) {
88+
bytes = IOUtils.toByteArray(inputStream);
89+
}
90+
final int numSkips = (random.nextInt(bytes.length) / 100) + 1;
91+
92+
final int skips[] = generateSkips(bytes, numSkips, random);
93+
final int[] expected;
94+
try (final InputStream inflate = inflate(bytes)) {
95+
expected = generateExpected(inflate, skips);
96+
}
97+
98+
final int numThreads = 2;
99+
final int iterations = 100;
100+
final ExecutorService executorService = Executors.newFixedThreadPool(numThreads);
101+
final ExecutorCompletionService<Integer> executorCompletionService = new ExecutorCompletionService<>(executorService);
102+
103+
for (int i = 0; i < numThreads; i++) {
104+
executorCompletionService.submit(() -> {
105+
for (int iteration = 0; iteration < iterations; iteration++) {
106+
try (InputStream is = new InflaterInputStream(new ByteArrayInputStream(bytes), new Inflater(true))) {
107+
for (int skipIndex = 0; skipIndex < skips.length; skipIndex++) {
108+
try {
109+
IOUtils.skipFully(is, skips[skipIndex], baSupplier);
110+
final int c = is.read();
111+
assertEquals(expected[skipIndex], c, "failed on seed=" + seed + " iteration=" + iteration);
112+
} catch (final EOFException e) {
113+
assertEquals(expected[skipIndex], is.read(), "failed on " + "seed=" + seed + " iteration=" + iteration);
114+
}
115+
}
116+
}
117+
}
118+
return 1;
119+
});
120+
}
121+
122+
int finished = 0;
123+
while (finished < numThreads) {
124+
// blocking
125+
final Future<Integer> future = executorCompletionService.take();
126+
try {
127+
future.get();
128+
} catch (final Exception e) {
129+
// printStackTrace() for simpler debugging
130+
e.printStackTrace();
131+
fail("failed on seed=" + seed);
132+
}
133+
finished++;
134+
}
135+
}
136+
137+
@Test
138+
public void testSkipFullyOnInflaterInputStream_New_bytes() throws Exception {
139+
testSkipFullyOnInflaterInputStream(() -> new byte[4096]);
140+
}
141+
142+
@Test
143+
public void testSkipFullyOnInflaterInputStream_ThreadLocal() throws Exception {
144+
testSkipFullyOnInflaterInputStream(threadLocal::get);
145+
}
146+
147+
}

‎src/test/java/org/apache/commons/io/IOUtilsTest.java‎

Lines changed: 53 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -60,6 +60,7 @@
6060
import java.nio.file.Path;
6161
import java.util.Arrays;
6262
import java.util.List;
63+
import java.util.function.Supplier;
6364
import java.util.stream.Stream;
6465

6566
import org.apache.commons.io.function.IOConsumer;
@@ -1310,13 +1311,53 @@ public void testSkip_ReadableByteChannel() throws Exception {
13101311
public void testSkipFully_InputStream() throws Exception {
13111312
final int size = 1027;
13121313

1313-
final InputStream input = new ByteArrayInputStream(new byte[size]);
1314-
assertThrows(IllegalArgumentException.class, () -> IOUtils.skipFully(input, -1), "Should have failed with IllegalArgumentException");
1314+
try (final InputStream input = new ByteArrayInputStream(new byte[size])) {
1315+
assertThrows(IllegalArgumentException.class, () -> IOUtils.skipFully(input, -1), "Should have failed with IllegalArgumentException");
13151316

1316-
IOUtils.skipFully(input, 0);
1317-
IOUtils.skipFully(input, size - 1);
1318-
assertThrows(IOException.class, () -> IOUtils.skipFully(input, 2), "Should have failed with IOException");
1319-
IOUtils.closeQuietly(input);
1317+
IOUtils.skipFully(input, 0);
1318+
IOUtils.skipFully(input, size - 1);
1319+
assertThrows(IOException.class, () -> IOUtils.skipFully(input, 2), "Should have failed with IOException");
1320+
}
1321+
}
1322+
1323+
@Test
1324+
public void testSkipFully_InputStream_Buffer_New_bytes() throws Exception {
1325+
final int size = 1027;
1326+
final Supplier<byte[]> bas = () -> new byte[size];
1327+
try (final InputStream input = new ByteArrayInputStream(new byte[size])) {
1328+
assertThrows(IllegalArgumentException.class, () -> IOUtils.skipFully(input, -1, bas), "Should have failed with IllegalArgumentException");
1329+
1330+
IOUtils.skipFully(input, 0, bas);
1331+
IOUtils.skipFully(input, size - 1, bas);
1332+
assertThrows(IOException.class, () -> IOUtils.skipFully(input, 2, bas), "Should have failed with IOException");
1333+
}
1334+
}
1335+
1336+
@Test
1337+
public void testSkipFully_InputStream_Buffer_Resuse_bytes() throws Exception {
1338+
final int size = 1027;
1339+
final byte[] ba = new byte[size];
1340+
final Supplier<byte[]> bas = () -> ba;
1341+
try (final InputStream input = new ByteArrayInputStream(new byte[size])) {
1342+
assertThrows(IllegalArgumentException.class, () -> IOUtils.skipFully(input, -1, bas), "Should have failed with IllegalArgumentException");
1343+
1344+
IOUtils.skipFully(input, 0, bas);
1345+
IOUtils.skipFully(input, size - 1, bas);
1346+
assertThrows(IOException.class, () -> IOUtils.skipFully(input, 2, bas), "Should have failed with IOException");
1347+
}
1348+
}
1349+
1350+
@Test
1351+
public void testSkipFully_InputStream_Buffer_Resuse_ThreadLocal() throws Exception {
1352+
final int size = 1027;
1353+
final ThreadLocal<byte[]> tl = ThreadLocal.withInitial(() -> new byte[size]);
1354+
try (final InputStream input = new ByteArrayInputStream(new byte[size])) {
1355+
assertThrows(IllegalArgumentException.class, () -> IOUtils.skipFully(input, -1, tl::get), "Should have failed with IllegalArgumentException");
1356+
1357+
IOUtils.skipFully(input, 0, tl::get);
1358+
IOUtils.skipFully(input, size - 1, tl::get);
1359+
assertThrows(IOException.class, () -> IOUtils.skipFully(input, 2, tl::get), "Should have failed with IOException");
1360+
}
13201361
}
13211362

13221363
@Test
@@ -1336,13 +1377,12 @@ public void testSkipFully_ReadableByteChannel() throws Exception {
13361377
@Test
13371378
public void testSkipFully_Reader() throws Exception {
13381379
final int size = 1027;
1339-
final Reader input = new CharArrayReader(new char[size]);
1340-
1341-
IOUtils.skipFully(input, 0);
1342-
IOUtils.skipFully(input, size - 3);
1343-
assertThrows(IllegalArgumentException.class, () -> IOUtils.skipFully(input, -1), "Should have failed with IllegalArgumentException");
1344-
assertThrows(IOException.class, () -> IOUtils.skipFully(input, 5), "Should have failed with IOException");
1345-
IOUtils.closeQuietly(input);
1380+
try (final Reader input = new CharArrayReader(new char[size])) {
1381+
IOUtils.skipFully(input, 0);
1382+
IOUtils.skipFully(input, size - 3);
1383+
assertThrows(IllegalArgumentException.class, () -> IOUtils.skipFully(input, -1), "Should have failed with IllegalArgumentException");
1384+
assertThrows(IOException.class, () -> IOUtils.skipFully(input, 5), "Should have failed with IOException");
1385+
}
13461386
}
13471387

13481388
@Test

‎src/test/resources/org/apache/commons/io/TIKA-4065.bin‎

1.91 KB
Binary file not shown.

0 commit comments

Comments
 (0)