CloseableIteratorBufferAsInputStream: fix CLOSED marker instance (#…

…3086)

Motivation:

`CloseableIteratorBufferAsInputStream` uses `CLOSED` as a marker for
`isClosed()`. However, `fromAscii("")` returns a static reference to
`EmptyBuffer`, which opens a risk of getting such `Buffer` via
underlying iterator.

Modifications:
- Use `wrap` method that guarantees allocation of a new unique instance
of a `Buffer`;

Result:

`CLOSED` marker is guaranteed to be a unique instance.

Additional cleanup of `AbstractCloseableIteratorAsInputStream`:
- Remove `len - toRead` that always results in `0`;
- Avoid arithmetic if `initialLen == len`;

servicetalk-concurrent-api-internal/src/main/java/io/servicetalk/concurrent/api/internal/CloseableIteratorBufferAsInputStream.java

@@ -21,6 +21,7 @@
 
 import java.io.IOException;
 import java.io.InputStream;
+import java.nio.ByteBuffer;
 import javax.annotation.Nullable;
 
 import static io.servicetalk.buffer.api.ReadOnlyBufferAllocators.DEFAULT_RO_ALLOCATOR;
@@ -29,7 +30,8 @@
  * Conversion from a {@link CloseableIterator} of {@link Buffer}s to a {@link InputStream}.
  */
 public final class CloseableIteratorBufferAsInputStream extends AbstractCloseableIteratorAsInputStream<Buffer> {
-    private static final Buffer CLOSED = DEFAULT_RO_ALLOCATOR.fromAscii("");
+    // Use `wrap` instead of `fromAscii("")` to avoid getting a reference to EmptyBuffer constant
+    private static final Buffer CLOSED = DEFAULT_RO_ALLOCATOR.wrap(ByteBuffer.allocate(0).asReadOnlyBuffer());
     @Nullable
     private Buffer leftover;
 

servicetalk-concurrent-api-internal/src/test/java/io/servicetalk/concurrent/api/internal/CloseableIteratorBufferAsInputStreamTest.java

@@ -23,6 +23,7 @@
 import java.io.IOException;
 import java.io.InputStream;
 
+import static io.servicetalk.buffer.api.EmptyBuffer.EMPTY_BUFFER;
 import static io.servicetalk.buffer.api.ReadOnlyBufferAllocators.DEFAULT_RO_ALLOCATOR;
 import static io.servicetalk.buffer.netty.BufferAllocators.DEFAULT_ALLOCATOR;
 import static io.servicetalk.concurrent.internal.BlockingIterables.from;
@@ -248,6 +249,19 @@ void testNullAndEmptyIteratorValues() throws IOException {
         }
     }
 
+    @Test
+    void emptyBufferFollowedByData() throws IOException {
+        Buffer src = DEFAULT_RO_ALLOCATOR.fromAscii("1234");
+        try (InputStream stream = new CloseableIteratorBufferAsInputStream(
+                from(asList(EMPTY_BUFFER, src.duplicate())).iterator())) {
+            byte[] data = new byte[4];
+            int read = stream.read(data, 0, 4);
+            assertThat("Unexpected number of bytes read.", read, is(4));
+            assertThat("Unexpected bytes read.", bytesToBuffer(data, read), equalTo(src));
+            assertThat("Bytes read after complete.", stream.read(), is(-1));
+        }
+    }
+
     private static CloseableIterator<Buffer> createIterator(Buffer src) {
         return from(singletonList(src.duplicate())).iterator();
     }

servicetalk-concurrent-internal/src/main/java/io/servicetalk/concurrent/internal/AbstractCloseableIteratorAsInputStream.java

@@ -111,7 +111,7 @@ public final int read(final byte[] b, int off, int len) throws IOException {
                     len -= toRead;
                 } else { // toRead == len, i.e. we filled the buffer.
                     leftOverCheckReset();
-                    return initialLen - (len - toRead);
+                    return initialLen;
                 }
             }
             // Avoid fetching a new element if we have no more space to read to. This prevents serializing and
@@ -120,8 +120,7 @@ public final int read(final byte[] b, int off, int len) throws IOException {
                 return initialLen;
             }
             if (!iterator.hasNext()) {
-                final int bytesRead = initialLen - len;
-                return bytesRead == 0 ? -1 : bytesRead;
+                return initialLen == len ? -1 : (initialLen - len);
             }
             nextLeftOver(iterator);
         }