Support Multipart with HelidonConnector/WebClient

jersey/connector/pom.xml

@@ -50,6 +50,21 @@
             <groupId>io.helidon.webclient</groupId>
             <artifactId>helidon-webclient</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.glassfish.jersey.media</groupId>
+            <artifactId>jersey-media-multipart</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>io.helidon.microprofile.tests</groupId>
+            <artifactId>helidon-microprofile-tests-junit5</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>io.helidon.microprofile.server</groupId>
+            <artifactId>helidon-microprofile-server</artifactId>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>io.helidon.config</groupId>
             <artifactId>helidon-config-testing</artifactId>

jersey/connector/src/main/java/io/helidon/jersey/connector/HelidonConnector.java

@@ -60,7 +60,7 @@
 class HelidonConnector implements Connector {
 
     private static final String HELIDON_VERSION = "Helidon/" + Version.VERSION + " (java "
-            + PropertiesHelper.getSystemProperty("java.runtime.version") + ")";
+            + PropertiesHelper.getSystemProperty("java.runtime.version").run() + ")";
     static final Logger LOGGER = Logger.getLogger(HelidonConnector.class.getName());
 
     private final WebClient webClient;
@@ -148,8 +148,6 @@ private CompletionStage<ClientResponse> applyInternal(ClientRequest request) {
         final WebClientRequestBuilder webClientRequestBuilder = webClient.method(request.getMethod());
         webClientRequestBuilder.uri(request.getUri());
 
-        webClientRequestBuilder.headers(HelidonStructures.createHeaders(request.getRequestHeaders()));
-
         for (String propertyName : request.getConfiguration().getPropertyNames()) {
             Object property = request.getConfiguration().getProperty(propertyName);
             if (!propertyName.startsWith("jersey") && String.class.isInstance(property)) {
@@ -177,6 +175,7 @@ private CompletionStage<ClientResponse> applyInternal(ClientRequest request) {
                     entityType, request, webClientRequestBuilder, executorServiceKeeper.getExecutorService(request)
             );
         } else {
+            webClientRequestBuilder.headers(HelidonStructures.createHeaders(request.getRequestHeaders()));
             responseStage = webClientRequestBuilder.submit();
         }
 

jersey/connector/src/main/java/io/helidon/jersey/connector/HelidonEntity.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2020, 2021 Oracle and/or its affiliates.
+ * Copyright (c) 2020, 2024 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.
@@ -19,6 +19,7 @@
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Flow;
@@ -106,27 +107,45 @@ static CompletionStage<WebClientResponse> submit(HelidonEntityType type,
         if (type != null) {
             final int bufferSize = requestContext.resolveProperty(
                     ClientProperties.OUTBOUND_CONTENT_LENGTH_BUFFER, 8192);
+            CompletableFuture<Void> firstWrite = new CompletableFuture<>();
             switch (type) {
                 case BYTE_ARRAY_OUTPUT_STREAM:
                     final ByteArrayOutputStream baos = new ByteArrayOutputStream(bufferSize);
-                    requestContext.setStreamProvider(contentLength -> baos);
-                    ((ProcessingRunnable) requestContext::writeEntity).run();
-                    stage = requestBuilder.submit(baos);
+                    requestContext.setStreamProvider(unused -> {
+                        requestBuilder.headers(HelidonStructures.createHeaders(requestContext.getRequestHeaders()));
+                        firstWrite.complete(null);
+                        return baos;
+                    });
+                    try {
+                        requestContext.writeEntity();
+                        stage = requestBuilder.submit(baos);
+                    } catch (IOException e) {
+                        stage = CompletableFuture.failedStage(e);
+                    }
                     break;
                 case READABLE_BYTE_CHANNEL:
                     final OutputStreamChannel channel = new OutputStreamChannel(bufferSize);
-                    requestContext.setStreamProvider(contentLength -> channel);
+                    requestContext.setStreamProvider(unused -> {
+                        requestBuilder.headers(HelidonStructures.createHeaders(requestContext.getRequestHeaders()));
+                        firstWrite.complete(null);
+                        return channel;
+                    });
                     executorService.execute((ProcessingRunnable) requestContext::writeEntity);
-                    stage = requestBuilder.submit(channel);
+                    stage = firstWrite.thenCompose(unused -> requestBuilder.submit(channel));
                     break;
                 case OUTPUT_STREAM_MULTI:
                     final OutputStreamMulti publisher = IoMulti.outputStreamMulti();
-                    requestContext.setStreamProvider(contentLength -> publisher);
+                    requestContext.setStreamProvider(unused -> {
+                        requestBuilder.headers(HelidonStructures.createHeaders(requestContext.getRequestHeaders()));
+                        firstWrite.complete(null);
+                        return publisher;
+                    });
                     executorService.execute((ProcessingRunnable) () -> {
                         requestContext.writeEntity();
                         publisher.close();
                     });
-                    stage = requestBuilder.submit(Multi.create(publisher).map(DataChunk::create));
+                    Multi<DataChunk> m = Multi.create(publisher).map(DataChunk::create);
+                    stage = firstWrite.thenCompose(unused -> requestBuilder.submit(m));
                     break;
                 default:
             }

jersey/connector/src/test/java/io/helidon/jersey/connector/MultipartTest.java

@@ -0,0 +1,105 @@
+/*
+ * Copyright (c) 2024 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.jersey.connector;
+
+import io.helidon.microprofile.server.JaxRsCdiExtension;
+import io.helidon.microprofile.server.ServerCdiExtension;
+import io.helidon.microprofile.tests.junit5.AddBean;
+import io.helidon.microprofile.tests.junit5.AddExtension;
+import io.helidon.microprofile.tests.junit5.DisableDiscovery;
+import io.helidon.microprofile.tests.junit5.HelidonTest;
+import org.glassfish.jersey.ext.cdi1x.internal.CdiComponentProvider;
+import org.glassfish.jersey.media.multipart.BodyPart;
+import org.glassfish.jersey.media.multipart.BodyPartEntity;
+import org.glassfish.jersey.media.multipart.MultiPart;
+import org.glassfish.jersey.media.multipart.MultiPartFeature;
+import org.glassfish.jersey.message.internal.ReaderWriter;
+
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+
+import jakarta.ws.rs.Consumes;
+import jakarta.ws.rs.POST;
+import jakarta.ws.rs.Path;
+import jakarta.ws.rs.client.ClientBuilder;
+import jakarta.ws.rs.client.Entity;
+import jakarta.ws.rs.client.WebTarget;
+import jakarta.ws.rs.core.Application;
+import jakarta.ws.rs.core.Context;
+import jakarta.ws.rs.core.HttpHeaders;
+import jakarta.ws.rs.core.MediaType;
+import jakarta.ws.rs.core.Response;
+
+import java.io.IOException;
+import java.util.HashSet;
+import java.util.Set;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+
+@HelidonTest
+@DisableDiscovery
+@AddExtension(ServerCdiExtension.class)
+@AddExtension(JaxRsCdiExtension.class)
+@AddExtension(CdiComponentProvider.class)
+
+@AddBean(MultipartTest.MultipartApplication.class)
+public class MultipartTest {
+    private static final String ENTITY = "hello";
+
+    public static class MultipartApplication extends Application {
+        @Override
+        public Set<Class<?>> getClasses() {
+            Set<Class<?>> set = new HashSet<>();
+            set.add(MultiPartFeature.class);
+            set.add(MultipartResource.class);
+            return set;
+        }
+    }
+
+    @Path("/")
+    public static class MultipartResource {
+        @POST
+        @Path("upload")
+        @Consumes(MediaType.MULTIPART_FORM_DATA)
+        public String upload(@Context HttpHeaders headers, MultiPart multiPart) throws IOException {
+            return ReaderWriter.readFromAsString(
+                    ((BodyPartEntity) multiPart.getBodyParts().get(0).getEntity()).getInputStream(),
+                    MediaType.TEXT_PLAIN_TYPE);
+        }
+    }
+
+    @ParameterizedTest
+    @EnumSource(value = HelidonEntity.HelidonEntityType.class)
+    void testMultipart(HelidonEntity.HelidonEntityType entityType, WebTarget webTarget) {
+        // For each entity type make 10 consecutive requests
+        for (int i = 0; i != 10; i++) {
+            MultiPart multipart = new MultiPart().bodyPart(new BodyPart().entity(ENTITY));
+            multipart.setMediaType(MediaType.MULTIPART_FORM_DATA_TYPE);
+            try (Response r = ClientBuilder.newBuilder()
+                    .property(HelidonConnector.INTERNAL_ENTITY_TYPE, entityType.name())
+                    .build().target(webTarget.getUri())
+                    .path("upload")
+                    .register(MultiPartFeature.class)
+                    .request()
+                    .post(Entity.entity(multipart, multipart.getMediaType()))) {
+                assertThat(r.getStatus(), is(Response.Status.OK.getStatusCode()));
+                assertThat(r.readEntity(String.class), is(ENTITY));
+            }
+        }
+    }
+}