Support for WebFlux on Reactor Netty 2 with Netty 5

See gh-28847

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

@@ -0,0 +1,283 @@
+/*
+ * Copyright 2002-2022 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.reactive;
+
+import java.util.AbstractSet;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import io.netty5.handler.codec.http.HttpHeaders;
+
+import org.springframework.lang.Nullable;
+import org.springframework.util.CollectionUtils;
+import org.springframework.util.MultiValueMap;
+
+/**
+ * {@code MultiValueMap} implementation for wrapping Netty HTTP headers.
+ *
+ * <p>There is a duplicate of this class in the server package!
+ *
+ * This class is based on {@link NettyHeadersAdapter}.
+ *
+ * @author Violeta Georgieva
+ * @since 6.0
+ */
+class Netty5HeadersAdapter implements MultiValueMap<String, String> {
+
+	private final HttpHeaders headers;
+
+
+	Netty5HeadersAdapter(HttpHeaders headers) {
+		this.headers = headers;
+	}
+
+
+	@Override
+	@Nullable
+	public String getFirst(String key) {
+		return this.headers.get(key);
+	}
+
+	@Override
+	public void add(String key, @Nullable String value) {
+		if (value != null) {
+			this.headers.add(key, value);
+		}
+	}
+
+	@Override
+	public void addAll(String key, List<? extends String> values) {
+		this.headers.add(key, values);
+	}
+
+	@Override
+	public void addAll(MultiValueMap<String, String> values) {
+		values.forEach(this.headers::add);
+	}
+
+	@Override
+	public void set(String key, @Nullable String value) {
+		if (value != null) {
+			this.headers.set(key, value);
+		}
+	}
+
+	@Override
+	public void setAll(Map<String, String> values) {
+		values.forEach(this.headers::set);
+	}
+
+	@Override
+	public Map<String, String> toSingleValueMap() {
+		Map<String, String> singleValueMap = CollectionUtils.newLinkedHashMap(this.headers.size());
+		this.headers.entries()
+				.forEach(entry -> {
+					if (!singleValueMap.containsKey(entry.getKey())) {
+						singleValueMap.put(entry.getKey(), entry.getValue());
+					}
+				});
+		return singleValueMap;
+	}
+
+	@Override
+	public int size() {
+		return this.headers.names().size();
+	}
+
+	@Override
+	public boolean isEmpty() {
+		return this.headers.isEmpty();
+	}
+
+	@Override
+	public boolean containsKey(Object key) {
+		return (key instanceof String headerName && this.headers.contains(headerName));
+	}
+
+	@Override
+	public boolean containsValue(Object value) {
+		return (value instanceof String &&
+				this.headers.entries().stream()
+						.anyMatch(entry -> value.equals(entry.getValue())));
+	}
+
+	@Override
+	@Nullable
+	public List<String> get(Object key) {
+		if (containsKey(key)) {
+			return this.headers.getAll((String) key);
+		}
+		return null;
+	}
+
+	@Nullable
+	@Override
+	public List<String> put(String key, @Nullable List<String> value) {
+		List<String> previousValues = this.headers.getAll(key);
+		this.headers.set(key, value);
+		return previousValues;
+	}
+
+	@Nullable
+	@Override
+	public List<String> remove(Object key) {
+		if (key instanceof String headerName) {
+			List<String> previousValues = this.headers.getAll(headerName);
+			this.headers.remove(headerName);
+			return previousValues;
+		}
+		return null;
+	}
+
+	@Override
+	public void putAll(Map<? extends String, ? extends List<String>> map) {
+		map.forEach(this.headers::set);
+	}
+
+	@Override
+	public void clear() {
+		this.headers.clear();
+	}
+
+	@Override
+	public Set<String> keySet() {
+		return new HeaderNames();
+	}
+
+	@Override
+	public Collection<List<String>> values() {
+		return this.headers.names().stream()
+				.map(this.headers::getAll).collect(Collectors.toList());
+	}
+
+	@Override
+	public Set<Entry<String, List<String>>> entrySet() {
+		return new AbstractSet<>() {
+			@Override
+			public Iterator<Entry<String, List<String>>> iterator() {
+				return new EntryIterator();
+			}
+
+			@Override
+			public int size() {
+				return headers.size();
+			}
+		};
+	}
+
+
+	@Override
+	public String toString() {
+		return org.springframework.http.HttpHeaders.formatHeaders(this);
+	}
+
+
+	private class EntryIterator implements Iterator<Entry<String, List<String>>> {
+
+		private final Iterator<String> names = headers.names().iterator();
+
+		@Override
+		public boolean hasNext() {
+			return this.names.hasNext();
+		}
+
+		@Override
+		public Entry<String, List<String>> next() {
+			return new HeaderEntry(this.names.next());
+		}
+	}
+
+
+	private class HeaderEntry implements Entry<String, List<String>> {
+
+		private final String key;
+
+		HeaderEntry(String key) {
+			this.key = key;
+		}
+
+		@Override
+		public String getKey() {
+			return this.key;
+		}
+
+		@Override
+		public List<String> getValue() {
+			return headers.getAll(this.key);
+		}
+
+		@Override
+		public List<String> setValue(List<String> value) {
+			List<String> previousValues = headers.getAll(this.key);
+			headers.set(this.key, value);
+			return previousValues;
+		}
+	}
+
+
+	private class HeaderNames extends AbstractSet<String> {
+
+		@Override
+		public Iterator<String> iterator() {
+			return new HeaderNamesIterator(headers.names().iterator());
+		}
+
+		@Override
+		public int size() {
+			return headers.names().size();
+		}
+	}
+
+	private final class HeaderNamesIterator implements Iterator<String> {
+
+		private final Iterator<String> iterator;
+
+		@Nullable
+		private String currentName;
+
+		private HeaderNamesIterator(Iterator<String> iterator) {
+			this.iterator = iterator;
+		}
+
+		@Override
+		public boolean hasNext() {
+			return this.iterator.hasNext();
+		}
+
+		@Override
+		public String next() {
+			this.currentName = this.iterator.next();
+			return this.currentName;
+		}
+
+		@Override
+		public void remove() {
+			if (this.currentName == null) {
+				throw new IllegalStateException("No current Header in iterator");
+			}
+			if (!headers.contains(this.currentName)) {
+				throw new IllegalStateException("Header not present: " + this.currentName);
+			}
+			headers.remove(this.currentName);
+		}
+	}
+
+}

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

