Introduce JettyClientHttpRequestFactory

This commit introduces an implementation of ClientHttpRequestFactory
based on Jetty's HttpClient.

Closes gh-30564

spring-web/src/main/java/org/springframework/http/client/JettyClientHttpRequest.java

@@ -0,0 +1,102 @@
+/*
+ * Copyright 2002-2023 the original author or authors.
+ *
+ * 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
+ *
+ *      https://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 org.springframework.http.client;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.URI;
+import java.time.Duration;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import org.eclipse.jetty.client.api.Request;
+import org.eclipse.jetty.client.api.Response;
+import org.eclipse.jetty.client.util.InputStreamResponseListener;
+import org.eclipse.jetty.client.util.OutputStreamRequestContent;
+
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.HttpMethod;
+import org.springframework.lang.Nullable;
+import org.springframework.util.StreamUtils;
+
+/**
+ * {@link ClientHttpRequest} implementation based on Jetty's
+ * {@link org.eclipse.jetty.client.HttpClient}.
+ *
+ * @author Arjen Poutsma
+ * @since 6.1
+ * @see JettyClientHttpRequestFactory
+ */
+class JettyClientHttpRequest extends AbstractStreamingClientHttpRequest {
+
+	private final Request request;
+
+	private final Duration timeOut;
+
+
+	public JettyClientHttpRequest(Request request, Duration timeOut) {
+		this.request = request;
+		this.timeOut = timeOut;
+	}
+
+	@Override
+	public HttpMethod getMethod() {
+		return HttpMethod.valueOf(this.request.getMethod());
+	}
+
+	@Override
+	public URI getURI() {
+		return this.request.getURI();
+	}
+
+	@Override
+	protected ClientHttpResponse executeInternal(HttpHeaders headers, @Nullable Body body) throws IOException {
+		if (!headers.isEmpty()) {
+			this.request.headers(httpFields -> {
+				headers.forEach((headerName, headerValues) -> {
+					for (String headerValue : headerValues) {
+						httpFields.add(headerName, headerValue);
+					}
+				});
+			});
+		}
+		String contentType = null;
+		if (headers.getContentType() != null) {
+			contentType = headers.getContentType().toString();
+		}
+		try {
+			InputStreamResponseListener responseListener = new InputStreamResponseListener();
+			if (body != null) {
+				OutputStreamRequestContent requestContent = new OutputStreamRequestContent(contentType);
+				this.request.body(requestContent)
+						.send(responseListener);
+				try (OutputStream outputStream = requestContent.getOutputStream()) {
+					body.writeTo(StreamUtils.nonClosing(outputStream));
+				}
+			}
+			else {
+				this.request.send(responseListener);
+			}
+			Response response = responseListener.get(TimeUnit.MILLISECONDS.convert(this.timeOut), TimeUnit.MILLISECONDS);
+			return new JettyClientHttpResponse(response, responseListener.getInputStream());
+		}
+		catch (InterruptedException | TimeoutException | ExecutionException ex) {
+			throw new IOException("Could not send request: " + ex.getMessage(), ex);
+		}
+	}
+}

spring-web/src/main/java/org/springframework/http/client/JettyClientHttpRequestFactory.java

