fix: Sync-flush gzip so flushed stream data reaches the client

README.md

@@ -523,7 +523,8 @@ shared caches keep the two forms apart. It is merged into any `Vary` your handle
 A handler that sets its own `Content-Encoding` is left alone, and so is a payload the coding
 fails to shrink. Statuses that carry no content never get a coding.
 
-Streamed responses (`Response.stream(...)`) are coded as they are written. A length declared by
+Streamed responses (`Response.stream(...)`) are coded as they are written, and a handler's
+`flush()` sends the client everything written so far, coded. A length declared by
 the sized overload describes the uncoded body, so a coded stream goes out chunked; a stream of
 unknown length is coded regardless of the threshold, since measuring it would defeat streaming it.
 For the same reason a `HEAD` whose `GET` would be compressed omits `Content-Length` rather than

src/main/java/com/retailsvc/http/ContentCoding.java

@@ -49,6 +49,11 @@ default Set<String> aliases() {
   /**
    * Wraps a response stream so that everything written to the returned stream reaches {@code sink}
    * coded. Closing the returned stream must finish the coded payload and close {@code sink}.
+   *
+   * <p>Flushing the returned stream must write everything written so far to {@code sink} in a form
+   * the client can already decode, then flush {@code sink}. A coder that holds it back stalls
+   * streaming responses. For deflate-based codings this means a sync flush, which compressors
+   * usually leave off by default.
    */
   OutputStream encode(OutputStream sink) throws IOException;
 }

src/main/java/com/retailsvc/http/internal/GzipCoding.java

@@ -29,8 +29,12 @@ public InputStream decode(InputStream coded) throws IOException {
     return new GZIPInputStream(coded, BUFFER_SIZE);
   }
 
+  /**
+   * Sync-flushes, so a flushed streamed response reaches the client then rather than when the
+   * deflater's buffer fills. Whole bodies are coded without a flush, so they pay nothing for it.
+   */
   @Override
   public OutputStream encode(OutputStream sink) throws IOException {
-    return new GZIPOutputStream(sink);
+    return new GZIPOutputStream(sink, true);
   }
 }

src/test/java/com/retailsvc/http/GzipIT.java

@@ -6,6 +6,7 @@
 import static java.net.HttpURLConnection.HTTP_UNSUPPORTED_TYPE;
 import static java.nio.charset.StandardCharsets.UTF_8;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively;
 
 import com.retailsvc.http.start.TextEchoHandler;
 import java.io.ByteArrayInputStream;
@@ -17,7 +18,10 @@
 import java.net.http.HttpRequest.BodyPublishers;
 import java.net.http.HttpResponse;
 import java.net.http.HttpResponse.BodyHandlers;
+import java.time.Duration;
 import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
 import java.util.function.UnaryOperator;
 import java.util.zip.GZIPInputStream;
 import java.util.zip.GZIPOutputStream;
@@ -240,6 +244,52 @@ void headKeepsContentLengthWithoutAcceptEncoding() throws Exception {
     }
   }
 
+  @Test
+  void flushedEventOfGzippedStreamReachesClientBeforeStreamEnds() throws Exception {
+    byte[] first = "{\"event\":1}\n".getBytes(UTF_8);
+    byte[] second = "{\"event\":2}\n".getBytes(UTF_8);
+    CountDownLatch clientSawFirst = new CountDownLatch(1);
+    RequestHandler events =
+        req ->
+            Response.stream(
+                HTTP_OK,
+                "application/x-ndjson",
+                out -> {
+                  out.write(first);
+                  out.flush();
+                  try {
+                    clientSawFirst.await(10, TimeUnit.SECONDS);
+                  } catch (InterruptedException e) {
+                    Thread.currentThread().interrupt();
+                    throw new IOException(e);
+                  }
+                  out.write(second);
+                });
+    try (var s = serverWith(Map.of(), builder -> builder.extraRoute("/events", events));
+        var client = httpClient()) {
+      var request =
+          HttpRequest.newBuilder()
+              .uri(URI.create("http://localhost:%d/events".formatted(s.listenPort())))
+              .header(ACCEPT_ENCODING, "gzip")
+              .GET()
+              .build();
+
+      HttpResponse<InputStream> response = client.send(request, BodyHandlers.ofInputStream());
+
+      assertThat(response.headers().firstValue(CONTENT_ENCODING)).contains("gzip");
+      try (var in = new GZIPInputStream(response.body())) {
+        try {
+          byte[] got =
+              assertTimeoutPreemptively(Duration.ofSeconds(5), () -> in.readNBytes(first.length));
+          assertThat(got).isEqualTo(first);
+        } finally {
+          clientSawFirst.countDown();
+        }
+        assertThat(in.readAllBytes()).isEqualTo(second);
+      }
+    }
+  }
+
   // -- fixtures --
 
   private OpenApiServer echoServer() {

src/test/java/com/retailsvc/http/internal/GzipCodingTest.java

@@ -0,0 +1,44 @@
+package com.retailsvc.http.internal;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.util.zip.GZIPInputStream;
+import org.junit.jupiter.api.Test;
+
+class GzipCodingTest {
+
+  private final GzipCoding coding = new GzipCoding();
+
+  @Test
+  void flushEmitsEverythingWrittenSoFarInDecodableForm() throws IOException {
+    byte[] event = "{\"event\":1}\n".getBytes(UTF_8);
+    var sink = new ByteArrayOutputStream();
+
+    var out = coding.encode(sink);
+    out.write(event);
+    out.flush();
+
+    try (var in = new GZIPInputStream(new ByteArrayInputStream(sink.toByteArray()))) {
+      assertThat(in.readNBytes(event.length)).isEqualTo(event);
+    }
+  }
+
+  @Test
+  void closedStreamRoundTrips() throws IOException {
+    byte[] body = "{\"event\":1}\n{\"event\":2}\n".getBytes(UTF_8);
+    var sink = new ByteArrayOutputStream();
+
+    try (var out = coding.encode(sink)) {
+      out.write(body);
+      out.flush();
+    }
+
+    try (var in = coding.decode(new ByteArrayInputStream(sink.toByteArray()))) {
+      assertThat(in.readAllBytes()).isEqualTo(body);
+    }
+  }
+}