@@ -0,0 +1,132 @@
+/*
+ * Copyright 2002-2022 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.reactive;
+
+import java.net.URI;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Function;
+
+import reactor.core.publisher.Mono;
+import reactor.netty5.NettyOutbound;
+import reactor.netty5.http.client.HttpClient;
+import reactor.netty5.http.client.HttpClientRequest;
+import reactor.netty5.resources.ConnectionProvider;
+import reactor.netty5.resources.LoopResources;
+
+import org.springframework.http.HttpMethod;
+import org.springframework.util.Assert;
+
+/**
+ * Reactor Netty 2 (Netty 5) implementation of {@link ClientHttpConnector}.
+ *
+ * <p>This class is based on {@link ReactorClientHttpConnector}.
+ *
+ * @author Violeta Georgieva
+ * @since 6.0
+ * @see HttpClient
+ */
+public class ReactorNetty2ClientHttpConnector implements ClientHttpConnector {
+
+	private final static Function<HttpClient, HttpClient> defaultInitializer = client -> client.compress(true);
+
+
+	private final HttpClient httpClient;
+
+
+	/**
+	 * Default constructor. Initializes {@link HttpClient} via:
+	 * <pre class="code">
+	 * HttpClient.create().compress()
+	 * </pre>
+	 */
+	public ReactorNetty2ClientHttpConnector() {
+		this.httpClient = defaultInitializer.apply(HttpClient.create().wiretap(true));
+	}
+
+	/**
+	 * Constructor with externally managed Reactor Netty resources, including
+	 * {@link LoopResources} for event loop threads, and {@link ConnectionProvider}
+	 * for the connection pool.
+	 * <p>This constructor should be used only when you don't want the client
+	 * to participate in the Reactor Netty global resources. By default, the
+	 * client participates in the Reactor Netty global resources held in
+	 * {@link reactor.netty5.http.HttpResources}, which is recommended since
+	 * fixed, shared resources are favored for event loop concurrency. However,
+	 * consider declaring a {@link ReactorNetty2ResourceFactory} bean with
+	 * {@code globalResources=true} in order to ensure the Reactor Netty global
+	 * resources are shut down when the Spring ApplicationContext is closed.
+	 * @param factory the resource factory to obtain the resources from
+	 * @param mapper a mapper for further initialization of the created client
+	 * @since 5.1
+	 */
+	public ReactorNetty2ClientHttpConnector(ReactorNetty2ResourceFactory factory, Function<HttpClient, HttpClient> mapper) {
+		ConnectionProvider provider = factory.getConnectionProvider();
+		Assert.notNull(provider, "No ConnectionProvider: is ReactorNetty2ResourceFactory not initialized yet?");
+		this.httpClient = defaultInitializer.andThen(mapper).andThen(applyLoopResources(factory))
+				.apply(HttpClient.create(provider));
+	}
+
+	private static Function<HttpClient, HttpClient> applyLoopResources(ReactorNetty2ResourceFactory factory) {
+		return httpClient -> {
+			LoopResources resources = factory.getLoopResources();
+			Assert.notNull(resources, "No LoopResources: is ReactorNetty2ResourceFactory not initialized yet?");
+			return httpClient.runOn(resources);
+		};
+	}
+
+
+	/**
+	 * Constructor with a pre-configured {@code HttpClient} instance.
+	 * @param httpClient the client to use
+	 * @since 5.1
+	 */
+	public ReactorNetty2ClientHttpConnector(HttpClient httpClient) {
+		Assert.notNull(httpClient, "HttpClient is required");
+		this.httpClient = httpClient;
+	}
+
+
+	@Override
+	public Mono<ClientHttpResponse> connect(HttpMethod method, URI uri,
+			Function<? super ClientHttpRequest, Mono<Void>> requestCallback) {
+
+		AtomicReference<ReactorNetty2ClientHttpResponse> responseRef = new AtomicReference<>();
+
+		return this.httpClient
+				.request(io.netty5.handler.codec.http.HttpMethod.valueOf(method.name()))
+				.uri(uri.toString())
+				.send((request, outbound) -> requestCallback.apply(adaptRequest(method, uri, request, outbound)))
+				.responseConnection((response, connection) -> {
+					responseRef.set(new ReactorNetty2ClientHttpResponse(response, connection));
+					return Mono.just((ClientHttpResponse) responseRef.get());
+				})
+				.next()
+				.doOnCancel(() -> {
+					ReactorNetty2ClientHttpResponse response = responseRef.get();
+					if (response != null) {
+						response.releaseAfterCancel(method);
+					}
+				});
+	}
+
+	private ReactorNetty2ClientHttpRequest adaptRequest(HttpMethod method, URI uri, HttpClientRequest request,
+			NettyOutbound nettyOutbound) {
+
+		return new ReactorNetty2ClientHttpRequest(method, uri, request, nettyOutbound);
+	}
+
+}

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

@@ -0,0 +1,143 @@
+/*
+ * Copyright 2002-2022 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.reactive;
+
+import java.net.URI;
+import java.nio.file.Path;
+import java.util.Collection;
+
+import io.netty5.buffer.api.Buffer;
+import io.netty5.handler.codec.http.cookie.DefaultCookie;
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.netty5.NettyOutbound;
+import reactor.netty5.http.client.HttpClientRequest;
+
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.DataBufferFactory;
+import org.springframework.core.io.buffer.Netty5DataBufferFactory;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.HttpMethod;
+import org.springframework.http.ZeroCopyHttpOutputMessage;
+
+/**
+ * {@link ClientHttpRequest} implementation for the Reactor Netty 2 (Netty 5) HTTP client.
+ *
+ * <p>This class is based on {@link ReactorClientHttpRequest}.
+ *
+ * @author Violeta Georgieva
+ * @since 6.0
+ * @see reactor.netty5.http.client.HttpClient
+ */
+class ReactorNetty2ClientHttpRequest extends AbstractClientHttpRequest implements ZeroCopyHttpOutputMessage {
+
+	private final HttpMethod httpMethod;
+
+	private final URI uri;
+
+	private final HttpClientRequest request;
+
+	private final NettyOutbound outbound;
+
+	private final Netty5DataBufferFactory bufferFactory;
+
+
+	public ReactorNetty2ClientHttpRequest(HttpMethod method, URI uri, HttpClientRequest request, NettyOutbound outbound) {
+		this.httpMethod = method;
+		this.uri = uri;
+		this.request = request;
+		this.outbound = outbound;
+		this.bufferFactory = new Netty5DataBufferFactory(outbound.alloc());
+	}
+
+
+	@Override
+	public HttpMethod getMethod() {
+		return this.httpMethod;
+	}
+
+	@Override
+	public URI getURI() {
+		return this.uri;
+	}
+
+	@Override
+	public DataBufferFactory bufferFactory() {
+		return this.bufferFactory;
+	}
+
+	@Override
+	@SuppressWarnings("unchecked")
+	public <T> T getNativeRequest() {
+		return (T) this.request;
+	}
+
+	@Override
+	public Mono<Void> writeWith(Publisher<? extends DataBuffer> body) {
+		return doCommit(() -> {
+			// Send as Mono if possible as an optimization hint to Reactor Netty
+			if (body instanceof Mono) {
+				Mono<Buffer> bufferMono = Mono.from(body).map(Netty5DataBufferFactory::toBuffer);
+				return this.outbound.send(bufferMono).then();
+
+			}
+			else {
+				Flux<Buffer> bufferFlux = Flux.from(body).map(Netty5DataBufferFactory::toBuffer);
+				return this.outbound.send(bufferFlux).then();
+			}
+		});
+	}
+
+	@Override
+	public Mono<Void> writeAndFlushWith(Publisher<? extends Publisher<? extends DataBuffer>> body) {
+		Publisher<Publisher<Buffer>> buffers = Flux.from(body).map(ReactorNetty2ClientHttpRequest::toBuffers);
+		return doCommit(() -> this.outbound.sendGroups(buffers).then());
+	}
+
+	private static Publisher<Buffer> toBuffers(Publisher<? extends DataBuffer> dataBuffers) {
+		return Flux.from(dataBuffers).map(Netty5DataBufferFactory::toBuffer);
+	}
+
+	@Override
+	public Mono<Void> writeWith(Path file, long position, long count) {
+		return doCommit(() -> this.outbound.sendFile(file, position, count).then());
+	}
+
+	@Override
+	public Mono<Void> setComplete() {
+		return doCommit(this.outbound::then);
+	}
+
+	@Override
+	protected void applyHeaders() {
+		getHeaders().forEach((key, value) -> this.request.requestHeaders().set(key, value));
+	}
+
+	@Override
+	protected void applyCookies() {
+		getCookies().values().stream().flatMap(Collection::stream)
+				.map(cookie -> new DefaultCookie(cookie.getName(), cookie.getValue()))
+				.forEach(this.request::addCookie);
+	}
+
+	@Override
+	protected HttpHeaders initReadOnlyHeaders() {
+		return HttpHeaders.readOnlyHttpHeaders(new Netty5HeadersAdapter(this.request.requestHeaders()));
+	}
+
+}

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