@@ -0,0 +1,110 @@
+/*
+ * Copyright 2002-2023 the original author or authors.
+ *
+ * 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
+ *
+ *      https://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 org.springframework.http.client;
+
+import java.io.IOException;
+import java.net.URI;
+import java.time.Duration;
+
+import org.eclipse.jetty.client.HttpClient;
+import org.eclipse.jetty.client.api.Request;
+
+import org.springframework.beans.factory.DisposableBean;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.http.HttpMethod;
+import org.springframework.util.Assert;
+
+/**
+ * {@link ClientHttpRequestFactory} implementation based on Jetty's {@link HttpClient}.
+ *
+ * @author Arjen Poutsma
+ * @since 6.1
+ * @see <a href="https://www.eclipse.org/jetty/documentation/jetty-11/programming-guide/index.html#pg-client-http">Jetty HttpClient</a>
+ */
+public class JettyClientHttpRequestFactory implements ClientHttpRequestFactory, InitializingBean, DisposableBean {
+
+	private final HttpClient httpClient;
+
+	private final boolean defaultClient;
+
+	private Duration timeOut = Duration.ofSeconds(1);
+
+
+	/**
+	 * Default constructor that creates a new instance of {@link HttpClient}.
+	 */
+	public JettyClientHttpRequestFactory() {
+		this(new HttpClient(), true);
+	}
+
+	/**
+	 * Constructor that takes a customized {@code HttpClient} instance.
+	 * @param httpClient the
+	 */
+	public JettyClientHttpRequestFactory(HttpClient httpClient) {
+		this(httpClient, false);
+	}
+
+	private JettyClientHttpRequestFactory(HttpClient httpClient, boolean defaultClient) {
+		this.httpClient = httpClient;
+		this.defaultClient = defaultClient;
+	}
+
+
+	/**
+	 * Sets the maximum time to wait until all headers have been received.
+	 * The default value is 1 second.
+	 */
+	public void setTimeOut(Duration timeOut) {
+		Assert.notNull(timeOut, "TimeOut must not be null");
+		Assert.isTrue(!timeOut.isNegative(), "TimeOut must not be negative");
+		this.timeOut = timeOut;
+	}
+
+	@Override
+	public void afterPropertiesSet() throws Exception {
+		startHttpClient();
+	}
+
+	private void startHttpClient() throws Exception {
+		if (!this.httpClient.isStarted()) {
+			this.httpClient.start();
+		}
+	}
+
+	@Override
+	public void destroy() throws Exception {
+		if (this.defaultClient) {
+			if (!this.httpClient.isStopped()) {
+				this.httpClient.stop();
+			}
+		}
+	}
+
+	@Override
+	public ClientHttpRequest createRequest(URI uri, HttpMethod httpMethod) throws IOException {
+		try {
+			startHttpClient();
+		}
+		catch (Exception ex) {
+			throw new IOException("Could not start HttpClient: " + ex.getMessage(), ex);
+		}
+
+		Request request = this.httpClient.newRequest(uri).method(httpMethod.name());
+		return new JettyClientHttpRequest(request, this.timeOut);
+	}
+}

spring-web/src/main/java/org/springframework/http/client/JettyClientHttpResponse.java

@@ -0,0 +1,81 @@
+/*
+ * Copyright 2002-2023 the original author or authors.
+ *
+ * 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
+ *
+ *      https://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 org.springframework.http.client;
+
+import java.io.IOException;
+import java.io.InputStream;
+
+import org.eclipse.jetty.client.api.Response;
+
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.HttpStatusCode;
+import org.springframework.util.MultiValueMap;
+
+/**
+ * {@link ClientHttpResponse} implementation based on based on Jetty's
+ * {@link org.eclipse.jetty.client.HttpClient}.
+ *
+ * @author Arjen Poutsma
+ * @since 6.1
+ */
+class JettyClientHttpResponse implements ClientHttpResponse {
+
+	private final Response response;
+
+	private final InputStream body;
+
+	private final HttpHeaders headers;
+
+
+	public JettyClientHttpResponse(Response response, InputStream inputStream) {
+		this.response = response;
+		this.body = inputStream;
+
+		MultiValueMap<String, String> headers = new JettyHeadersAdapter(response.getHeaders());
+		this.headers = HttpHeaders.readOnlyHttpHeaders(headers);
+	}
+
+
+	@Override
+	public HttpStatusCode getStatusCode() throws IOException {
+		return HttpStatusCode.valueOf(this.response.getStatus());
+	}
+
+	@Override
+	public String getStatusText() throws IOException {
+		return this.response.getReason();
+	}
+
+	@Override
+	public HttpHeaders getHeaders() {
+		return this.headers;
+	}
+
+	@Override
+	public InputStream getBody() throws IOException {
+		return this.body;
+	}
+
+	@Override
+	public void close() {
+		try {
+			this.body.close();
+		}
+		catch (IOException ignored) {
+		}
+	}
+}

