From c1b019dfa10125b8e665b6044f77c3f2d65f075d Mon Sep 17 00:00:00 2001 From: jencymaryjoseph <35571282+jencymaryjoseph@users.noreply.github.com> Date: Fri, 28 Aug 2026 12:53:17 -0700 Subject: [PATCH] fix: prevent concatenated gzip response truncation when read via GZIPInputStream --- .../bugfix-AWSSDKforJavav2-c8ce5ba.json | 6 + .../awssdk/core/ResponseInputStream.java | 5 +- .../io/GzipAvailabilityInputStream.java | 164 +++++++++ .../awssdk/core/ResponseInputStreamTest.java | 261 ++++++++++++++ .../io/GzipAvailabilityInputStreamTest.java | 327 ++++++++++++++++++ 5 files changed, 761 insertions(+), 2 deletions(-) create mode 100644 .changes/next-release/bugfix-AWSSDKforJavav2-c8ce5ba.json create mode 100644 core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/io/GzipAvailabilityInputStream.java create mode 100644 core/sdk-core/src/test/java/software/amazon/awssdk/core/internal/io/GzipAvailabilityInputStreamTest.java diff --git a/.changes/next-release/bugfix-AWSSDKforJavav2-c8ce5ba.json b/.changes/next-release/bugfix-AWSSDKforJavav2-c8ce5ba.json new file mode 100644 index 000000000000..69ccf94f9839 --- /dev/null +++ b/.changes/next-release/bugfix-AWSSDKforJavav2-c8ce5ba.json @@ -0,0 +1,6 @@ +{ + "type": "bugfix", + "category": "AWS SDK for Java v2", + "contributor": "", + "description": "Fixed an issue where concatenated gzip response streams could be truncated to the first member when decoded with GZIPInputStream. A transient available()==0 at a gzip member boundary was treated as end of stream; ResponseInputStream now reports available() as at least 1 for gzip content while the stream is open." +} diff --git a/core/sdk-core/src/main/java/software/amazon/awssdk/core/ResponseInputStream.java b/core/sdk-core/src/main/java/software/amazon/awssdk/core/ResponseInputStream.java index 8f87186d0edd..f517f7993118 100644 --- a/core/sdk-core/src/main/java/software/amazon/awssdk/core/ResponseInputStream.java +++ b/core/sdk-core/src/main/java/software/amazon/awssdk/core/ResponseInputStream.java @@ -24,6 +24,7 @@ import java.util.concurrent.TimeUnit; import software.amazon.awssdk.annotations.SdkPublicApi; import software.amazon.awssdk.annotations.SdkTestInternalApi; +import software.amazon.awssdk.core.internal.io.GzipAvailabilityInputStream; import software.amazon.awssdk.core.io.SdkFilterInputStream; import software.amazon.awssdk.http.Abortable; import software.amazon.awssdk.http.AbortableInputStream; @@ -72,7 +73,7 @@ public ResponseInputStream(ResponseT resp, AbortableInputStream in) { } public ResponseInputStream(ResponseT resp, AbortableInputStream in, Duration timeout) { - super(in); + super(new GzipAvailabilityInputStream(in)); this.response = Validate.paramNotNull(resp, "response"); this.abortable = Validate.paramNotNull(in, "abortableInputStream"); @@ -85,7 +86,7 @@ public ResponseInputStream(ResponseT resp, InputStream in) { } public ResponseInputStream(ResponseT resp, InputStream in, Duration timeout) { - super(in); + super(new GzipAvailabilityInputStream(in)); this.response = Validate.paramNotNull(resp, "response"); this.abortable = in instanceof Abortable ? (Abortable) in : null; diff --git a/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/io/GzipAvailabilityInputStream.java b/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/io/GzipAvailabilityInputStream.java new file mode 100644 index 000000000000..b1b5b5cc6716 --- /dev/null +++ b/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/io/GzipAvailabilityInputStream.java @@ -0,0 +1,164 @@ +/* + * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"). + * You may not use this file except in compliance with the License. + * A copy of the License is located at + * + * http://aws.amazon.com/apache2.0 + * + * or in the "license" file accompanying this file. This file 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 software.amazon.awssdk.core.internal.io; + +import java.io.FilterInputStream; +import java.io.IOException; +import java.io.InputStream; +import software.amazon.awssdk.annotations.SdkInternalApi; +import software.amazon.awssdk.utils.IoUtils; + +/** + * Wraps a response body so {@code available()} never returns {@code 0} for gzip content while the stream is open. + * {@link java.util.zip.GZIPInputStream} treats a transient {@code 0} from {@code available()} at a member boundary + * as end of stream and stops, truncating concatenated gzip. Gzip is detected passively from the leading bytes + * ({@code 1f 8b 08}); non-gzip streams keep honest {@code available()}. + */ +@SdkInternalApi +public final class GzipAvailabilityInputStream extends FilterInputStream implements Releasable { + + private static final int GZIP_MAGIC_1 = 0x1f; + private static final int GZIP_MAGIC_2 = 0x8b; + private static final int GZIP_METHOD_DEFLATE = 0x08; + private static final int HEADER_LENGTH = 3; + + private final byte[] header = new byte[HEADER_LENGTH]; + private volatile int headerLen; + private volatile boolean classified; + private volatile boolean gzipDetected; + private volatile boolean eof; + private volatile boolean closed; + + private int markHeaderLen; + private boolean markClassified; + private boolean markGzipDetected; + private boolean markEof; + private boolean marked; + + public GzipAvailabilityInputStream(InputStream in) { + super(in); + } + + @Override + public int read() throws IOException { + int b = in.read(); + if (b == -1) { + eof = true; + } else { + if (eof) { + eof = false; + } + observe((byte) b); + } + return b; + } + + @Override + public int read(byte[] b, int off, int len) throws IOException { + int n = in.read(b, off, len); + if (n == -1) { + eof = true; + } else if (n > 0) { + if (eof) { + eof = false; + } + observe(b, off, n); + } + return n; + } + + @Override + public int available() throws IOException { + if (closed) { + return 0; + } + int available = in.available(); + return available == 0 && gzipDetected && !eof ? 1 : available; + } + + @Override + public long skip(long n) throws IOException { + long skipped = in.skip(n); + if (skipped > 0) { + classified = true; + } + return skipped; + } + + @Override + public synchronized void mark(int readlimit) { + markHeaderLen = headerLen; + markClassified = classified; + markGzipDetected = gzipDetected; + markEof = eof; + marked = true; + in.mark(readlimit); + } + + @Override + public synchronized void reset() throws IOException { + in.reset(); + if (marked) { + headerLen = markHeaderLen; + classified = markClassified; + gzipDetected = markGzipDetected; + eof = markEof; + } else { + headerLen = 0; + classified = false; + gzipDetected = false; + eof = false; + } + } + + @Override + public void close() throws IOException { + closed = true; + in.close(); + } + + @Override + public void release() { + IoUtils.closeQuietly(this, null); + if (in instanceof Releasable) { + ((Releasable) in).release(); + } + } + + private void observe(byte[] b, int off, int len) { + if (classified) { + return; + } + for (int i = 0; i < len && !classified; i++) { + observe(b[off + i]); + } + } + + private void observe(byte b) { + if (classified) { + return; + } + int len = headerLen; + header[len] = b; + headerLen = len + 1; + if (headerLen == HEADER_LENGTH) { + classified = true; + gzipDetected = (header[0] & 0xff) == GZIP_MAGIC_1 + && (header[1] & 0xff) == GZIP_MAGIC_2 + && (header[2] & 0xff) == GZIP_METHOD_DEFLATE; + } + } +} diff --git a/core/sdk-core/src/test/java/software/amazon/awssdk/core/ResponseInputStreamTest.java b/core/sdk-core/src/test/java/software/amazon/awssdk/core/ResponseInputStreamTest.java index 6710465a899d..75c0617cc707 100644 --- a/core/sdk-core/src/test/java/software/amazon/awssdk/core/ResponseInputStreamTest.java +++ b/core/sdk-core/src/test/java/software/amazon/awssdk/core/ResponseInputStreamTest.java @@ -17,12 +17,23 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatCode; +import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively; import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; +import java.io.BufferedReader; +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; import java.io.IOException; import java.io.InputStream; +import java.io.InputStreamReader; +import java.nio.charset.StandardCharsets; import java.time.Duration; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.zip.GZIPInputStream; +import java.util.zip.GZIPOutputStream; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.mockito.Mockito; @@ -133,7 +144,257 @@ void negativeTimeout_disablesTimeout() throws Exception { assertThat(responseInputStream.hasTimeoutTask()).isFalse(); } + @Test + void gzipConcatenatedMembers_whenAvailableTransientlyZero_decodesAllMembers() throws IOException { + InputStream underlying = new TrickleStream(concatenatedGzip("PART_ONE;", "PART_TWO;"), false); + ResponseInputStream ris = new ResponseInputStream<>(new Object(), underlying, Duration.ZERO); + + assertThat(readAllGzip(ris)).isEqualTo("PART_ONE;PART_TWO;"); + } + + @Test + void gzipManyMembers_whenAvailableTransientlyZero_decodesAllMembers() throws IOException { + InputStream underlying = new TrickleStream(concatenatedGzip("A;", "B;", "C;", "D;", "E;"), false); + ResponseInputStream ris = new ResponseInputStream<>(new Object(), underlying, Duration.ZERO); + + assertThat(readAllGzip(ris)).isEqualTo("A;B;C;D;E;"); + } + + @Test + void gzipSingleMember_whenAvailableZero_decodesWithoutHanging() { + // Preemptive timeout guards against the final-boundary probe hanging. + String decoded = assertTimeoutPreemptively(Duration.ofSeconds(5), () -> { + InputStream underlying = new TrickleStream(concatenatedGzip("ONLY_ONE_MEMBER;"), false); + ResponseInputStream ris = new ResponseInputStream<>(new Object(), underlying, Duration.ZERO); + return readAllGzip(ris); + }); + + assertThat(decoded).isEqualTo("ONLY_ONE_MEMBER;"); + } + + @Test + void gzipConcatenatedMembers_whenNeverZeroAvailable_decodesAllMembers() throws IOException { + InputStream underlying = new TrickleStream(concatenatedGzip("PART_ONE;", "PART_TWO;"), true); + ResponseInputStream ris = new ResponseInputStream<>(new Object(), underlying, Duration.ZERO); + + assertThat(readAllGzip(ris)).isEqualTo("PART_ONE;PART_TWO;"); + } + + @Test + void bufferedReader_whenNonGzip_deliversLineWithoutBlocking() { + ControllableStream underlying = new ControllableStream(); + underlying.feed("event: E1\n"); + ResponseInputStream ris = new ResponseInputStream<>(new Object(), underlying, Duration.ZERO); + BufferedReader reader = new BufferedReader(new InputStreamReader(ris, StandardCharsets.UTF_8)); + + assertTimeoutPreemptively(Duration.ofSeconds(2), () -> + assertThat(reader.readLine()).isEqualTo("event: E1")); + } + + @Test + void abort_whenGzipWrapperInserted_propagatesToOriginalAbortable() throws IOException { + AtomicBoolean aborted = new AtomicBoolean(false); + InputStream body = new TrickleStream(concatenatedGzip("HELLO"), false); + AbortableInputStream abortableBody = AbortableInputStream.create(body, () -> aborted.set(true)); + ResponseInputStream ris = new ResponseInputStream<>(new Object(), abortableBody, Duration.ZERO); + + ris.abort(); + + assertThat(aborted).isTrue(); + } + + @Test + void gzip_whenSourceBlocksThenSignalsEof_decodesWithoutHanging() { + // A genuinely blocking source: after the coerced available()==1 triggers a final-boundary probe, the read must + // still terminate once the source signals EOF rather than hang. + String decoded = assertTimeoutPreemptively(Duration.ofSeconds(5), () -> { + InputStream underlying = new BlockingEofStream(concatenatedGzip("ONLY_ONE;")); + ResponseInputStream ris = new ResponseInputStream<>(new Object(), underlying, Duration.ZERO); + return readAllGzip(ris); + }); + + assertThat(decoded).isEqualTo("ONLY_ONE;"); + } + + @Test + void markReset_throughWrapper_reReadsSameBytes() throws IOException { + InputStream body = new ByteArrayInputStream("hello-world".getBytes(StandardCharsets.UTF_8)); + ResponseInputStream ris = new ResponseInputStream<>(new Object(), body, Duration.ZERO); + + assertThat(ris.markSupported()).isTrue(); + ris.mark(16); + int first = ris.read(); + int second = ris.read(); + ris.reset(); + + assertThat(ris.read()).isEqualTo(first); + assertThat(ris.read()).isEqualTo(second); + } + private ResponseInputStream responseInputStream(Duration timeout) { return new ResponseInputStream<>(new Object(), abortableInputStream, timeout); } + + private static String readAllGzip(InputStream in) throws IOException { + try (GZIPInputStream gz = new GZIPInputStream(in)) { + ByteArrayOutputStream out = new ByteArrayOutputStream(); + byte[] buf = new byte[64]; + int n; + while ((n = gz.read(buf)) != -1) { + out.write(buf, 0, n); + } + return new String(out.toByteArray(), StandardCharsets.UTF_8); + } + } + + private static byte[] concatenatedGzip(String... members) throws IOException { + ByteArrayOutputStream out = new ByteArrayOutputStream(); + for (String member : members) { + ByteArrayOutputStream one = new ByteArrayOutputStream(); + try (GZIPOutputStream gz = new GZIPOutputStream(one)) { + gz.write(member.getBytes(StandardCharsets.UTF_8)); + } + out.write(one.toByteArray()); + } + return out.toByteArray(); + } + + /** Serves bytes one at a time; reports available()==0 unless {@code neverZero} */ + private static final class TrickleStream extends InputStream { + private final byte[] data; + private final boolean neverZero; + private int pos = 0; + + TrickleStream(byte[] data, boolean neverZero) { + this.data = data; + this.neverZero = neverZero; + } + + @Override + public int read() { + return pos < data.length ? (data[pos++] & 0xff) : -1; + } + + @Override + public int read(byte[] b, int off, int len) { + if (len == 0) { + return 0; + } + if (pos >= data.length) { + return -1; + } + b[off] = (byte) (data[pos++] & 0xff); + return 1; + } + + @Override + public int available() { + return neverZero ? 1 : 0; + } + } + + /** A blocking "live feed": read() waits for fed data or finish(); read(byte[]) returns only buffered bytes. */ + private static final class ControllableStream extends InputStream { + private final LinkedBlockingQueue queue = new LinkedBlockingQueue<>(); + private volatile boolean finished = false; + + void feed(String s) { + for (byte b : s.getBytes(StandardCharsets.UTF_8)) { + queue.add(b & 0xff); + } + } + + void finish() { + finished = true; + } + + @Override + public int read() throws IOException { + try { + Integer b; + while ((b = queue.poll(50, TimeUnit.MILLISECONDS)) == null) { + if (finished) { + return -1; + } + } + return b; + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IOException(e); + } + } + + @Override + public int read(byte[] b, int off, int len) throws IOException { + if (len == 0) { + return 0; + } + int first = read(); + if (first < 0) { + return -1; + } + b[off] = (byte) first; + int n = 1; + while (n < len) { + Integer next = queue.poll(); + if (next == null) { + break; + } + b[off + n] = (byte) (int) next; + n++; + } + return n; + } + + @Override + public int available() { + return queue.size(); + } + } + + /** Serves bytes one at a time (available()==0), then blocks briefly once before signalling EOF. */ + private static final class BlockingEofStream extends InputStream { + private final byte[] data; + private int pos = 0; + private boolean blocked = false; + + BlockingEofStream(byte[] data) { + this.data = data; + } + + @Override + public int read() throws IOException { + if (pos < data.length) { + return data[pos++] & 0xff; + } + if (!blocked) { + blocked = true; + try { + Thread.sleep(150); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IOException(e); + } + } + return -1; + } + + @Override + public int read(byte[] b, int off, int len) throws IOException { + if (len == 0) { + return 0; + } + int first = read(); + if (first < 0) { + return -1; + } + b[off] = (byte) first; + return 1; + } + + @Override + public int available() { + return 0; + } + } } diff --git a/core/sdk-core/src/test/java/software/amazon/awssdk/core/internal/io/GzipAvailabilityInputStreamTest.java b/core/sdk-core/src/test/java/software/amazon/awssdk/core/internal/io/GzipAvailabilityInputStreamTest.java new file mode 100644 index 000000000000..ec2757e628ce --- /dev/null +++ b/core/sdk-core/src/test/java/software/amazon/awssdk/core/internal/io/GzipAvailabilityInputStreamTest.java @@ -0,0 +1,327 @@ +/* + * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"). + * You may not use this file except in compliance with the License. + * A copy of the License is located at + * + * http://aws.amazon.com/apache2.0 + * + * or in the "license" file accompanying this file. This file 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 software.amazon.awssdk.core.internal.io; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.junit.jupiter.params.provider.Arguments.arguments; + +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.InputStream; +import java.nio.charset.StandardCharsets; +import java.util.stream.Stream; +import java.util.zip.GZIPOutputStream; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; + +/** Unit tests for {@link GzipAvailabilityInputStream}. */ +class GzipAvailabilityInputStreamTest { + + @ParameterizedTest + @MethodSource + void available_afterHeaderRead_returnsExpected(byte[] payload, int expected) throws IOException { + GzipAvailabilityInputStream stream = new GzipAvailabilityInputStream(new ZeroAvailableStream(payload)); + + stream.read(); + stream.read(); + stream.read(); + + assertThat(stream.available()).isEqualTo(expected); + } + + static Stream available_afterHeaderRead_returnsExpected() throws IOException { + return Stream.of( + arguments(gzip("HELLO"), 1), + arguments("event: E1\n".getBytes(StandardCharsets.UTF_8), 0), + arguments(new byte[] {(byte) 0x1f, (byte) 0x8b, 0x09, 0, 0}, 0)); // wrong method (09), not gzip + } + + @Test + void available_whenGzipButDelegateNonZero_returnsDelegateValue() throws IOException { + GzipAvailabilityInputStream stream = + new GzipAvailabilityInputStream(new FixedAvailableStream(gzip("HELLO"), 5)); + + stream.read(); + stream.read(); + stream.read(); + + assertThat(stream.available()).isEqualTo(5); + } + + @Test + void available_whenGzipAtEof_returnsZero() throws IOException { + GzipAvailabilityInputStream stream = new GzipAvailabilityInputStream(new ZeroAvailableStream(gzip("HI"))); + + while (stream.read() != -1) { + } + + assertThat(stream.available()).isEqualTo(0); + } + + @Test + void available_whenClosed_returnsZero() throws IOException { + GzipAvailabilityInputStream stream = + new GzipAvailabilityInputStream(new ZeroAvailableStream(gzip("HELLO"))); + + stream.read(); + stream.read(); + stream.read(); + assertThat(stream.available()).isEqualTo(1); + + stream.close(); + + assertThat(stream.available()).isEqualTo(0); + } + + @Test + void available_whenGzipDetectedViaBulkRead_returnsOne() throws IOException { + GzipAvailabilityInputStream stream = + new GzipAvailabilityInputStream(new BulkZeroStream(gzip("HELLO"))); + + stream.read(new byte[8], 0, 8); + + assertThat(stream.available()).isEqualTo(1); + } + + @Test + void available_whenPartialHeaderThenEof_returnsZero() throws IOException { + byte[] partial = {(byte) 0x1f, (byte) 0x8b}; + GzipAvailabilityInputStream stream = + new GzipAvailabilityInputStream(new ZeroAvailableStream(partial)); + + stream.read(); + stream.read(); + assertThat(stream.read()).isEqualTo(-1); + assertThat(stream.available()).isEqualTo(0); + } + + @Test + void read_whenZeroLengthAfterEof_keepsAvailableZero() throws IOException { + // A zero-length read after EOF returns 0 without moving the stream and must not clear EOF. + GzipAvailabilityInputStream stream = + new GzipAvailabilityInputStream(new ZeroAvailableStream(gzip("HELLO"))); + + while (stream.read() != -1) { + } + assertThat(stream.available()).isEqualTo(0); + + int n = stream.read(new byte[4], 0, 0); + + assertThat(n).isEqualTo(0); + assertThat(stream.available()).isEqualTo(0); + } + + @Test + void skip_whenBeforeClassification_abandonsGzipDetection() throws IOException { + // Junk prefix then a real gzip header: without abandoning detection, skipping the 2 junk bytes would + // expose 1f 8b 08 and be misdetected as gzip. + byte[] gz = gzip("HELLO"); + byte[] data = new byte[gz.length + 2]; + System.arraycopy(gz, 0, data, 2, gz.length); + GzipAvailabilityInputStream stream = new GzipAvailabilityInputStream(new ZeroAvailableStream(data)); + + stream.skip(2); + stream.read(); + stream.read(); + stream.read(); + + assertThat(stream.available()).isEqualTo(0); + } + + @Test + void reset_whenMarkedMidHeader_restoresClassification() throws IOException { + GzipAvailabilityInputStream stream = + new GzipAvailabilityInputStream(new MarkableZeroStream(gzip("HELLO"))); + + stream.mark(100); + stream.read(); + stream.read(); + stream.reset(); + + stream.read(); + stream.read(); + stream.read(); + + assertThat(stream.available()).isEqualTo(1); + } + + @Test + void reset_whenAfterEofWithoutMark_reDetectsGzip() throws IOException { + // reset() without a prior mark rewinds to position 0 (like ByteArrayInputStream, whose mark defaults to 0); + // the wrapper must clear its stale EOF/classification so gzip is re-detected on re-read. + GzipAvailabilityInputStream stream = + new GzipAvailabilityInputStream(new MarkableZeroStream(gzip("HELLO"))); + + while (stream.read() != -1) { + } + stream.reset(); + + stream.read(); + stream.read(); + stream.read(); + + assertThat(stream.available()).isEqualTo(1); + } + + @Test + void release_whenDelegateReleasable_propagates() { + ReleasableZeroStream delegate = new ReleasableZeroStream(); + GzipAvailabilityInputStream stream = new GzipAvailabilityInputStream(delegate); + + stream.release(); + + assertThat(delegate.released).isTrue(); + } + + private static byte[] gzip(String s) throws IOException { + ByteArrayOutputStream bos = new ByteArrayOutputStream(); + try (GZIPOutputStream g = new GZIPOutputStream(bos)) { + g.write(s.getBytes(StandardCharsets.UTF_8)); + } + return bos.toByteArray(); + } + + /** Serves bytes but always reports available()==0. */ + private static final class ZeroAvailableStream extends InputStream { + private final byte[] data; + private int pos; + + ZeroAvailableStream(byte[] data) { + this.data = data; + } + + @Override + public int read() { + return pos < data.length ? (data[pos++] & 0xff) : -1; + } + + @Override + public int available() { + return 0; + } + } + + /** {@link ZeroAvailableStream} that also supports mark/reset (mark defaults to 0). */ + private static final class MarkableZeroStream extends InputStream { + private final byte[] data; + private int pos; + private int markPos; + + MarkableZeroStream(byte[] data) { + this.data = data; + } + + @Override + public int read() { + return pos < data.length ? (data[pos++] & 0xff) : -1; + } + + @Override + public int available() { + return 0; + } + + @Override + public boolean markSupported() { + return true; + } + + @Override + public synchronized void mark(int readlimit) { + markPos = pos; + } + + @Override + public synchronized void reset() { + pos = markPos; + } + } + + /** Returns bytes in bulk (up to len per read) but reports available()==0. */ + private static final class BulkZeroStream extends InputStream { + private final byte[] data; + private int pos; + + BulkZeroStream(byte[] data) { + this.data = data; + } + + @Override + public int read() { + return pos < data.length ? (data[pos++] & 0xff) : -1; + } + + @Override + public int read(byte[] b, int off, int len) { + if (pos >= data.length) { + return -1; + } + int n = Math.min(len, data.length - pos); + System.arraycopy(data, pos, b, off, n); + pos += n; + return n; + } + + @Override + public int available() { + return 0; + } + } + + /** Records whether release() was called; its close() is a no-op. */ + private static final class ReleasableZeroStream extends InputStream implements Releasable { + private boolean released; + + @Override + public int read() { + return -1; + } + + @Override + public int available() { + return 0; + } + + @Override + public void release() { + released = true; + } + } + + /** Serves bytes but reports a fixed available() value. */ + private static final class FixedAvailableStream extends InputStream { + private final byte[] data; + private final int avail; + private int pos; + + FixedAvailableStream(byte[] data, int avail) { + this.data = data; + this.avail = avail; + } + + @Override + public int read() { + return pos < data.length ? (data[pos++] & 0xff) : -1; + } + + @Override + public int available() { + return avail; + } + } +}