@@ -0,0 +1,214 @@
+/*
+ * Copyright 2002-2022 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.reactive;
+
+import java.util.Collection;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.BiFunction;
+
+import io.netty5.buffer.api.BufferAllocator;
+import io.netty5.handler.codec.http.cookie.Cookie;
+import io.netty5.handler.codec.http.cookie.DefaultCookie;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import reactor.core.publisher.Flux;
+import reactor.netty5.Connection;
+import reactor.netty5.NettyInbound;
+import reactor.netty5.http.client.HttpClientResponse;
+
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.Netty5DataBufferFactory;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.HttpMethod;
+import org.springframework.http.HttpStatusCode;
+import org.springframework.http.ResponseCookie;
+import org.springframework.lang.Nullable;
+import org.springframework.util.ClassUtils;
+import org.springframework.util.CollectionUtils;
+import org.springframework.util.LinkedMultiValueMap;
+import org.springframework.util.MultiValueMap;
+import org.springframework.util.ObjectUtils;
+
+/**
+ * {@link ClientHttpResponse} implementation for the Reactor Netty 2 (Netty 5) HTTP client.
+ *
+ * <p>This class is based on {@link ReactorClientHttpResponse}.
+ *
+ * @author Violeta Georgieva
+ * @since 6.0
+ * @see reactor.netty5.http.client.HttpClient
+ */
+class ReactorNetty2ClientHttpResponse implements ClientHttpResponse {
+
+	/** Reactor Netty 1.0.5+. */
+	static final boolean reactorNettyRequestChannelOperationsIdPresent = ClassUtils.isPresent(
+			"reactor.netty5.ChannelOperationsId", ReactorNetty2ClientHttpResponse.class.getClassLoader());
+
+
+	private static final Log logger = LogFactory.getLog(ReactorNetty2ClientHttpResponse.class);
+
+	private final HttpClientResponse response;
+
+	private final HttpHeaders headers;
+
+	private final NettyInbound inbound;
+
+	private final Netty5DataBufferFactory bufferFactory;
+
+	// 0 - not subscribed, 1 - subscribed, 2 - cancelled via connector (before subscribe)
+	private final AtomicInteger state = new AtomicInteger();
+
+
+	/**
+	 * Constructor that matches the inputs from
+	 * {@link reactor.netty5.http.client.HttpClient.ResponseReceiver#responseConnection(BiFunction)}.
+	 * @since 5.2.8
+	 */
+	public ReactorNetty2ClientHttpResponse(HttpClientResponse response, Connection connection) {
+		this.response = response;
+		MultiValueMap<String, String> adapter = new Netty5HeadersAdapter(response.responseHeaders());
+		this.headers = HttpHeaders.readOnlyHttpHeaders(adapter);
+		this.inbound = connection.inbound();
+		this.bufferFactory = new Netty5DataBufferFactory(connection.outbound().alloc());
+	}
+
+	/**
+	 * Constructor with inputs extracted from a {@link Connection}.
+	 * @deprecated as of 5.2.8, in favor of {@link #ReactorNetty2ClientHttpResponse(HttpClientResponse, Connection)}
+	 */
+	@Deprecated
+	public ReactorNetty2ClientHttpResponse(HttpClientResponse response, NettyInbound inbound, BufferAllocator alloc) {
+		this.response = response;
+		MultiValueMap<String, String> adapter = new Netty5HeadersAdapter(response.responseHeaders());
+		this.headers = HttpHeaders.readOnlyHttpHeaders(adapter);
+		this.inbound = inbound;
+		this.bufferFactory = new Netty5DataBufferFactory(alloc);
+	}
+
+
+	@Override
+	public String getId() {
+		String id = null;
+		if (reactorNettyRequestChannelOperationsIdPresent) {
+			id = ChannelOperationsIdHelper.getId(this.response);
+		}
+		if (id == null && this.response instanceof Connection connection) {
+			id = connection.channel().id().asShortText();
+		}
+		return (id != null ? id : ObjectUtils.getIdentityHexString(this));
+	}
+
+	@Override
+	public Flux<DataBuffer> getBody() {
+		return this.inbound.receive()
+				.doOnSubscribe(s -> {
+					if (this.state.compareAndSet(0, 1)) {
+						return;
+					}
+					if (this.state.get() == 2) {
+						throw new IllegalStateException(
+								"The client response body has been released already due to cancellation.");
+					}
+				})
+				.map(buffer -> this.bufferFactory.wrap(buffer.split()));
+	}
+
+	@Override
+	public HttpHeaders getHeaders() {
+		return this.headers;
+	}
+
+	@Override
+	public HttpStatusCode getStatusCode() {
+		return HttpStatusCode.valueOf(this.response.status().code());
+	}
+
+	@Override
+	@Deprecated
+	public int getRawStatusCode() {
+		return this.response.status().code();
+	}
+
+	@Override
+	public MultiValueMap<String, ResponseCookie> getCookies() {
+		MultiValueMap<String, ResponseCookie> result = new LinkedMultiValueMap<>();
+		this.response.cookies().values().stream()
+				.flatMap(Collection::stream)
+				.forEach(cookie -> result.add(cookie.name(),
+						ResponseCookie.fromClientResponse(cookie.name(), cookie.value())
+								.domain(cookie.domain())
+								.path(cookie.path())
+								.maxAge(cookie.maxAge())
+								.secure(cookie.isSecure())
+								.httpOnly(cookie.isHttpOnly())
+								.sameSite(getSameSite(cookie))
+								.build()));
+		return CollectionUtils.unmodifiableMultiValueMap(result);
+	}
+
+	@Nullable
+	private static String getSameSite(Cookie cookie) {
+		if (cookie instanceof DefaultCookie defaultCookie) {
+			if (defaultCookie.sameSite() != null) {
+				return defaultCookie.sameSite().name();
+			}
+		}
+		return null;
+	}
+
+	/**
+	 * Called by {@link ReactorNetty2ClientHttpConnector} when a cancellation is detected
+	 * but the content has not been subscribed to. If the subscription never
+	 * materializes then the content will remain not drained. Or it could still
+	 * materialize if the cancellation happened very early, or the response
+	 * reading was delayed for some reason.
+	 */
+	void releaseAfterCancel(HttpMethod method) {
+		if (mayHaveBody(method) && this.state.compareAndSet(0, 2)) {
+			if (logger.isDebugEnabled()) {
+				logger.debug("[" + getId() + "]" + "Releasing body, not yet subscribed.");
+			}
+			this.inbound.receive().doOnNext(buffer -> {}).subscribe(buffer -> {}, ex -> {});
+		}
+	}
+
+	private boolean mayHaveBody(HttpMethod method) {
+		int code = this.getRawStatusCode();
+		return !((code >= 100 && code < 200) || code == 204 || code == 205 ||
+				method.equals(HttpMethod.HEAD) || getHeaders().getContentLength() == 0);
+	}
+
+	@Override
+	public String toString() {
+		return "ReactorNetty2ClientHttpResponse{" +
+				"request=[" + this.response.method().name() + " " + this.response.uri() + "]," +
+				"status=" + getRawStatusCode() + '}';
+	}
+
+
+	private static class ChannelOperationsIdHelper {
+
+		@Nullable
+		public static String getId(HttpClientResponse response) {
+			if (response instanceof reactor.netty5.ChannelOperationsId id) {
+				return (logger.isDebugEnabled() ? id.asLongText() : id.asShortText());
+			}
+			return null;
+		}
+	}
+
+}

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

