Flush HTTP/2 request body per event-stream message

Outbound event streams previously lost their message boundaries twice on the
write path: DefaultEventStreamWriter.toDataStream() wrapped the per-event
EventPipeStream queue as a plain InputStream, and H2StreamRequestBody drained
it with asInputStream().transferTo(out) — byte-oriented, so small events were
coalesced into the 16 KB frame buffer and not sent until it filled or the
stream closed. A request/response event protocol (send a small message, await
the peer's reply, then send the next) would deadlock: the message never left
the buffer, so the reply never came.

Fix, via the existing DataStream.writeTo seam (no new public API, no reactive
types in the blocking client):
- EventPipeStream.writeMessagesTo(out) drains one queued event ByteBuffer at a
  time and flushes after each, preserving boundaries.
- DefaultEventStreamWriter.toDataStream() returns a DataStream whose writeTo
  routes through writeMessagesTo (and still works as a normal InputStream for
  any other consumer).
- H2StreamRequestBody drives the streaming write through body.writeTo(out)
  instead of asInputStream().transferTo(out). The default writeTo is still
  transferTo, so bulk uploads (S3 etc.) are unchanged.

Tests: EventPipeStreamTest.writeMessagesToFlushesOncePerEvent (unit, one flush
per event) and PerMessageFlushHttp2Test (integ, verified discriminator — a
ping-pong body that awaits each message's echo before sending the next; passes
with per-message flush, deadlocks to the @timeout on the coalescing path).

Author: mtdowling

core/src/main/java/software/amazon/smithy/java/core/serde/event/DefaultEventStreamWriter.java

@@ -5,6 +5,10 @@
 
 package software.amazon.smithy.java.core.serde.event;
 
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.io.UncheckedIOException;
 import java.util.Objects;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
@@ -31,16 +35,19 @@ final class DefaultEventStreamWriter<IE extends SerializableStruct, T extends Se
         F extends Frame<?>>
         implements ProtocolEventStreamWriter<T, IE, F> {
     private static final InternalLogger LOGGER = InternalLogger.getLogger(DefaultEventStreamWriter.class);
+
     /**
      * Default timeout to block waiting to write.
      */
     private static final int WRITE_TIMEOUT_MILLIS = 10_000;
+
     /**
      * This latch is used to ensure that the protocol handler writes the initial event
      * before any other event is written. Protocols that don't require the initial event still have
      * to unlatch the writer by bootstrapping it with a null value.
      */
     private final CountDownLatch readyLatch = new CountDownLatch(1);
+
     /**
      * Pipes bytes written by this writer to an input stream used
      * to send them over the wire.
@@ -184,8 +191,7 @@ private void checkState() {
 
     @Override
     public EventStreamReader<T> asReader() {
-        throw new UnsupportedOperationException(
-                "This writer cannot be converted to a reader");
+        throw new UnsupportedOperationException("This writer cannot be converted to a reader");
     }
 
     @Override
@@ -233,6 +239,64 @@ public void close() {
      */
     @Override
     public DataStream toDataStream() {
-        return DataStream.ofInputStream(pipeStream);
+        return new EventStreamDataStream(pipeStream);
+    }
+
+    /**
+     * A {@link DataStream} over the event pipe whose {@link #writeTo(OutputStream)} flushes after each event
+     * (see {@link EventPipeStream#writeMessagesTo}), so a transport draining via {@code writeTo} emits one
+     * frame per event. Still works as an ordinary unknown-length, non-replayable {@code InputStream}-backed
+     * stream via {@link #asInputStream()}.
+     */
+    private static final class EventStreamDataStream implements DataStream {
+        private final EventPipeStream pipeStream;
+        // The pipe is single-use: once drained (via writeTo/asInputStream) or closed it cannot be replayed.
+        private final AtomicBoolean consumed = new AtomicBoolean(false);
+
+        EventStreamDataStream(EventPipeStream pipeStream) {
+            this.pipeStream = pipeStream;
+        }
+
+        @Override
+        public void writeTo(OutputStream out) throws IOException {
+            consumed.set(true);
+            pipeStream.writeMessagesTo(out);
+        }
+
+        @Override
+        public InputStream asInputStream() {
+            consumed.set(true);
+            return pipeStream;
+        }
+
+        @Override
+        public long contentLength() {
+            return -1;
+        }
+
+        @Override
+        public String contentType() {
+            return null;
+        }
+
+        @Override
+        public boolean isReplayable() {
+            return false;
+        }
+
+        @Override
+        public boolean isAvailable() {
+            return !consumed.get();
+        }
+
+        @Override
+        public void close() {
+            consumed.set(true);
+            try {
+                pipeStream.close();
+            } catch (IOException e) {
+                throw new UncheckedIOException("Failed to close event stream", e);
+            }
+        }
     }
 }

core/src/main/java/software/amazon/smithy/java/core/serde/event/EventPipeStream.java

@@ -7,6 +7,7 @@
 
 import java.io.IOException;
 import java.io.InputStream;
+import java.io.OutputStream;
 import java.nio.ByteBuffer;
 import java.util.Objects;
 import java.util.concurrent.ArrayBlockingQueue;
@@ -30,6 +31,7 @@
  */
 final class EventPipeStream extends InputStream {
     private static final InternalLogger LOGGER = InternalLogger.getLogger(EventPipeStream.class);
+
     /**
      * Poison pill used to signal the end of the stream.
      */
@@ -160,6 +162,56 @@ public int read() throws IOException {
         return b & 0xFF;
     }
 
+    /**
+     * Drains queued event messages to {@code out}, writing each whole and flushing after it. This keeps
+     * per-event boundaries (unlike a byte-oriented {@code transferTo}), so a flushing transport sends each
+     * event as its own frame instead of buffering. Blocks the consumer thread until {@link #complete()}
+     * (returns) or {@link #completeWithError} (throws).
+     *
+     * @param out the sink to write messages to
+     * @throws IOException if writing fails, the producer signalled an error, or the thread is interrupted
+     */
+    void writeMessagesTo(OutputStream out) throws IOException {
+        if (closed) {
+            throw new IOException("Stream is closed");
+        }
+
+        // Flush any buffer left partially consumed by a prior read() first.
+        if (current != null && current != POISON_PILL) {
+            writeBuffer(out, current);
+            current = null;
+            out.flush();
+        }
+
+        while (true) {
+            ByteBuffer message;
+            try {
+                message = queue.take();
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                throw new IOException("Interrupted while reading", e);
+            }
+
+            if (message == POISON_PILL) {
+                checkError();
+                return;
+            }
+
+            writeBuffer(out, message);
+            out.flush();
+        }
+    }
+
+    private static void writeBuffer(OutputStream out, ByteBuffer message) throws IOException {
+        if (message.hasArray()) {
+            out.write(message.array(), message.arrayOffset() + message.position(), message.remaining());
+        } else {
+            byte[] tmp = new byte[message.remaining()];
+            message.get(tmp);
+            out.write(tmp);
+        }
+    }
+
     @Override
     public int available() throws IOException {
         checkError();

core/src/test/java/software/amazon/smithy/java/core/serde/event/EventPipeStreamTest.java

@@ -9,8 +9,12 @@
 
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
+import java.io.OutputStream;
 import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
+import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.MethodSource;
 
@@ -83,4 +87,42 @@ static String[] sources() {
     static byte[] charToUtf8Bytes(char c) {
         return Character.toString(c).getBytes(StandardCharsets.UTF_8);
     }
+
+    @Test
+    void writeMessagesToFlushesOncePerEvent() throws IOException {
+        var pipe = new EventPipeStream();
+        var events = new String[] {"syn", "event-two", "fin"};
+        Thread.ofVirtual().start(() -> {
+            for (var e : events) {
+                pipe.write(ByteBuffer.wrap(e.getBytes(StandardCharsets.UTF_8)));
+            }
+            pipe.complete();
+        });
+
+        // Records the bytes accumulated at each flush() so we can assert one flush == one whole event.
+        var flushes = new ArrayList<String>();
+        var current = new ByteArrayOutputStream();
+        var recordingSink = new OutputStream() {
+            @Override
+            public void write(int b) {
+                current.write(b);
+            }
+
+            @Override
+            public void write(byte[] b, int off, int len) {
+                current.write(b, off, len);
+            }
+
+            @Override
+            public void flush() {
+                flushes.add(current.toString(StandardCharsets.UTF_8));
+                current.reset();
+            }
+        };
+
+        pipe.writeMessagesTo(recordingSink);
+
+        // One flush per event, in order, proving boundaries are preserved (transferTo would coalesce them).
+        assertEquals(List.of("syn", "event-two", "fin"), flushes);
+    }
 }

http/http-client/src/it/java/software/amazon/smithy/java/http/client/it/h2/BidirectionalStreamingHttp2Test.java

@@ -26,24 +26,15 @@
  * Proves true HTTP/2 bidirectional (full-duplex) streaming: the response is read while the request body is
  * still open and unfinished.
  *
- * <p>The request body ({@link BlockingInputStream}) hands over a leading payload and then <em>blocks
- * indefinitely</em> before signalling end-of-stream — it only unblocks once the test thread releases it,
- * which the test does only <em>after</em> {@code client.send(...)} has returned and the leading echo has
- * been read. The server ({@link EchoHttp2ClientHandler}) sends response HEADERS as soon as the request
- * HEADERS arrive and echoes each request DATA frame straight back.
+ * <p>The request body ({@link BlockingInputStream}) hands over a leading payload then blocks indefinitely,
+ * unblocking only after the test releases it, which the test does only after {@code client.send(...)} has
+ * returned and the leading echo has been read. The server ({@link EchoHttp2ClientHandler}) sends response
+ * HEADERS on the request HEADERS and echoes each request DATA frame back.
  *
- * <p>This is a strict discriminator for duplex:
- * <ul>
- *   <li><b>Full duplex (correct):</b> the client writes the request body on a background virtual thread
- *       (see {@code DefaultHttpClient.sendForRoute}), so {@code send()} returns on the response HEADERS
- *       while the body is still blocked. The test reads the leading echo, then releases the body to finish.</li>
- *   <li><b>Serialized (broken):</b> if the client wrote the request body inline before reading the
- *       response, {@code send()} would block inside the body write forever — the test thread that releases
- *       the body never runs — and the test fails on the timeout.</li>
- * </ul>
- *
- * <p>Verified to be a real discriminator: forcing the client onto the inline-write path makes this test
- * hang to the timeout, while the duplex path passes.
+ * <p>Verified discriminator: the client writes the request body on a background virtual thread (see
+ * {@code DefaultHttpClient.sendForRoute}), so {@code send()} returns on the response HEADERS while the body
+ * is still blocked. Forcing the inline-write path instead makes {@code send()} block in the body write
+ * forever and the test times out.
  */
 public class BidirectionalStreamingHttp2Test extends BaseHttpClientIntegTest {
 
@@ -89,11 +80,9 @@ void readsResponseWhileRequestBodyIsStillOpen() throws Exception {
             assertEquals(200, response.statusCode());
 
             try (InputStream in = response.body().asInputStream()) {
-                // Read one full 16 KB frame of the echo while the request body is still blocked. Receiving
-                // any response body before releasing the request is the proof that the two directions
-                // interleave. (Reading the whole leading payload could block on bytes the client hasn't
-                // flushed yet, since an OutputStream-backed body only flushes on a full frame buffer or
-                // close — one frame is enough.)
+                // Read one full 16 KB frame of the echo while the request body is still blocked. Getting any
+                // response body before releasing the request proves the two directions interleave. (One
+                // frame, not the whole payload, since the body only flushes on a full frame buffer.)
                 int prefixLen = 16 * 1024;
                 byte[] leadingEcho = readN(in, prefixLen);
                 byte[] expectedPrefix = Arrays.copyOf(LEADING_PAYLOAD, prefixLen);