spring-web/src/main/java/org/springframework/http/client/reactive/JettyHeadersAdapter.java → spring-web/src/main/java/org/springframework/http/client/JettyHeadersAdapter.java

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.http.client.reactive;
+package org.springframework.http.client;
 
 import java.util.AbstractSet;
 import java.util.Collection;
@@ -28,6 +28,7 @@
 
 import org.springframework.http.HttpHeaders;
 import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
 import org.springframework.util.CollectionUtils;
 import org.springframework.util.MultiValueMap;
 
@@ -41,13 +42,20 @@
  * @author Sam Brannen
  * @since 5.3
  */
-class JettyHeadersAdapter implements MultiValueMap<String, String> {
+public final class JettyHeadersAdapter implements MultiValueMap<String, String> {
 
 	private final HttpFields headers;
 
 	private static final String IMMUTABLE_HEADER_ERROR = "Immutable headers";
 
-	JettyHeadersAdapter(HttpFields headers) {
+
+	/**
+	 * Creates a new {@code JettyHeadersAdapter} based on the given
+	 * {@code HttpFields} instance.
+	 * @param headers the {@code HttpFields} to base this adapter on
+	 */
+	public JettyHeadersAdapter(HttpFields headers) {
+		Assert.notNull(headers, "Headers must not be null");
 		this.headers = headers;
 	}
 

spring-web/src/main/java/org/springframework/http/client/reactive/JettyClientHttpRequest.java

@@ -37,6 +37,7 @@
 import org.springframework.http.HttpHeaders;
 import org.springframework.http.HttpMethod;
 import org.springframework.http.MediaType;
+import org.springframework.http.client.JettyHeadersAdapter;
 
 /**
  * {@link ClientHttpRequest} implementation for the Jetty ReactiveStreams HTTP client.

spring-web/src/main/java/org/springframework/http/client/reactive/JettyClientHttpResponse.java

@@ -29,6 +29,7 @@
 import org.springframework.http.HttpHeaders;
 import org.springframework.http.HttpStatusCode;
 import org.springframework.http.ResponseCookie;
+import org.springframework.http.client.JettyHeadersAdapter;
 import org.springframework.lang.Nullable;
 import org.springframework.util.CollectionUtils;
 import org.springframework.util.LinkedMultiValueMap;

spring-web/src/test/java/org/springframework/http/client/AbstractHttpRequestFactoryTests.java

@@ -111,20 +111,19 @@ void multipleWrites() throws Exception {
 		ClientHttpRequest request = factory.createRequest(URI.create(baseUrl + "/echo"), HttpMethod.POST);
 
 		final byte[] body = "Hello World".getBytes(StandardCharsets.UTF_8);
+		request.getHeaders().setContentLength(body.length);
 		if (request instanceof StreamingHttpOutputMessage streamingRequest) {
-			streamingRequest.setBody(outputStream -> {
-				StreamUtils.copy(body, outputStream);
-				outputStream.flush();
-				outputStream.close();
-			});
+			streamingRequest.setBody(outputStream -> StreamUtils.copy(body, outputStream));
 		}
 		else {
 			StreamUtils.copy(body, request.getBody());
 		}
 
-		request.execute();
-		assertThatIllegalStateException().isThrownBy(() ->
-				FileCopyUtils.copy(body, request.getBody()));
+		try (ClientHttpResponse response = request.execute()) {
+			assertThatIllegalStateException().isThrownBy(() ->
+					FileCopyUtils.copy(body, request.getBody()));
+			assertThat(response.getStatusCode()).as("Invalid status code").isEqualTo(HttpStatus.OK);
+		}
 	}
 
 	@Test