@@ -0,0 +1,250 @@
+/*
+ * Copyright 2002-2022 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.reactive;
+
+import java.time.Duration;
+import java.util.function.Consumer;
+import java.util.function.Supplier;
+
+import reactor.netty5.http.HttpResources;
+import reactor.netty5.resources.ConnectionProvider;
+import reactor.netty5.resources.LoopResources;
+
+import org.springframework.beans.factory.DisposableBean;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
+
+/**
+ * Factory to manage Reactor Netty resources, i.e. {@link LoopResources} for
+ * event loop threads, and {@link ConnectionProvider} for the connection pool,
+ * within the lifecycle of a Spring {@code ApplicationContext}.
+ *
+ * <p>This factory implements {@link InitializingBean} and {@link DisposableBean}
+ * and is expected typically to be declared as a Spring-managed bean.
+ *
+ * <p>This class is based on {@link ReactorResourceFactory}.
+ *
+ * @author Violeta Georgieva
+ * @since 6.0
+ */
+public class ReactorNetty2ResourceFactory implements InitializingBean, DisposableBean {
+
+	private boolean useGlobalResources = true;
+
+	@Nullable
+	private Consumer<HttpResources> globalResourcesConsumer;
+
+	private Supplier<ConnectionProvider> connectionProviderSupplier = () -> ConnectionProvider.create("webflux", 500);
+
+	@Nullable
+	private ConnectionProvider connectionProvider;
+
+	private Supplier<LoopResources> loopResourcesSupplier = () -> LoopResources.create("webflux-http");
+
+	@Nullable
+	private LoopResources loopResources;
+
+	private boolean manageConnectionProvider = false;
+
+	private boolean manageLoopResources = false;
+
+	private Duration shutdownQuietPeriod = Duration.ofSeconds(LoopResources.DEFAULT_SHUTDOWN_QUIET_PERIOD);
+
+	private Duration shutdownTimeout = Duration.ofSeconds(LoopResources.DEFAULT_SHUTDOWN_TIMEOUT);
+
+
+	/**
+	 * Whether to use global Reactor Netty resources via {@link HttpResources}.
+	 * <p>Default is "true" in which case this factory initializes and stops the
+	 * global Reactor Netty resources within Spring's {@code ApplicationContext}
+	 * lifecycle. If set to "false" the factory manages its resources independent
+	 * of the global ones.
+	 * @param useGlobalResources whether to expose and manage the global resources
+	 * @see #addGlobalResourcesConsumer(Consumer)
+	 */
+	public void setUseGlobalResources(boolean useGlobalResources) {
+		this.useGlobalResources = useGlobalResources;
+	}
+
+	/**
+	 * Whether this factory exposes the global
+	 * {@link HttpResources HttpResources} holder.
+	 */
+	public boolean isUseGlobalResources() {
+		return this.useGlobalResources;
+	}
+
+	/**
+	 * Add a Consumer for configuring the global Reactor Netty resources on
+	 * startup. When this option is used, {@link #setUseGlobalResources} is also
+	 * enabled.
+	 * @param consumer the consumer to apply
+	 * @see #setUseGlobalResources(boolean)
+	 */
+	public void addGlobalResourcesConsumer(Consumer<HttpResources> consumer) {
+		this.useGlobalResources = true;
+		this.globalResourcesConsumer = (this.globalResourcesConsumer != null ?
+				this.globalResourcesConsumer.andThen(consumer) : consumer);
+	}
+
+	/**
+	 * Use this when you don't want to participate in global resources and
+	 * you want to customize the creation of the managed {@code ConnectionProvider}.
+	 * <p>By default, {@code ConnectionProvider.elastic("http")} is used.
+	 * <p>Note that this option is ignored if {@code userGlobalResources=false} or
+	 * {@link #setConnectionProvider(ConnectionProvider)} is set.
+	 * @param supplier the supplier to use
+	 */
+	public void setConnectionProviderSupplier(Supplier<ConnectionProvider> supplier) {
+		this.connectionProviderSupplier = supplier;
+	}
+
+	/**
+	 * Use this when you want to provide an externally managed
+	 * {@link ConnectionProvider} instance.
+	 * @param connectionProvider the connection provider to use as is
+	 */
+	public void setConnectionProvider(ConnectionProvider connectionProvider) {
+		this.connectionProvider = connectionProvider;
+	}
+
+	/**
+	 * Return the configured {@link ConnectionProvider}.
+	 */
+	public ConnectionProvider getConnectionProvider() {
+		Assert.state(this.connectionProvider != null, "ConnectionProvider not initialized yet");
+		return this.connectionProvider;
+	}
+
+	/**
+	 * Use this when you don't want to participate in global resources and
+	 * you want to customize the creation of the managed {@code LoopResources}.
+	 * <p>By default, {@code LoopResources.create("reactor-http")} is used.
+	 * <p>Note that this option is ignored if {@code userGlobalResources=false} or
+	 * {@link #setLoopResources(LoopResources)} is set.
+	 * @param supplier the supplier to use
+	 */
+	public void setLoopResourcesSupplier(Supplier<LoopResources> supplier) {
+		this.loopResourcesSupplier = supplier;
+	}
+
+	/**
+	 * Use this option when you want to provide an externally managed
+	 * {@link LoopResources} instance.
+	 * @param loopResources the loop resources to use as is
+	 */
+	public void setLoopResources(LoopResources loopResources) {
+		this.loopResources = loopResources;
+	}
+
+	/**
+	 * Return the configured {@link LoopResources}.
+	 */
+	public LoopResources getLoopResources() {
+		Assert.state(this.loopResources != null, "LoopResources not initialized yet");
+		return this.loopResources;
+	}
+
+	/**
+	 * Configure the amount of time we'll wait before shutting down resources.
+	 * If a task is submitted during the {@code shutdownQuietPeriod}, it is guaranteed
+	 * to be accepted and the {@code shutdownQuietPeriod} will start over.
+	 * <p>By default, this is set to
+	 * {@link LoopResources#DEFAULT_SHUTDOWN_QUIET_PERIOD} which is 2 seconds but
+	 * can also be overridden with the system property
+	 * {@link reactor.netty5.ReactorNetty#SHUTDOWN_QUIET_PERIOD
+	 * ReactorNetty.SHUTDOWN_QUIET_PERIOD}.
+	 * @since 5.2.4
+	 * @see #setShutdownTimeout(Duration)
+	 */
+	public void setShutdownQuietPeriod(Duration shutdownQuietPeriod) {
+		Assert.notNull(shutdownQuietPeriod, "shutdownQuietPeriod should not be null");
+		this.shutdownQuietPeriod = shutdownQuietPeriod;
+	}
+
+	/**
+	 * Configure the maximum amount of time to wait until the disposal of the
+	 * underlying resources regardless if a task was submitted during the
+	 * {@code shutdownQuietPeriod}.
+	 * <p>By default, this is set to
+	 * {@link LoopResources#DEFAULT_SHUTDOWN_TIMEOUT} which is 15 seconds but
+	 * can also be overridden with the system property
+	 * {@link reactor.netty5.ReactorNetty#SHUTDOWN_TIMEOUT
+	 * ReactorNetty.SHUTDOWN_TIMEOUT}.
+	 * @since 5.2.4
+	 * @see #setShutdownQuietPeriod(Duration)
+	 */
+	public void setShutdownTimeout(Duration shutdownTimeout) {
+		Assert.notNull(shutdownTimeout, "shutdownTimeout should not be null");
+		this.shutdownTimeout = shutdownTimeout;
+	}
+
+
+	@Override
+	public void afterPropertiesSet() {
+		if (this.useGlobalResources) {
+			Assert.isTrue(this.loopResources == null && this.connectionProvider == null,
+					"'useGlobalResources' is mutually exclusive with explicitly configured resources");
+			HttpResources httpResources = HttpResources.get();
+			if (this.globalResourcesConsumer != null) {
+				this.globalResourcesConsumer.accept(httpResources);
+			}
+			this.connectionProvider = httpResources;
+			this.loopResources = httpResources;
+		}
+		else {
+			if (this.loopResources == null) {
+				this.manageLoopResources = true;
+				this.loopResources = this.loopResourcesSupplier.get();
+			}
+			if (this.connectionProvider == null) {
+				this.manageConnectionProvider = true;
+				this.connectionProvider = this.connectionProviderSupplier.get();
+			}
+		}
+	}
+
+	@Override
+	public void destroy() {
+		if (this.useGlobalResources) {
+			HttpResources.disposeLoopsAndConnectionsLater(this.shutdownQuietPeriod, this.shutdownTimeout).block();
+		}
+		else {
+			try {
+				ConnectionProvider provider = this.connectionProvider;
+				if (provider != null && this.manageConnectionProvider) {
+					provider.disposeLater().block();
+				}
+			}
+			catch (Throwable ex) {
+				// ignore
+			}
+
+			try {
+				LoopResources resources = this.loopResources;
+				if (resources != null && this.manageLoopResources) {
+					resources.disposeLater(this.shutdownQuietPeriod, this.shutdownTimeout).block();
+				}
+			}
+			catch (Throwable ex) {
+				// ignore
+			}
+		}
+	}
+
+}

