netty/jetty: HTTP Outputstream.close method must close once and ignore any other call

- fix #3554

Author: jknack

modules/jooby-jetty/src/main/java/io/jooby/internal/jetty/JettyOutputStream.java

@@ -6,13 +6,15 @@
 package io.jooby.internal.jetty;
 
 import java.io.OutputStream;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import io.jooby.SneakyThrows;
 
 public class JettyOutputStream extends OutputStream {
 
-  private JettyContext jetty;
-  private OutputStream out;
+  private final JettyContext jetty;
+  private final OutputStream out;
+  private AtomicBoolean closed = new AtomicBoolean(false);
 
   public JettyOutputStream(OutputStream out, JettyContext jetty) {
     this.out = out;
@@ -41,7 +43,9 @@ public void flush() {
 
   @Override
   public void close() {
-    block(out::close, true);
+    if (closed.compareAndSet(false, true)) {
+      block(out::close, true);
+    }
   }
 
   private void block(SneakyThrows.Runnable task, boolean complete) {

modules/jooby-jetty/src/test/java/io/jooby/internal/jetty/Issue3554.java

@@ -0,0 +1,28 @@
+/*
+ * Jooby https://jooby.io
+ * Apache License Version 2.0 https://jooby.io/LICENSE.txt
+ * Copyright 2014 Edgar Espina
+ */
+package io.jooby.internal.jetty;
+
+import static org.mockito.Mockito.*;
+
+import java.io.IOException;
+import java.io.OutputStream;
+
+import org.junit.jupiter.api.Test;
+
+public class Issue3554 {
+
+  @Test
+  public void shouldCloseOutputStreamOnce() throws IOException {
+    var out = mock(OutputStream.class);
+    var ctx = mock(JettyContext.class);
+
+    var jettyOutputStream = new JettyOutputStream(out, ctx);
+    jettyOutputStream.close();
+    jettyOutputStream.close();
+
+    verify(out, times(1)).close();
+  }
+}

modules/jooby-netty/pom.xml

@@ -70,6 +70,18 @@
       <classifier>runtime</classifier>
       <scope>test</scope>
     </dependency>
+
+    <dependency>
+      <groupId>org.mockito</groupId>
+      <artifactId>mockito-core</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.slf4j</groupId>
+      <artifactId>slf4j-simple</artifactId>
+      <version>${slf4j.version}</version>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
 
   <profiles>

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

@@ -7,6 +7,7 @@
 
 import java.io.IOException;
 import java.io.OutputStream;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.ChannelFuture;
@@ -22,6 +23,7 @@ public class NettyOutputStream extends OutputStream {
   private final ChannelHandlerContext context;
   private final ChannelFutureListener closeListener;
   private HttpResponse headers;
+  private AtomicBoolean closed = new AtomicBoolean(false);
 
   public NettyOutputStream(
       NettyContext ctx, ChannelHandlerContext context, int bufferSize, HttpResponse headers) {
@@ -106,10 +108,12 @@ private void flush(ChannelFutureListener callback, ChannelFutureListener listene
 
   @Override
   public void close() {
-    try {
-      flush(null, closeListener);
-    } finally {
-      ctx.requestComplete();
+    if (closed.compareAndSet(false, true)) {
+      try {
+        flush(null, closeListener);
+      } finally {
+        ctx.requestComplete();
+      }
     }
   }
 }

modules/jooby-netty/src/test/java/io/jooby/internal/netty/Issue3554.java

@@ -0,0 +1,45 @@
+/*
+ * Jooby https://jooby.io
+ * Apache License Version 2.0 https://jooby.io/LICENSE.txt
+ * Copyright 2014 Edgar Espina
+ */
+package io.jooby.internal.netty;
+
+import static org.mockito.Mockito.*;
+
+import java.io.IOException;
+
+import org.junit.jupiter.api.Test;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufAllocator;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.http.HttpResponse;
+import io.netty.handler.codec.http.LastHttpContent;
+
+public class Issue3554 {
+
+  @Test
+  public void shouldCloseOutputStreamOnce() throws IOException {
+    var ctx = mock(NettyContext.class);
+    var headers = mock(HttpResponse.class);
+
+    var buffer = mock(ByteBuf.class);
+    when(buffer.readableBytes()).thenReturn(0);
+
+    var bufferAllocator = mock(ByteBufAllocator.class);
+    when(bufferAllocator.buffer(0, 1024)).thenReturn(buffer);
+
+    var future = mock(ChannelFuture.class);
+    var channelContext = mock(ChannelHandlerContext.class);
+    when(channelContext.alloc()).thenReturn(bufferAllocator);
+    when(channelContext.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT)).thenReturn(future);
+
+    var httpOutputStream = new NettyOutputStream(ctx, channelContext, 1024, headers);
+    httpOutputStream.close();
+    httpOutputStream.close();
+
+    verify(channelContext, times(1)).writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT);
+  }
+}

tests/src/test/java/io/jooby/i3554/Issue3554.java

@@ -0,0 +1,62 @@
+/*
+ * Jooby https://jooby.io
+ * Apache License Version 2.0 https://jooby.io/LICENSE.txt
+ * Copyright 2014 Edgar Espina
+ */
+package io.jooby.i3554;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.io.BufferedOutputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+import io.jooby.ExecutionMode;
+import io.jooby.ReactiveSupport;
+import io.jooby.ServerOptions;
+import io.jooby.junit.ServerTest;
+import io.jooby.junit.ServerTestRunner;
+
+public class Issue3554 {
+  @ServerTest(executionMode = ExecutionMode.EVENT_LOOP)
+  public void shouldNotThrowErrorOnCompletableWithSideEffect(ServerTestRunner runner) {
+    runner
+        .define(
+            app -> {
+              ExecutorService threadPool = Executors.newSingleThreadExecutor();
+
+              var serverOptions = new ServerOptions().setPort(9000).setDefaultHeaders(false);
+              app.setServerOptions(serverOptions);
+
+              app.use(ReactiveSupport.concurrent());
+
+              app.onStop(threadPool::shutdown);
+
+              app.get(
+                  "/3554",
+                  ctx ->
+                      CompletableFuture.supplyAsync(
+                          () -> {
+                            try (var os = ctx.responseStream();
+                                var bos = new BufferedOutputStream(os)) {
+                              bos.write("test".getBytes(StandardCharsets.UTF_8));
+                            } catch (Exception ex) {
+                              throw new RuntimeException(ex);
+                            }
+                            return ctx;
+                          },
+                          threadPool));
+            })
+        .ready(
+            http -> {
+              http.get(
+                  "/3554",
+                  rsp -> {
+                    assertEquals("test", rsp.body().string());
+                    assertEquals(200, rsp.code());
+                  });
+            });
+  }
+}