| /* |
| * Licensed to the Apache Software Foundation (ASF) under one or more |
| * contributor license agreements. See the NOTICE file distributed with |
| * this work for additional information regarding copyright ownership. |
| * The ASF licenses this file to You under the Apache License, Version 2.0 |
| * (the "License"); you may not use this file except in compliance with |
| * the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software |
| * distributed under the License is distributed on an "AS IS" BASIS, |
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| * See the License for the specific language governing permissions and |
| * limitations under the License. |
| */ |
| |
| package org.apache.commons.io; |
| |
| import static org.junit.jupiter.api.Assertions.assertEquals; |
| import static org.junit.jupiter.api.Assertions.fail; |
| |
| import java.io.ByteArrayInputStream; |
| import java.io.ByteArrayOutputStream; |
| import java.io.EOFException; |
| import java.io.IOException; |
| import java.io.InputStream; |
| import java.util.Random; |
| import java.util.concurrent.ExecutorCompletionService; |
| import java.util.concurrent.ExecutorService; |
| import java.util.concurrent.Executors; |
| import java.util.concurrent.Future; |
| import java.util.function.Supplier; |
| import java.util.zip.Inflater; |
| import java.util.zip.InflaterInputStream; |
| |
| import org.junit.jupiter.api.BeforeEach; |
| import org.junit.jupiter.api.Test; |
| |
| /** |
| * See Jira ticket IO-802. |
| */ |
| public class IOUtilsMultithreadedSkipTest { |
| |
| private static final String FIXTURE = "TIKA-4065.bin"; |
| long seed = 1; |
| private final ThreadLocal<byte[]> threadLocal = ThreadLocal.withInitial(() -> new byte[4096]); |
| |
| private int[] generateExpected(final InputStream is, final int[] skips) throws IOException { |
| final int[] testBytes = new int[skips.length]; |
| for (int i = 0; i < skips.length; i++) { |
| try { |
| IOUtils.skipFully(is, skips[i]); |
| testBytes[i] = is.read(); |
| } catch (final EOFException e) { |
| testBytes[i] = -1; |
| } |
| } |
| return testBytes; |
| } |
| |
| private int[] generateSkips(final byte[] bytes, final int numSkips, final Random random) { |
| final int[] skips = new int[numSkips]; |
| for (int i = 0; i < skips.length; i++) { |
| skips[i] = random.nextInt(bytes.length / numSkips) + bytes.length / 10; |
| } |
| return skips; |
| } |
| |
| private InputStream inflate(final byte[] deflated) throws IOException { |
| final ByteArrayOutputStream bos = new ByteArrayOutputStream(); |
| IOUtils.copy(new InflaterInputStream(new ByteArrayInputStream(deflated), new Inflater(true)), bos); |
| return new ByteArrayInputStream(bos.toByteArray()); |
| } |
| |
| @BeforeEach |
| public void setUp() { |
| // Not the best random we can use but good enough here. |
| seed = new Random().nextLong(); |
| } |
| |
| private void testSkipFullyOnInflaterInputStream(final Supplier<byte[]> baSupplier) throws Exception { |
| final long thisSeed = seed; |
| // thisSeed = -727624427837034313l; |
| final Random random = new Random(thisSeed); |
| final byte[] bytes; |
| try (final InputStream inputStream = getClass().getResourceAsStream(FIXTURE)) { |
| bytes = IOUtils.toByteArray(inputStream); |
| } |
| final int numSkips = (random.nextInt(bytes.length) / 100) + 1; |
| |
| final int skips[] = generateSkips(bytes, numSkips, random); |
| final int[] expected; |
| try (final InputStream inflate = inflate(bytes)) { |
| expected = generateExpected(inflate, skips); |
| } |
| |
| final int numThreads = 2; |
| final int iterations = 100; |
| final ExecutorService executorService = Executors.newFixedThreadPool(numThreads); |
| final ExecutorCompletionService<Integer> executorCompletionService = new ExecutorCompletionService<>(executorService); |
| |
| for (int i = 0; i < numThreads; i++) { |
| executorCompletionService.submit(() -> { |
| for (int iteration = 0; iteration < iterations; iteration++) { |
| try (InputStream is = new InflaterInputStream(new ByteArrayInputStream(bytes), new Inflater(true))) { |
| for (int skipIndex = 0; skipIndex < skips.length; skipIndex++) { |
| try { |
| IOUtils.skipFully(is, skips[skipIndex], baSupplier); |
| final int c = is.read(); |
| assertEquals(expected[skipIndex], c, "failed on seed=" + seed + " iteration=" + iteration); |
| } catch (final EOFException e) { |
| assertEquals(expected[skipIndex], is.read(), "failed on " + "seed=" + seed + " iteration=" + iteration); |
| } |
| } |
| } |
| } |
| return 1; |
| }); |
| } |
| |
| int finished = 0; |
| while (finished < numThreads) { |
| // blocking |
| final Future<Integer> future = executorCompletionService.take(); |
| try { |
| future.get(); |
| } catch (final Exception e) { |
| // printStackTrace() for simpler debugging |
| e.printStackTrace(); |
| fail("failed on seed=" + seed); |
| } |
| finished++; |
| } |
| } |
| |
| @Test |
| public void testSkipFullyOnInflaterInputStream_New_bytes() throws Exception { |
| testSkipFullyOnInflaterInputStream(() -> new byte[4096]); |
| } |
| |
| @Test |
| public void testSkipFullyOnInflaterInputStream_ThreadLocal() throws Exception { |
| testSkipFullyOnInflaterInputStream(threadLocal::get); |
| } |
| |
| } |