spring-web/src/main/java/org/springframework/http/server/reactive/Netty5HeadersAdapter.java

@@ -0,0 +1,280 @@
+/*
+ * Copyright 2002-2022 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.server.reactive;
+
+import java.util.AbstractSet;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import io.netty5.handler.codec.http.HttpHeaders;
+
+import org.springframework.lang.Nullable;
+import org.springframework.util.CollectionUtils;
+import org.springframework.util.MultiValueMap;
+
+/**
+ * {@code MultiValueMap} implementation for wrapping Netty HTTP headers.
+ *
+ * <p>This class is based on {@link NettyHeadersAdapter}.
+ *
+ * @author Violeta Georgieva
+ * @since 6.0
+ */
+final class Netty5HeadersAdapter implements MultiValueMap<String, String> {
+
+	private final HttpHeaders headers;
+
+
+	Netty5HeadersAdapter(HttpHeaders headers) {
+		this.headers = headers;
+	}
+
+
+	@Override
+	@Nullable
+	public String getFirst(String key) {
+		return this.headers.get(key);
+	}
+
+	@Override
+	public void add(String key, @Nullable String value) {
+		if (value != null) {
+			this.headers.add(key, value);
+		}
+	}
+
+	@Override
+	public void addAll(String key, List<? extends String> values) {
+		this.headers.add(key, values);
+	}
+
+	@Override
+	public void addAll(MultiValueMap<String, String> values) {
+		values.forEach(this.headers::add);
+	}
+
+	@Override
+	public void set(String key, @Nullable String value) {
+		if (value != null) {
+			this.headers.set(key, value);
+		}
+	}
+
+	@Override
+	public void setAll(Map<String, String> values) {
+		values.forEach(this.headers::set);
+	}
+
+	@Override
+	public Map<String, String> toSingleValueMap() {
+		Map<String, String> singleValueMap = CollectionUtils.newLinkedHashMap(this.headers.size());
+		this.headers.entries()
+				.forEach(entry -> {
+					if (!singleValueMap.containsKey(entry.getKey())) {
+						singleValueMap.put(entry.getKey(), entry.getValue());
+					}
+				});
+		return singleValueMap;
+	}
+
+	@Override
+	public int size() {
+		return this.headers.names().size();
+	}
+
+	@Override
+	public boolean isEmpty() {
+		return this.headers.isEmpty();
+	}
+
+	@Override
+	public boolean containsKey(Object key) {
+		return (key instanceof String headerName && this.headers.contains(headerName));
+	}
+
+	@Override
+	public boolean containsValue(Object value) {
+		return (value instanceof String &&
+				this.headers.entries().stream()
+						.anyMatch(entry -> value.equals(entry.getValue())));
+	}
+
+	@Override
+	@Nullable
+	public List<String> get(Object key) {
+		if (containsKey(key)) {
+			return this.headers.getAll((String) key);
+		}
+		return null;
+	}
+
+	@Nullable
+	@Override
+	public List<String> put(String key, @Nullable List<String> value) {
+		List<String> previousValues = this.headers.getAll(key);
+		this.headers.set(key, value);
+		return previousValues;
+	}
+
+	@Nullable
+	@Override
+	public List<String> remove(Object key) {
+		if (key instanceof String headerName) {
+			List<String> previousValues = this.headers.getAll(headerName);
+			this.headers.remove(headerName);
+			return previousValues;
+		}
+		return null;
+	}
+
+	@Override
+	public void putAll(Map<? extends String, ? extends List<String>> map) {
+		map.forEach(this.headers::set);
+	}
+
+	@Override
+	public void clear() {
+		this.headers.clear();
+	}
+
+	@Override
+	public Set<String> keySet() {
+		return new HeaderNames();
+	}
+
+	@Override
+	public Collection<List<String>> values() {
+		return this.headers.names().stream()
+				.map(this.headers::getAll).collect(Collectors.toList());
+	}
+
+	@Override
+	public Set<Entry<String, List<String>>> entrySet() {
+		return new AbstractSet<>() {
+			@Override
+			public Iterator<Entry<String, List<String>>> iterator() {
+				return new EntryIterator();
+			}
+
+			@Override
+			public int size() {
+				return headers.size();
+			}
+		};
+	}
+
+
+	@Override
+	public String toString() {
+		return org.springframework.http.HttpHeaders.formatHeaders(this);
+	}
+
+
+	private class EntryIterator implements Iterator<Entry<String, List<String>>> {
+
+		private final Iterator<String> names = headers.names().iterator();
+
+		@Override
+		public boolean hasNext() {
+			return this.names.hasNext();
+		}
+
+		@Override
+		public Entry<String, List<String>> next() {
+			return new HeaderEntry(this.names.next());
+		}
+	}
+
+
+	private class HeaderEntry implements Entry<String, List<String>> {
+
+		private final String key;
+
+		HeaderEntry(String key) {
+			this.key = key;
+		}
+
+		@Override
+		public String getKey() {
+			return this.key;
+		}
+
+		@Override
+		public List<String> getValue() {
+			return headers.getAll(this.key);
+		}
+
+		@Override
+		public List<String> setValue(List<String> value) {
+			List<String> previousValues = headers.getAll(this.key);
+			headers.set(this.key, value);
+			return previousValues;
+		}
+	}
+
+	private class HeaderNames extends AbstractSet<String> {
+
+		@Override
+		public Iterator<String> iterator() {
+			return new HeaderNamesIterator(headers.names().iterator());
+		}
+
+		@Override
+		public int size() {
+			return headers.names().size();
+		}
+	}
+
+	private final class HeaderNamesIterator implements Iterator<String> {
+
+		private final Iterator<String> iterator;
+
+		@Nullable
+		private String currentName;
+
+		private HeaderNamesIterator(Iterator<String> iterator) {
+			this.iterator = iterator;
+		}
+
+		@Override
+		public boolean hasNext() {
+			return this.iterator.hasNext();
+		}
+
+		@Override
+		public String next() {
+			this.currentName = this.iterator.next();
+			return this.currentName;
+		}
+
+		@Override
+		public void remove() {
+			if (this.currentName == null) {
+				throw new IllegalStateException("No current Header in iterator");
+			}
+			if (!headers.contains(this.currentName)) {
+				throw new IllegalStateException("Header not present: " + this.currentName);
+			}
+			headers.remove(this.currentName);
+		}
+	}
+
+}

