SSL (for jooby-http2-netty) broken for reactor Flux

Fix #2372

Author: jknack

modules/jooby-netty/src/main/java/io/jooby/internal/netty/NettyContext.java

@@ -73,6 +73,7 @@
 import io.netty.buffer.Unpooled;
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelFutureListener;
+import io.netty.channel.ChannelHandler;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelPipeline;
 import io.netty.channel.ChannelPromise;
@@ -801,7 +802,7 @@ private void prepareChunked() {
     // remove flusher, doesn't play well with streaming/chunked responses
     ChannelPipeline pipeline = ctx.pipeline();
     if (pipeline.get("chunker") == null) {
-      String base = Stream.of("compressor", "codec")
+      String base = Stream.of("compressor", "codec", "http2")
           .filter(name -> pipeline.get(name) != null)
           .findFirst()
           .orElseThrow(() -> new IllegalStateException("No available handler for chunk writer"));

tests/src/test/java/io/jooby/FeaturedTest.java

@@ -12,6 +12,7 @@
 import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static reactor.core.scheduler.Schedulers.elastic;
+import static reactor.core.scheduler.Schedulers.parallel;
 
 import java.io.ByteArrayInputStream;
 import java.io.IOException;
@@ -1060,12 +1061,12 @@ public void reactor(ServerTestRunner runner) {
       app.get("/reactor/mono", ctx ->
           Mono.fromCallable(() -> "Mono")
               .map(s -> "Hello " + s)
-              .subscribeOn(elastic())
+              .subscribeOn(parallel())
       );
       app.get("/reactor/flux", ctx ->
           Flux.range(1, 10)
               .map(i -> i + ",")
-              .subscribeOn(elastic())
+              .subscribeOn(parallel())
       );
     }).ready(client -> {
       client.get("/reactor/mono", rsp -> {

tests/src/test/java/io/jooby/Issue2372.java

@@ -0,0 +1,46 @@
+package io.jooby;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.util.Arrays;
+
+import io.jooby.junit.ServerTest;
+import io.jooby.junit.ServerTestRunner;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+public class Issue2372 {
+
+  @ServerTest
+  public void http2(ServerTestRunner runner) {
+    runner.define(app -> {
+
+      app.setServerOptions(
+          new ServerOptions()
+              .setHttp2(true)
+              .setSecurePort(8443)
+      );
+
+      app.before(new SSLHandler());
+
+      app.get("/2372/mono", ctx -> {
+        return Mono.fromCallable(() -> "Welcome to Jooby!");
+      });
+
+      app.get("/2372/flux", ctx -> {
+        return Flux.fromIterable(Arrays.asList("Welcome", "to", "Jooby!"))
+            .map(it -> it + " ");
+      });
+    }).ready((http, https) -> {
+      https.get("/2372/flux", rsp -> {
+        assertEquals("Welcome to Jooby!",
+            rsp.body().string().trim());
+      });
+      https.get("/2372/mono", rsp -> {
+        assertEquals("Welcome to Jooby!",
+            rsp.body().string());
+      });
+    });
+  }
+
+}