netty: exceed file size are not detected #3790

- fix #3790

Author: jknack

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

@@ -31,7 +31,6 @@ public class NettyHandler extends ChannelInboundHandlerAdapter {
   private final boolean defaultHeaders;
   private final long maxRequestSize;
   private final int maxFormFields;
-  private long contentLength;
   private long chunkSize;
   private final boolean http2;
   private NettyContext context;
@@ -83,7 +82,7 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) {
         router.match(context).execute(context);
       } else {
         // possibly body:
-        contentLength = contentLength(req);
+        long contentLength = contentLength(req);
         if (contentLength > 0 || isTransferEncodingChunked(req)) {
           context.httpDataFactory = new DefaultHttpDataFactory(bufferSize);
           context.httpDataFactory.setBaseDir(app.getTmpdir().toString());
@@ -93,20 +92,6 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) {
           router.match(context).execute(context);
         }
       }
-    } else if (isLastHttpContent(msg)) {
-      var chunk = (HttpContent) msg;
-      try {
-        // when decoder == null, chunk is always a LastHttpContent.EMPTY, ignore it
-        if (context.decoder != null) {
-          if (offer(context, chunk)) {
-            Router.Match route = router.match(context);
-            resetDecoderState(context, !route.matches());
-            route.execute(context);
-          }
-        }
-      } finally {
-        release(chunk);
-      }
     } else if (isHttpContent(msg)) {
       this.read = true;
       var chunk = (HttpContent) msg;
@@ -119,10 +104,13 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) {
             router.match(context).execute(context, Route.REQUEST_ENTITY_TOO_LARGE);
             return;
           }