spring-web/src/main/java/org/springframework/http/server/reactive/NettyHeadersAdapter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2021 the original author or authors.
+ * Copyright 2002-2022 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.
@@ -38,7 +38,7 @@ import org.springframework.util.MultiValueMap;
  * @author Brian Clozel
  * @since 5.1.1
  */
-class NettyHeadersAdapter implements MultiValueMap<String, String> {
+final class NettyHeadersAdapter implements MultiValueMap<String, String> {
 
 	private final HttpHeaders headers;
 

spring-web/src/main/java/org/springframework/http/server/reactive/ReactorNetty2HttpHandlerAdapter.java

@@ -0,0 +1,79 @@
+/*
+ * Copyright 2002-2022 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.server.reactive;
+
+import java.net.URISyntaxException;
+import java.util.function.BiFunction;
+
+import io.netty5.handler.codec.http.HttpResponseStatus;
+import org.apache.commons.logging.Log;
+import reactor.core.publisher.Mono;
+import reactor.netty5.http.server.HttpServerRequest;
+import reactor.netty5.http.server.HttpServerResponse;
+
+import org.springframework.core.io.buffer.Netty5DataBufferFactory;
+import org.springframework.http.HttpLogging;
+import org.springframework.http.HttpMethod;
+import org.springframework.util.Assert;
+
+/**
+ * Adapt {@link HttpHandler} to the Reactor Netty 5 channel handling function.
+ *
+ * <p>This class is based on {@link ReactorHttpHandlerAdapter}.
+ *
+ * @author Violeta Georgieva
+ * @since 6.0
+ */
+public class ReactorNetty2HttpHandlerAdapter implements BiFunction<HttpServerRequest, HttpServerResponse, Mono<Void>> {
+
+	private static final Log logger = HttpLogging.forLogName(ReactorNetty2HttpHandlerAdapter.class);
+
+
+	private final HttpHandler httpHandler;
+
+
+	public ReactorNetty2HttpHandlerAdapter(HttpHandler httpHandler) {
+		Assert.notNull(httpHandler, "HttpHandler must not be null");
+		this.httpHandler = httpHandler;
+	}
+
+
+	@Override
+	public Mono<Void> apply(HttpServerRequest reactorRequest, HttpServerResponse reactorResponse) {
+		Netty5DataBufferFactory bufferFactory = new Netty5DataBufferFactory(reactorResponse.alloc());
+		try {
+			ReactorNetty2ServerHttpRequest request = new ReactorNetty2ServerHttpRequest(reactorRequest, bufferFactory);
+			ServerHttpResponse response = new ReactorNetty2ServerHttpResponse(reactorResponse, bufferFactory);
+
+			if (request.getMethod() == HttpMethod.HEAD) {
+				response = new HttpHeadResponseDecorator(response);
+			}
+
+			return this.httpHandler.handle(request, response)
+					.doOnError(ex -> logger.trace(request.getLogPrefix() + "Failed to complete: " + ex.getMessage()))
+					.doOnSuccess(aVoid -> logger.trace(request.getLogPrefix() + "Handling completed"));
+		}
+		catch (URISyntaxException ex) {
+			if (logger.isDebugEnabled()) {
+				logger.debug("Failed to get request URI: " + ex.getMessage());
+			}
+			reactorResponse.status(HttpResponseStatus.BAD_REQUEST);
+			return Mono.empty();
+		}
+	}
+
+}

spring-web/src/main/java/org/springframework/http/server/reactive/ReactorNetty2ServerHttpRequest.java

@@ -0,0 +1,243 @@
+/*
+ * Copyright 2002-2022 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.server.reactive;
+
+import java.net.InetSocketAddress;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.util.concurrent.atomic.AtomicLong;
+
+import javax.net.ssl.SSLSession;
+
+import io.netty5.channel.Channel;
+import io.netty5.handler.codec.http.HttpHeaderNames;
+import io.netty5.handler.codec.http.cookie.Cookie;
+import io.netty5.handler.ssl.SslHandler;
+import org.apache.commons.logging.Log;
+import reactor.core.publisher.Flux;
+import reactor.netty5.Connection;
+import reactor.netty5.http.server.HttpServerRequest;
+
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.Netty5DataBufferFactory;
+import org.springframework.http.HttpCookie;
+import org.springframework.http.HttpLogging;
+import org.springframework.http.HttpMethod;
+import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
+import org.springframework.util.ClassUtils;
+import org.springframework.util.LinkedMultiValueMap;
+import org.springframework.util.MultiValueMap;
+
+/**
+ * Adapt {@link ServerHttpRequest} to the Reactor {@link HttpServerRequest}.
+ *
+ * <p>This class is based on {@link ReactorServerHttpRequest}.
+ *
+ * @author Violeta Georgieva
+ * @since 6.0
+ */
+class ReactorNetty2ServerHttpRequest extends AbstractServerHttpRequest {
+
+	/** Reactor Netty 1.0.5+. */
+	static final boolean reactorNettyRequestChannelOperationsIdPresent = ClassUtils.isPresent(
+			"reactor.netty.ChannelOperationsId", ReactorNetty2ServerHttpRequest.class.getClassLoader());
+
+	private static final Log logger = HttpLogging.forLogName(ReactorNetty2ServerHttpRequest.class);
+
+
+	private static final AtomicLong logPrefixIndex = new AtomicLong();
+
+
+	private final HttpServerRequest request;
+
+	private final Netty5DataBufferFactory bufferFactory;
+
+
+	public ReactorNetty2ServerHttpRequest(HttpServerRequest request, Netty5DataBufferFactory bufferFactory)
+			throws URISyntaxException {
+
+		super(initUri(request), "", new Netty5HeadersAdapter(request.requestHeaders()));
+		Assert.notNull(bufferFactory, "DataBufferFactory must not be null");
+		this.request = request;
+		this.bufferFactory = bufferFactory;
+	}
+
+	private static URI initUri(HttpServerRequest request) throws URISyntaxException {
+		Assert.notNull(request, "HttpServerRequest must not be null");
+		return new URI(resolveBaseUrl(request) + resolveRequestUri(request));
+	}
+
+	private static URI resolveBaseUrl(HttpServerRequest request) throws URISyntaxException {
+		String scheme = getScheme(request);
+		String header = request.requestHeaders().get(HttpHeaderNames.HOST);
+		if (header != null) {
+			final int portIndex;
+			if (header.startsWith("[")) {
+				portIndex = header.indexOf(':', header.indexOf(']'));
+			}
+			else {
+				portIndex = header.indexOf(':');
+			}
+			if (portIndex != -1) {
+				try {
+					return new URI(scheme, null, header.substring(0, portIndex),
+							Integer.parseInt(header, portIndex + 1, header.length(), 10), null, null, null);
+				}
+				catch (NumberFormatException ex) {
+					throw new URISyntaxException(header, "Unable to parse port", portIndex);
+				}
+			}
+			else {
+				return new URI(scheme, header, null, null);
+			}
+		}
+		else {
+			InetSocketAddress localAddress = request.hostAddress();
+			Assert.state(localAddress != null, "No host address available");
+			return new URI(scheme, null, localAddress.getHostString(),
+					localAddress.getPort(), null, null, null);
+		}
+	}
+
+	private static String getScheme(HttpServerRequest request) {
+		return request.scheme();
+	}
+
+	private static String resolveRequestUri(HttpServerRequest request) {
+		String uri = request.uri();
+		for (int i = 0; i < uri.length(); i++) {
+			char c = uri.charAt(i);
+			if (c == '/' || c == '?' || c == '#') {
+				break;
+			}
+			if (c == ':' && (i + 2 < uri.length())) {
+				if (uri.charAt(i + 1) == '/' && uri.charAt(i + 2) == '/') {
+					for (int j = i + 3; j < uri.length(); j++) {
+						c = uri.charAt(j);
+						if (c == '/' || c == '?' || c == '#') {
+							return uri.substring(j);
+						}
+					}
+					return "";
+				}
+			}
+		}
+		return uri;
+	}
+
+	@Override
+	public HttpMethod getMethod() {
+		return HttpMethod.valueOf(this.request.method().name());
+	}
+
+	@Override
+	@Deprecated
+	public String getMethodValue() {
+		return this.request.method().name();
+	}
+
+	@Override
+	protected MultiValueMap<String, HttpCookie> initCookies() {
+		MultiValueMap<String, HttpCookie> cookies = new LinkedMultiValueMap<>();
+		for (CharSequence name : this.request.cookies().keySet()) {
+			for (Cookie cookie : this.request.cookies().get(name)) {
+				HttpCookie httpCookie = new HttpCookie(name.toString(), cookie.value());
+				cookies.add(name.toString(), httpCookie);
+			}
+		}
+		return cookies;
+	}
+
+	@Override
+	@Nullable
+	public InetSocketAddress getLocalAddress() {
+		return this.request.hostAddress();
+	}
+
+	@Override
+	@Nullable
+	public InetSocketAddress getRemoteAddress() {
+		return this.request.remoteAddress();
+	}
+
+	@Override
+	@Nullable
+	protected SslInfo initSslInfo() {
+		Channel channel = ((Connection) this.request).channel();
+		SslHandler sslHandler = channel.pipeline().get(SslHandler.class);
+		if (sslHandler == null && channel.parent() != null) { // HTTP/2
+			sslHandler = channel.parent().pipeline().get(SslHandler.class);
+		}
+		if (sslHandler != null) {
+			SSLSession session = sslHandler.engine().getSession();
+			return new DefaultSslInfo(session);
+		}
+		return null;
+	}
+
+	@Override
+	public Flux<DataBuffer> getBody() {
+		return this.request.receive().transferOwnership().map(this.bufferFactory::wrap);
+	}
+
+	@SuppressWarnings("unchecked")
+	@Override
+	public <T> T getNativeRequest() {
+		return (T) this.request;
+	}
+
+	@Override
+	@Nullable
+	protected String initId() {
+		if (this.request instanceof Connection) {
+			return ((Connection) this.request).channel().id().asShortText() +
+					"-" + logPrefixIndex.incrementAndGet();
+		}
+		return null;
+	}
+
+	@Override
+	protected String initLogPrefix() {
+		if (reactorNettyRequestChannelOperationsIdPresent) {
+			String id = (ChannelOperationsIdHelper.getId(this.request));
+			if (id != null) {
+				return id;
+			}
+		}
+		if (this.request instanceof Connection) {
+			return ((Connection) this.request).channel().id().asShortText() +
+					"-" + logPrefixIndex.incrementAndGet();
+		}
+		return getId();
+	}
+
+
+	private static class ChannelOperationsIdHelper {
+
+		@Nullable
+		public static String getId(HttpServerRequest request) {
+			if (request instanceof reactor.netty.ChannelOperationsId) {
+				return (logger.isDebugEnabled() ?
+						((reactor.netty.ChannelOperationsId) request).asLongText() :
+						((reactor.netty.ChannelOperationsId) request).asShortText());
+			}
+			return null;
+		}
+	}
+
+}

spring-web/src/main/java/org/springframework/http/server/reactive/ReactorNetty2ServerHttpResponse.java

@@ -0,0 +1,157 @@
+/*
+ * Copyright 2002-2022 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.server.reactive;
+
+import java.nio.file.Path;
+import java.util.List;
+
+import io.netty5.buffer.api.Buffer;
+import io.netty5.channel.ChannelId;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.netty5.ChannelOperationsId;
+import reactor.netty5.http.server.HttpServerResponse;
+
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.DataBufferFactory;
+import org.springframework.core.io.buffer.DataBufferUtils;
+import org.springframework.core.io.buffer.Netty5DataBufferFactory;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.HttpStatusCode;
+import org.springframework.http.ResponseCookie;
+import org.springframework.http.ZeroCopyHttpOutputMessage;
+import org.springframework.util.Assert;
+
+/**
+ * Adapt {@link ServerHttpResponse} to the {@link HttpServerResponse}.
+ *
+ * <p>This class is based on {@link ReactorServerHttpResponse}.
+ *
+ * @author Violeta Georgieva
+ * @since 6.0
+ */
+class ReactorNetty2ServerHttpResponse extends AbstractServerHttpResponse implements ZeroCopyHttpOutputMessage {
+
+	private static final Log logger = LogFactory.getLog(ReactorNetty2ServerHttpResponse.class);
+
+
+	private final HttpServerResponse response;
+
+
+	public ReactorNetty2ServerHttpResponse(HttpServerResponse response, DataBufferFactory bufferFactory) {
+		super(bufferFactory, new HttpHeaders(new Netty5HeadersAdapter(response.responseHeaders())));
+		Assert.notNull(response, "HttpServerResponse must not be null");
+		this.response = response;
+	}
+
+
+	@SuppressWarnings("unchecked")
+	@Override
+	public <T> T getNativeResponse() {
+		return (T) this.response;
+	}
+
+	@Override
+	public HttpStatusCode getStatusCode() {
+		HttpStatusCode status = super.getStatusCode();
+		return (status != null ? status : HttpStatusCode.valueOf(this.response.status().code()));
+	}
+
+	@Override
+	@Deprecated
+	public Integer getRawStatusCode() {
+		Integer status = super.getRawStatusCode();
+		return (status != null ? status : this.response.status().code());
+	}
+
+	@Override
+	protected void applyStatusCode() {
+		HttpStatusCode status = super.getStatusCode();
+		if (status != null) {
+			this.response.status(status.value());
+		}
+	}
+
+	@Override
+	protected Mono<Void> writeWithInternal(Publisher<? extends DataBuffer> publisher) {
+		return this.response.send(toByteBufs(publisher)).then();
+	}
+
+	@Override
+	protected Mono<Void> writeAndFlushWithInternal(Publisher<? extends Publisher<? extends DataBuffer>> publisher) {
+		return this.response.sendGroups(Flux.from(publisher).map(this::toByteBufs)).then();
+	}
+
+	@Override
+	protected void applyHeaders() {
+	}
+
+	@Override
+	protected void applyCookies() {
+		// Netty Cookie doesn't support sameSite. When this is resolved, we can adapt to it again:
+		// https://github.com/netty/netty/issues/8161
+		for (List<ResponseCookie> cookies : getCookies().values()) {
+			for (ResponseCookie cookie : cookies) {
+				this.response.addHeader(HttpHeaders.SET_COOKIE, cookie.toString());
+			}
+		}
+	}
+
+	@Override
+	public Mono<Void> writeWith(Path file, long position, long count) {
+		return doCommit(() -> this.response.sendFile(file, position, count).then());
+	}
+
+	private Publisher<Buffer> toByteBufs(Publisher<? extends DataBuffer> dataBuffers) {
+		return dataBuffers instanceof Mono ?
+				Mono.from(dataBuffers).map(Netty5DataBufferFactory::toBuffer) :
+				Flux.from(dataBuffers).map(Netty5DataBufferFactory::toBuffer);
+	}
+
+	@Override
+	protected void touchDataBuffer(DataBuffer buffer) {
+		if (logger.isDebugEnabled()) {
+			if (ReactorNetty2ServerHttpRequest.reactorNettyRequestChannelOperationsIdPresent) {
+				if (ChannelOperationsIdHelper.touch(buffer, this.response)) {
+					return;
+				}
+			}
+			this.response.withConnection(connection -> {
+				ChannelId id = connection.channel().id();
+				DataBufferUtils.touch(buffer, "Channel id: " + id.asShortText());
+			});
+		}
+	}
+
+
+	private static class ChannelOperationsIdHelper {
+
+		public static boolean touch(DataBuffer dataBuffer, HttpServerResponse response) {
+			if (response instanceof ChannelOperationsId) {
+				String id = ((ChannelOperationsId) response).asLongText();
+				DataBufferUtils.touch(dataBuffer, "Channel id: " + id);
+				return true;
+			}
+			return false;
+		}
+	}
+
+
+}