Make use of Reactor Netty's ChannelOperationsId

Closes gh-26649

Author: rstoyanchev

build.gradle

@@ -28,7 +28,7 @@ configure(allprojects) { project ->
 		imports {
 			mavenBom "com.fasterxml.jackson:jackson-bom:2.12.2"
 			mavenBom "io.netty:netty-bom:4.1.59.Final"
-			mavenBom "io.projectreactor:reactor-bom:2020.0.4"
+			mavenBom "io.projectreactor:reactor-bom:2020.0.5-SNAPSHOT"
 			mavenBom "io.r2dbc:r2dbc-bom:Arabba-SR9"
 			mavenBom "io.rsocket:rsocket-bom:1.1.0"
 			mavenBom "org.eclipse.jetty:jetty-bom:9.4.38.v20210224"
@@ -292,6 +292,7 @@ configure(allprojects) { project ->
 		repositories {
 			mavenCentral()
 			maven { url "https://repo.spring.io/libs-spring-framework-build" }
+			maven { url "https://repo.spring.io/snapshot" } // Reactor
 		}
 	}
 	configurations.all {

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2020 the original author or authors.
+ * Copyright 2002-2021 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.
@@ -27,15 +27,18 @@
 import io.netty.handler.codec.http.HttpHeaderNames;
 import io.netty.handler.codec.http.cookie.Cookie;
 import io.netty.handler.ssl.SslHandler;
+import org.apache.commons.logging.Log;
 import reactor.core.publisher.Flux;
 import reactor.netty.Connection;
 import reactor.netty.http.server.HttpServerRequest;
 
 import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.core.io.buffer.NettyDataBufferFactory;
 import org.springframework.http.HttpCookie;
+import org.springframework.http.HttpLogging;
 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;
 
@@ -48,6 +51,13 @@
  */
 class ReactorServerHttpRequest extends AbstractServerHttpRequest {
 
+	/** Reactor Netty 1.0.5+. */
+	static final boolean reactorNettyRequestChannelOperationsIdPresent = ClassUtils.isPresent(
+			"reactor.netty.ChannelOperationsId", ReactorServerHttpRequest.class.getClassLoader());
+
+	private static final Log logger = HttpLogging.forLogName(ReactorServerHttpRequest.class);
+
+
 	private static final AtomicLong logPrefixIndex = new AtomicLong();
 
 
@@ -187,11 +197,28 @@ public <T> T getNativeRequest() {
 	@Override
 	@Nullable
 	protected String initId() {
+		if (reactorNettyRequestChannelOperationsIdPresent) {
+			return (ChannelOperationsIdHelper.getId(this.request));
+		}
 		if (this.request instanceof Connection) {
 			return ((Connection) this.request).channel().id().asShortText() +
 					"-" + logPrefixIndex.incrementAndGet();
 		}
 		return null;
 	}
 
+
+	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/ReactorServerHttpResponse.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2020 the original author or authors.
+ * Copyright 2002-2021 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.
@@ -26,6 +26,7 @@
 import org.reactivestreams.Publisher;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.netty.ChannelOperationsId;
 import reactor.netty.http.server.HttpServerResponse;
 
 import org.springframework.core.io.buffer.DataBuffer;
@@ -125,11 +126,30 @@ private Publisher<ByteBuf> toByteBufs(Publisher<? extends DataBuffer> dataBuffer
 	@Override
 	protected void touchDataBuffer(DataBuffer buffer) {
 		if (logger.isDebugEnabled()) {
+			if (ReactorServerHttpRequest.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 reactor.netty.ChannelOperationsId) {
+				String id = ((ChannelOperationsId) response).asLongText();
+				DataBufferUtils.touch(dataBuffer, "Channel id: " + id);
+				return true;
+			}
+			return false;
+		}
+	}
+
+
 }