-          offer(context, chunk);
+          if (offer(context, chunk) && isLastHttpContent(msg)) {
+            Router.Match route = router.match(context);
+            resetDecoderState(context, !route.matches());
+            route.execute(context);
+          }
         }
       } finally {
-        // must be released
         release(chunk);
       }
     } else if (isWebSocketFrame(msg)) {
@@ -251,7 +239,6 @@ private boolean offer(NettyContext context, HttpContent chunk) {
 
   private void resetDecoderState(NettyContext context, boolean destroy) {
     chunkSize = 0;
-    contentLength = -1;
     if (destroy && context.decoder != null) {
       var decoder = context.decoder;
       var httpDataFactory = context.httpDataFactory;

tests/src/test/java/examples/ConcurrentTest.java

@@ -15,8 +15,7 @@
 public class ConcurrentTest {
 
   @ServerTest
-  public void makeSureBufferWork(ServerTestRunner runner)
-      throws ExecutionException, InterruptedException {
+  public void makeSureBufferWork(ServerTestRunner runner) {
     var payload = "Hello World!";
     runner
         .define(
@@ -32,12 +31,7 @@ public void makeSureBufferWork(ServerTestRunner runner)
             })
         .ready(
             http -> {
-              http.concurrent(
-                  "/plaintext",
-                  20,
-                  rsp -> {
-                    assertEquals(payload, rsp.body().string());
-                  });
+              http.get("/plaintext").execute(50, rsp -> assertEquals(payload, rsp.body().string()));
             });
   }
 }

tests/src/test/java/io/jooby/i2806/Issue2806.java

@@ -5,9 +5,9 @@
  */
 package io.jooby.i2806;
 
+import static io.jooby.test.TestUtil._19kb;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 
-import java.util.Arrays;
 import java.util.Map;
 
 import com.google.common.collect.ImmutableMap;
@@ -23,9 +23,6 @@ public class Issue2806 {
 
   @ServerTest
   public void renderShouldWorkFromErrorHandlerWhenLargeRequestAreSent(ServerTestRunner runner) {
-    char[] chars = new char[19 * 1024];
-    Arrays.fill(chars, 'S');
-    String _19kb = new String(chars);
     runner
         .options(
             new ServerOptions()
@@ -43,20 +40,32 @@ public void renderShouldWorkFromErrorHandlerWhenLargeRequestAreSent(ServerTestRu
                     ctx.render(map);
                   });
 
-              app.post("/2806", ctx -> ctx.body().value(""));
+              app.post(
+                  "/2806",
+                  ctx -> {
+                    return ctx.body().value("");
+                  });
 
               app.get("/2806", ctx -> ctx.body().value(""));
             })
         .ready(
             client -> {
               // Exceeds
-              client.post(
-                  "/2806",
-                  RequestBody.create(_19kb, MediaType.get("text/plain")),
-                  rsp -> {
-                    // assertEquals(413, rsp.code());
-                    assertEquals("{\"router\":true,\"route\":true}", rsp.body().string());
-                  });
+              client
+                  .post("/2806", RequestBody.create(_19kb, MediaType.get("text/plain")))
+                  .execute(
+                      rsp -> {
+                        assertEquals(413, rsp.code());
+                        assertEquals("{\"router\":true,\"route\":true}", rsp.body().string());
+                      });
+
+              client
+                  .get("/2806")
+                  .execute(
+                      rsp -> {
+                        assertEquals(200, rsp.code());
+                        assertEquals("", rsp.body().string());
+                      });
             });
   }
 }

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

@@ -134,17 +134,46 @@ public void close() {
 
   public class Request {
     private final okhttp3.Request.Builder req;
+    private SneakyThrows.Consumer<okhttp3.Request.Builder> configurer;
 
     public Request(okhttp3.Request.Builder req) {
       this.req = req;
     }
 
     public Request prepare(SneakyThrows.Consumer<okhttp3.Request.Builder> configurer) {
-      configurer.accept(req);
+      this.configurer = configurer;
       return this;
     }
 
     public void execute(SneakyThrows.Consumer<Response> callback) {
+      execute(1, callback);
+    }
+
+    public void execute(int concurrency, SneakyThrows.Consumer<Response> callback) {
+      if (configurer != null) {
+        configurer.accept(req);
+      }
+      if (concurrency > 1) {
+        var futures = new ArrayList<CompletableFuture<String>>();
+        for (var i = 0; i < concurrency; i++) {
+          futures.add(
+              CompletableFuture.supplyAsync(
+                  () -> {
+                    executeCall(callback);
+                    return "success";
+                  }));
+          try {
+            CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
+          } catch (CompletionException x) {
+            throw SneakyThrows.propagate(x.getCause());
+          }
+        }
+      } else {
+        executeCall(callback);
+      }
+    }
+
+    private void executeCall(SneakyThrows.Consumer<Response> callback) {
       okhttp3.Request r = req.build();
       try (Response rsp = client.newCall(r).execute()) {
         callback.accept(rsp);
@@ -168,16 +197,12 @@ public WebClient(String scheme, int port, boolean followRedirects) {
     try {
       this.scheme = scheme;
       this.port = port;
-      Dispatcher dispatcher = new Dispatcher();
-      dispatcher.setMaxRequests(100); // Maximum 20 concurrent requests overall
-      dispatcher.setMaxRequestsPerHost(100);
       OkHttpClient.Builder builder =
           new OkHttpClient.Builder()
               .connectTimeout(5, TimeUnit.MINUTES)
               .writeTimeout(5, TimeUnit.MINUTES)
               .readTimeout(5, TimeUnit.MINUTES)
-              .followRedirects(followRedirects)
-              .dispatcher(dispatcher);
+              .followRedirects(followRedirects);
       if (scheme.equalsIgnoreCase("https")) {
         configureSelfSigned(builder);
       }
@@ -225,32 +250,6 @@ public Request get(String path) {
     return invoke("GET", path, null);
   }
 
-  public void concurrent(String path, int concurrency, SneakyThrows.Consumer<Response> callback) {
-    var futures = new ArrayList<Future<String>>();
-    try (var executor = Executors.newFixedThreadPool(concurrency + 5)) {
-      for (var i = 0; i < concurrency; i++) {
-        futures.add(
-            executor.submit(
-                () -> {
-                  okhttp3.Request.Builder req = new okhttp3.Request.Builder();
-                  req.url(scheme + "://localhost:" + port + path);
-                  new Request(req).execute(callback);
-                  return "OK";
-                }));
-      }
-      for (Future<String> future : futures) {
-        try {
-          var result = future.get();
-          if (!"OK".equals(result)) {
-            throw new IllegalStateException(result);
-          }
-        } catch (Exception e) {
-          SneakyThrows.propagate(e);
-        }
-      }
-    }
-  }
-
   public ServerSentMessageIterator sse(String path) {
     okhttp3.Request.Builder req = new okhttp3.Request.Builder();
     setRequestHeaders(req);