output:clean up + concurrent test

Author: jknack

jooby/src/main/java/io/jooby/internal/output/OutputByteArrayStatic.java

@@ -10,7 +10,6 @@
 
 import edu.umd.cs.findbugs.annotations.NonNull;
 import io.jooby.internal.output.CompositeOutput;
-import io.jooby.internal.output.OutputByteArrayStatic;
 import io.jooby.internal.output.OutputStatic;
 import io.jooby.internal.output.WrappedOutput;
 
@@ -34,7 +33,7 @@ public Output wrap(@NonNull ByteBuffer buffer) {
 
     @Override
     public Output wrap(@NonNull String value, @NonNull Charset charset) {
-      return new OutputByteArrayStatic(value.getBytes(charset));
+      return new WrappedOutput(ByteBuffer.wrap(value.getBytes(charset)));
     }
 
     @Override
@@ -81,12 +80,12 @@ public Output wrap(@NonNull ByteBuffer buffer) {
 
   @Override
   public Output wrap(@NonNull byte[] bytes) {
-    return new OutputByteArrayStatic(bytes);
+    return wrap(bytes, 0, bytes.length);
   }
 
   @Override
   public Output wrap(@NonNull byte[] bytes, int offset, int length) {
-    return new OutputByteArrayStatic(bytes, offset, length);
+    return new OutputStatic(ByteBuffer.wrap(bytes, offset, length));
   }
 
   @Override

jooby/src/main/java/io/jooby/output/ByteBufferedOutputFactory.java

@@ -12,13 +12,7 @@
 import io.jooby.output.BufferedOutput;
 import io.netty.buffer.ByteBuf;
 
-public class NettyByteBufOutput implements BufferedOutput, NettyByteBufRef {
-
-  private final ByteBuf buffer;
-
-  protected NettyByteBufOutput(ByteBuf buffer) {
-    this.buffer = buffer;
-  }
+public record NettyByteBufOutput(ByteBuf buffer) implements BufferedOutput, NettyByteBufRef {
 
   @Override
   @NonNull public BufferedOutput write(byte b) {

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

@@ -37,8 +37,8 @@ default void send(Context ctx) {
   }
 
   static ByteBuf byteBuf(Output output) {
-    if (output instanceof NettyByteBufRef netty) {
-      return netty.byteBuf();
+    if (output instanceof NettyByteBufRef byteBuf) {
+      return byteBuf.byteBuf();
     } else {
       return Unpooled.wrappedBuffer(output.asByteBuffer());
     }

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

@@ -90,7 +90,7 @@ public ByteBufAllocator getAllocator() {
 
   @Override
   @NonNull public Output wrap(@NonNull byte[] bytes, int offset, int length) {
-    return new NettyOutputByteArrayStatic(bytes, offset, length);
+    return new NettyOutputUnsafeHeapByteBuf(bytes, offset, length);
   }
 
   @Override

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

@@ -9,12 +9,12 @@
 import io.jooby.Context;
 import io.netty.buffer.ByteBuf;
 
-public class NettyOutputByteArrayStaticNew implements NettyByteBufRef {
+public class NettyOutputUnsafeHeapByteBuf implements NettyByteBufRef {
   private final ByteBuf buf;
   private final NettyString contentLength;
   private final int length;
 
-  protected NettyOutputByteArrayStaticNew(byte[] bytes, int offset, int len) {
+  protected NettyOutputUnsafeHeapByteBuf(byte[] bytes, int offset, int len) {
     this.length = len - offset;
     this.buf = new NettyUnsafeHeapByteBuf(this.length, this.length);
     this.buf.writeBytes(bytes, offset, length);
@@ -33,7 +33,7 @@ public int size() {
   @Override
   public void send(Context ctx) {
     if (ctx.getClass() == NettyContext.class) {
-      ((NettyContext) ctx).send(buf.slice(), contentLength);
+      ((NettyContext) ctx).send(buf.slice(0, length), contentLength);
     } else {
       ctx.send(asByteBuffer());
     }

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

@@ -0,0 +1,43 @@
+/*
+ * Jooby https://jooby.io
+ * Apache License Version 2.0 https://jooby.io/LICENSE.txt
+ * Copyright 2014 Edgar Espina
+ */
+package examples;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.util.concurrent.*;
+
+import io.jooby.junit.ServerTest;
+import io.jooby.junit.ServerTestRunner;
+
+public class ConcurrentTest {
+
+  @ServerTest
+  public void makeSureBufferWork(ServerTestRunner runner)
+      throws ExecutionException, InterruptedException {
+    var payload = "Hello World!";
+    runner
+        .define(
+            app -> {
+              var outputFactory = app.getOutputFactory();
+              var message = outputFactory.wrap(payload);
+
+              app.get(
+                  "/plaintext",
+                  ctx -> {
+                    return ctx.send(message);
+                  });
+            })
+        .ready(
+            http -> {
+              http.concurrent(
+                  "/plaintext",
+                  20,
+                  rsp -> {
+                    assertEquals(payload, rsp.body().string());
+                  });
+            });
+  }
+}

…netty/NettyOutputByteArrayStaticNew.java …/netty/NettyOutputUnsafeHeapByteBuf.javamodules/jooby-netty/src/main/java/io/jooby/internal/netty/NettyOutputByteArrayStaticNew.java renamed to modules/jooby-netty/src/main/java/io/jooby/internal/netty/NettyOutputUnsafeHeapByteBuf.java modules/jooby-netty/src/main/java/io/jooby/internal/netty/NettyOutputByteArrayStaticNew.java renamed to modules/jooby-netty/src/main/java/io/jooby/internal/netty/NettyOutputUnsafeHeapByteBuf.java

@@ -14,7 +14,7 @@
 import io.jooby.Jooby;
 import io.jooby.ServerOptions;
 import io.jooby.StartupSummary;
-import io.jooby.netty.NettyServer;
+import io.jooby.jetty.JettyServer;
 
 public class Performance extends Jooby {
 
@@ -43,6 +43,6 @@ public class Performance extends Jooby {
 
   public static void main(final String[] args) {
     System.setProperty("io.netty.disableHttpHeadersValidation", "true");
-    runApp(args, new NettyServer(new ServerOptions()), EVENT_LOOP, Performance::new);
+    runApp(args, new JettyServer(new ServerOptions()), EVENT_LOOP, Performance::new);
   }
 }

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

@@ -14,10 +14,7 @@
 import java.security.cert.X509Certificate;
 import java.util.ArrayList;
 import java.util.List;
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.*;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.BiConsumer;
 import java.util.function.Consumer;
@@ -34,12 +31,7 @@
 import io.jooby.ServerSentMessage;
 import io.jooby.SneakyThrows;
 import io.jooby.WebSocketCloseStatus;
-import okhttp3.Headers;
-import okhttp3.OkHttpClient;
-import okhttp3.RequestBody;
-import okhttp3.Response;
-import okhttp3.WebSocket;
-import okhttp3.WebSocketListener;
+import okhttp3.*;
 import okhttp3.sse.EventSource;
 import okhttp3.sse.EventSourceListener;
 import okhttp3.sse.EventSources;
@@ -176,12 +168,16 @@ 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);
+              .followRedirects(followRedirects)
+              .dispatcher(dispatcher);
       if (scheme.equalsIgnoreCase("https")) {
         configureSelfSigned(builder);
       }
@@ -229,6 +225,32 @@ 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);