Output stream redirect support (#7366)

Output stream redirect support

Signed-off-by: David Kral <david.k.kral@oracle.com>

nima/tests/integration/webserver/webserver/src/test/java/io/helidon/nima/tests/integration/server/FollowRedirectTest.java

@@ -0,0 +1,245 @@
+/*
+ * Copyright (c) 2023 Oracle and/or its affiliates.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.helidon.nima.tests.integration.server;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.io.UncheckedIOException;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.concurrent.TimeUnit;
+
+import io.helidon.common.http.Http;
+import io.helidon.nima.testing.junit5.webserver.ServerTest;
+import io.helidon.nima.testing.junit5.webserver.SetUpRoute;
+import io.helidon.nima.webclient.api.HttpClientResponse;
+import io.helidon.nima.webclient.http1.Http1Client;
+import io.helidon.nima.webclient.http1.Http1ClientResponse;
+import io.helidon.nima.webserver.http.HttpRouting;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+import static io.helidon.common.http.Http.Status.INTERNAL_SERVER_ERROR_500;
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+@ServerTest
+class FollowRedirectTest {
+    private static final StringBuilder BUFFER = new StringBuilder();
+    private final Http1Client webClient;
+
+    FollowRedirectTest(Http1Client client) {
+        this.webClient = client;
+    }
+
+    @SetUpRoute
+    static void router(HttpRouting.Builder router) {
+        router.route(Http.Method.PUT, "/infiniteRedirect", (req, res) -> {
+            res.status(Http.Status.TEMPORARY_REDIRECT_307)
+                    .header(Http.HeaderNames.LOCATION, "/infiniteRedirect2")
+                    .send();
+        }).route(Http.Method.PUT, "/infiniteRedirect2", (req, res) -> {
+            res.status(Http.Status.TEMPORARY_REDIRECT_307)
+                    .header(Http.HeaderNames.LOCATION, "/infiniteRedirect")
+                    .send();
+        }).route(Http.Method.PUT, "/redirect", (req, res) -> {
+            res.status(Http.Status.TEMPORARY_REDIRECT_307)
+                    .header(Http.HeaderNames.LOCATION, "/plain")
+                    .send();
+        }).route(Http.Method.PUT, "/redirectNoEntity", (req, res) -> {
+            res.status(Http.Status.FOUND_302)
+                    .header(Http.HeaderNames.LOCATION, "/plain")
+                    .send();
+        }).route(Http.Method.PUT, "/plain", (req, res) -> {
+            try (InputStream in = req.content().inputStream()) {
+                byte[] buffer = new byte[128];
+                int read;
+                while ((read = in.read(buffer)) > 0) {
+                    BUFFER.append("\n").append(new String(buffer, 0, read));
+                }
+                res.send("Test data:" + BUFFER);
+            } catch (Exception e) {
+                res.status(INTERNAL_SERVER_ERROR_500)
+                        .send(e.getMessage());
+            }
+        }).route(Http.Method.PUT, "/redirectAfterUpload", (req, res) -> {
+            try (InputStream in = req.content().inputStream()) {
+                byte[] buffer = new byte[128];
+                int read;
+                while ((read = in.read(buffer)) > 0) {
+                    BUFFER.append("\n").append(new String(buffer, 0, read));
+                }
+                res.status(Http.Status.SEE_OTHER_303)
+                        .header(Http.HeaderNames.LOCATION, "/afterUpload")
+                        .send();
+            } catch (Exception e) {
+                res.status(INTERNAL_SERVER_ERROR_500)
+                        .send(e.getMessage());
+            }
+        }).route(Http.Method.GET, "/afterUpload", (req, res) -> {
+            res.send("Upload completed!" + BUFFER);
+        }).route(Http.Method.GET, "/plain", (req, res) -> {
+            res.send("GET plain endpoint reached");
+        }).route(Http.Method.PUT, "/close", (req, res) -> {
+            byte[] buffer = new byte[10];
+            try (InputStream in = req.content().inputStream()) {
+                in.read(buffer);
+                throw new RuntimeException("BOOM!");
+            } catch (IOException e) {
+                res.status(INTERNAL_SERVER_ERROR_500)
+                        .send(e.getMessage());
+            }
+        }).route(Http.Method.PUT, "/wait", (req, res) -> {
+            TimeUnit.MILLISECONDS.sleep(500);
+            try (InputStream in = req.content().inputStream()) {
+                byte[] buffer = new byte[128];
+                while (in.read(buffer) > 0) {
+                    //Do nothing and just drain the entity
+                }
+                res.send("Request did not timeout");
+            } catch (Exception e) {
+                res.status(INTERNAL_SERVER_ERROR_500)
+                        .send(e.getMessage());
+            }
+        });
+    }
+
+    @AfterEach
+    void clearBuffer() {
+        BUFFER.setLength(0);
+    }
+
+    @Test
+    void testOutputStreamFollowRedirect() {
+        String expected = """
+                Test data:
+                0123456789
+                0123456789
+                0123456789""";
+        try (Http1ClientResponse response = webClient.put()
+                .path("/redirect")
+                .outputStream(it -> {
+                    it.write("0123456789".getBytes(StandardCharsets.UTF_8));
+                    it.write("0123456789".getBytes(StandardCharsets.UTF_8));
+                    it.write("0123456789".getBytes(StandardCharsets.UTF_8));
+                    it.close();
+                })) {
+            assertThat(response.entity().as(String.class), is(expected));
+        }
+    }
+
+    @Test
+    void testOutputStreamEntityNotKept() {
+        String expected = "GET plain endpoint reached";
+        try (Http1ClientResponse response = webClient.put()
+                .path("/redirectNoEntity")
+                .outputStream(it -> {
+                    it.write("0123456789".getBytes(StandardCharsets.UTF_8));
+                    it.write("0123456789".getBytes(StandardCharsets.UTF_8));
+                    it.write("0123456789".getBytes(StandardCharsets.UTF_8));
+                    it.close();
+                })) {
+            assertThat(response.entity().as(String.class), is(expected));
+        }
+    }
+
+    @Test
+    void testEmptyOutputStreamWithRedirectAfter() {
+        String expected = "Upload completed!";
+        try (Http1ClientResponse response = webClient.put()
+                .path("/redirectAfterUpload")
+                .outputStream(OutputStream::close)) {
+            assertThat(response.entity().as(String.class), is(expected));
+        }
+    }
+
+    @Test
+    void testEntityOutputStreamWithRedirectAfter() {
+        String expected = """
+                Upload completed!
+                0123456789
+                0123456789
+                0123456789""";
+        try (Http1ClientResponse response = webClient.put()
+                .path("/redirectAfterUpload")
+                .outputStream(it -> {
+                    it.write("0123456789".getBytes(StandardCharsets.UTF_8));
+                    it.write("0123456789".getBytes(StandardCharsets.UTF_8));
+                    it.write("0123456789".getBytes(StandardCharsets.UTF_8));
+                    it.close();
+                })) {
+            assertThat(response.entity().as(String.class), is(expected));
+        }
+    }
+
+    @Test
+    void testOutputStreamEntityNotKeptIntercepted() {
+        String expected = "GET plain endpoint reached";
+        try (Http1ClientResponse response = webClient.put()
+                .path("/redirectNoEntity")
+                .outputStream(it -> {
+                    try {
+                        it.write("0123456789".getBytes(StandardCharsets.UTF_8));
+                        it.write("0123456789".getBytes(StandardCharsets.UTF_8));
+                        it.write("0123456789".getBytes(StandardCharsets.UTF_8));
+                        it.close();
+                    } catch (Exception ignore) {
+                    }
+                })) {
+            assertThat(response.entity().as(String.class), is(expected));
+        }
+    }
+
+    @Test
+    void testMaxNumberOfRedirections() {
+        IllegalStateException exception = assertThrows(IllegalStateException.class, () -> webClient.put()
+                .path("/infiniteRedirect")
+                .outputStream(it -> {
+                    it.write("0123456789".getBytes(StandardCharsets.UTF_8));
+                    it.write("0123456789".getBytes(StandardCharsets.UTF_8));
+                    it.close();
+                }));
+        assertThat(exception.getMessage(), is("Maximum number of request redirections (10) reached."));
+    }
+
+    @Test
+    void test100ContinueTimeout() {
+        UncheckedIOException exception = assertThrows(UncheckedIOException.class, () -> webClient.put()
+                .path("/wait")
+                .readContinueTimeout(Duration.ofMillis(200))
+                .outputStream(it -> {
+                    it.write("0123456789".getBytes(StandardCharsets.UTF_8));
+                    it.write("0123456789".getBytes(StandardCharsets.UTF_8));
+                    it.close();
+                }));
+        assertThat(exception.getMessage(), is("java.net.SocketTimeoutException: Read timed out"));
+
+        HttpClientResponse response = webClient
+                .put()
+                .path("/wait")
+                .outputStream(it -> {
+                    it.write("0123456789".getBytes(StandardCharsets.UTF_8));
+                    it.write("0123456789".getBytes(StandardCharsets.UTF_8));
+                    it.close();
+                });
+        assertThat(response.as(String.class), is("Request did not timeout"));
+    }
+
+}

nima/webclient/api/src/main/java/io/helidon/nima/webclient/api/ClientRequest.java

@@ -375,6 +375,16 @@ default <T> ClientResponseTyped<T> outputStream(OutputStreamHandler outputStream
      */
     T readTimeout(Duration readTimeout);
 
+    /**
+     * Read 100-Continue timeout for this request.
+     * This read timeout is used when 100-Continue is sent by the client, before it sends an entity.
+     *
+     * @param readContinueTimeout read 100-Continue timeout duration
+     * @return updated client request
+     * @see HttpClientConfig#readContinueTimeout()
+     */
+    T readContinueTimeout(Duration readContinueTimeout);
+
     /**
      * Handle output stream.
      */

nima/webclient/api/src/main/java/io/helidon/nima/webclient/api/ClientRequestBase.java

@@ -73,6 +73,7 @@ public abstract class ClientRequestBase<T extends ClientRequest<T>, R extends Ht
     private boolean followRedirects;
     private int maxRedirects;
     private Duration readTimeout;
+    private Duration readContinueTimeout;
     private Tls tls;
     private Proxy proxy;
     private boolean keepAlive;
@@ -93,6 +94,7 @@ protected ClientRequestBase(HttpClientConfig clientConfig,
 
         this.headers = clientConfig.defaultRequestHeaders();
         this.readTimeout = clientConfig.socketOptions().readTimeout();
+        this.readContinueTimeout = clientConfig.readContinueTimeout();
         this.mediaContext = clientConfig.mediaContext();
         this.followRedirects = clientConfig.followRedirects();
         this.maxRedirects = clientConfig.maxRedirects();
@@ -226,17 +228,18 @@ public T keepAlive(boolean keepAlive) {
         return identity();
     }
 
-    /**
-     * Read timeout for this request.
-     *
-     * @param readTimeout response read timeout
-     * @return updated client request
-     */
+    @Override
     public T readTimeout(Duration readTimeout) {
         this.readTimeout = readTimeout;
         return identity();
     }
 
+    @Override
+    public T readContinueTimeout(Duration readContinueTimeout) {
+        this.readContinueTimeout = readContinueTimeout;
+        return identity();
+    }
+
     @Override
     public T proxy(Proxy proxy) {
         this.proxy = Objects.requireNonNull(proxy);
@@ -330,6 +333,11 @@ public Duration readTimeout() {
         return readTimeout;
     }
 
+    @Override
+    public Duration readContinueTimeout() {
+        return readContinueTimeout;
+    }
+
     @Override
     public boolean keepAlive() {
         return keepAlive;

nima/webclient/api/src/main/java/io/helidon/nima/webclient/api/FullClientRequest.java

@@ -78,6 +78,13 @@ public interface FullClientRequest<T extends ClientRequest<T>> extends ClientReq
      */
     Duration readTimeout();
 
+    /**
+     * Read 100-Continue timeout.
+     *
+     * @return read 100-Continue timeout of this request
+     */
+    Duration readContinueTimeout();
+
     /**
      * TLS configuration (may be disabled - e.g. use plaintext).
      *

nima/webclient/api/src/main/java/io/helidon/nima/webclient/api/HttpClientConfigBlueprint.java

@@ -16,6 +16,7 @@
 
 package io.helidon.nima.webclient.api;
 
+import java.time.Duration;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -216,4 +217,13 @@ default ClientRequestHeaders defaultRequestHeaders() {
      */
     @ConfiguredOption
     Optional<WebClientCookieManager> cookieManager();
+
+    /**
+     * Socket 100-Continue read timeout. Default is 1 second.
+     * This read timeout is used when 100-Continue is sent by the client, before it sends an entity.
+     *
+     * @return read 100-Continue timeout duration
+     */
+    @ConfiguredOption("PT1S")
+    Duration readContinueTimeout();
 }

nima/webclient/api/src/test/java/io/helidon/nima/webclient/api/HttpClientTest.java

@@ -108,6 +108,11 @@ public FakeHttpClientRequest readTimeout(Duration readTimeout) {
             return this;
         }
 
+        @Override
+        public FakeHttpClientRequest readContinueTimeout(Duration readContinueTimeout) {
+            return this;
+        }
+
         @Override
         public FakeHttpClientRequest tls(Tls tls) {
             return this;