Merge pull request #4920 from michalszynkiewicz/read-timeout

read timeout for blocking http

extensions/undertow/deployment/src/test/java/io/quarkus/undertow/test/timeout/ReadTimeoutTestCase.java

@@ -0,0 +1,97 @@
+package io.quarkus.undertow.test.timeout;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.URI;
+import java.nio.ByteBuffer;
+import java.nio.channels.SocketChannel;
+import java.nio.charset.StandardCharsets;
+
+import org.hamcrest.MatcherAssert;
+import org.hamcrest.Matchers;
+import org.jboss.shrinkwrap.api.ShrinkWrap;
+import org.jboss.shrinkwrap.api.spec.JavaArchive;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import io.quarkus.test.QuarkusUnitTest;
+import io.restassured.RestAssured;
+
+public class ReadTimeoutTestCase {
+    @RegisterExtension
+    static QuarkusUnitTest runner = new QuarkusUnitTest()
+            .withConfigurationResource("application-timeout.properties")
+            .setArchiveProducer(() -> ShrinkWrap.create(JavaArchive.class)
+                    .addClasses(TimeoutTestServlet.class));
+
+    private String host;
+    private SocketChannel client;
+
+    @BeforeEach
+    public void init() throws IOException {
+        int port = RestAssured.port;
+        host = URI.create(RestAssured.baseURI).getHost();
+        InetSocketAddress hostAddress = new InetSocketAddress(host, port);
+        client = SocketChannel.open(hostAddress);
+        TimeoutTestServlet.reset();
+    }
+
+    @AfterEach
+    public void cleanUp() throws IOException {
+        client.close();
+    }
+
+    @Test
+    public void shouldNotProcessRequestWrittenTooSlowly() throws IOException, InterruptedException {
+        requestWithDelay(1000L);
+
+        ByteBuffer buffer = ByteBuffer.allocate(100000);
+        client.read(buffer);
+
+        assertFalse(TimeoutTestServlet.read);
+        assertNotNull(TimeoutTestServlet.error);
+    }
+
+    @Test
+    public void shouldProcessSlowlyProcessedRequest() throws IOException, InterruptedException {
+        requestWithDelay(100L, "Processing-Time: 1000");
+
+        ByteBuffer buffer = ByteBuffer.allocate(100000);
+        client.read(buffer);
+        MatcherAssert.assertThat(new String(buffer.array(), StandardCharsets.UTF_8),
+                Matchers.containsString(TimeoutTestServlet.TIMEOUT_SERVLET));
+        assertTrue(TimeoutTestServlet.read);
+    }
+
+    private void requestWithDelay(long sleepTime, String... headers)
+            throws IOException, InterruptedException {
+        String content = "message content";
+        writeToChannel("POST /timeout HTTP/1.1\r\n");
+        writeToChannel("Content-Length: " + ("The \r\n" + content).getBytes("UTF-8").length);
+        for (String header : headers) {
+            writeToChannel("\r\n" + header);
+        }
+        writeToChannel("\r\nHost: " + host);
+        writeToChannel("\r\nContent-Type: text/plain; charset=utf-8\r\n\r\n");
+        writeToChannel("The \r\n");
+        Thread.sleep(sleepTime);
+        writeToChannel(content);
+    }
+
+    private void writeToChannel(String s) {
+        try {
+            byte[] message = s.getBytes("UTF-8");
+            ByteBuffer buffer = ByteBuffer.wrap(message);
+            client.write(buffer);
+            buffer.clear();
+        } catch (IOException e) {
+            throw new RuntimeException("Failed to write to channel", e);
+        }
+    }
+}

extensions/undertow/deployment/src/test/java/io/quarkus/undertow/test/timeout/TimeoutTestServlet.java

@@ -0,0 +1,59 @@
+package io.quarkus.undertow.test.timeout;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.io.UncheckedIOException;
+import java.util.stream.Collectors;
+
+import javax.servlet.ServletException;
+import javax.servlet.annotation.WebServlet;
+import javax.servlet.http.HttpServlet;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+
+@WebServlet(urlPatterns = "/timeout")
+public class TimeoutTestServlet extends HttpServlet {
+
+    public static final String TIMEOUT_SERVLET = "timeout-servlet";
+    public static volatile boolean read = false;
+    public static volatile IOException error;
+
+    public static void reset() {
+        error = null;
+        read = false;
+    }
+
+    @Override
+    protected void doPost(HttpServletRequest req, HttpServletResponse resp) throws ServletException, IOException {
+        readRequestData(req);
+        try {
+            read = true;
+            mimicProcessing(req);
+            resp.getWriter().write(TIMEOUT_SERVLET);
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+        }
+    }
+
+    private void mimicProcessing(HttpServletRequest req) throws InterruptedException {
+        String header = req.getHeader("Processing-Time");
+        if (header != null) {
+            long sleepTime = Long.parseLong(header);
+            Thread.sleep(sleepTime);
+        }
+    }
+
+    private String readRequestData(HttpServletRequest req) {
+        try (InputStreamReader isReader = new InputStreamReader(req.getInputStream());
+                BufferedReader reader = new BufferedReader(isReader)) {
+            return reader.lines().collect(Collectors.joining("\n"));
+        } catch (IOException e) {
+            error = e;
+            throw new UncheckedIOException(e);
+        } catch (UncheckedIOException e) {
+            error = e.getCause();
+            throw e;
+        }
+    }
+}

extensions/undertow/deployment/src/test/resources/application-timeout.properties

@@ -0,0 +1 @@
+quarkus.http.read-timeout=0.5S

extensions/undertow/runtime/src/main/java/io/quarkus/undertow/runtime/UndertowDeploymentRecorder.java

@@ -5,6 +5,7 @@
 import java.net.SocketAddress;
 import java.nio.file.Path;
 import java.security.SecureRandom;
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.EventListener;
 import java.util.List;
@@ -359,6 +360,8 @@ public void handle(RoutingContext event) {
                 if (maxBodySize.isPresent()) {
                     exchange.setMaxEntitySize(maxBodySize.get().asLongValue());
                 }
+                Duration readTimeout = httpConfiguration.readTimeout;
+                exchange.setReadTimeout(readTimeout.toMillis());
                 defaultHandler.handle(exchange);
             }
         };

extensions/vertx-http/runtime/src/main/java/io/quarkus/vertx/http/runtime/HttpConfiguration.java

@@ -110,6 +110,12 @@ public class HttpConfiguration {
     @ConfigItem(defaultValue = "30M", name = "idle-timeout")
     public Duration idleTimeout;
 
+    /**
+     * Http connection read timeout
+     */
+    @ConfigItem(defaultValue = "60s", name = "read-timeout")
+    public Duration readTimeout;
+
     /**
      * Request body related settings
      */