+ * WARNING: Using this feature can lead to serious performance issues. This should be only
+ * used for debugging purposes.
+ *
+ * @since 3.1.9
+ */
+public class TracingChannelInboundHandler extends ChannelInboundHandlerAdapter {
+
+ static final AttributeKey SPAN_ATTRIBUTE_KEY = AttributeKey.valueOf(Span.class.getName());
+
+ final CurrentTraceContext currentTraceContext;
+
+ /**
+ * Creates a new instance of {@link TracingChannelInboundHandler}.
+ * @param currentTraceContext current trace context
+ */
+ public TracingChannelInboundHandler(CurrentTraceContext currentTraceContext) {
+ this.currentTraceContext = currentTraceContext;
+ }
+
+ @Override
+ public void channelRegistered(ChannelHandlerContext ctx) {
+ if (instrumentOperation(ctx, () -> ctx.fireChannelRegistered())) {
+ return;
+ }
+
+ ctx.fireChannelRegistered();
+ }
+
+ @Override
+ public void channelUnregistered(ChannelHandlerContext ctx) {
+ if (instrumentOperation(ctx, () -> ctx.fireChannelUnregistered())) {
+ return;
+ }
+
+ ctx.fireChannelUnregistered();
+ }
+
+ @Override
+ public void channelActive(ChannelHandlerContext ctx) {
+ if (instrumentOperation(ctx, () -> ctx.fireChannelActive())) {
+ return;
+ }
+
+ ctx.fireChannelActive();
+ }
+
+ @Override
+ public void channelInactive(ChannelHandlerContext ctx) {
+ if (instrumentOperation(ctx, () -> ctx.fireChannelInactive())) {
+ return;
+ }
+
+ ctx.fireChannelInactive();
+ }
+
+ @Override
+ public void channelRead(ChannelHandlerContext ctx, Object msg) {
+ if (instrumentOperation(ctx, () -> ctx.fireChannelRead(msg))) {
+ return;
+ }
+
+ ctx.fireChannelRead(msg);
+ }
+
+ @Override
+ public void channelReadComplete(ChannelHandlerContext ctx) {
+ if (instrumentOperation(ctx, () -> ctx.fireChannelReadComplete())) {
+ return;
+ }
+
+ ctx.fireChannelReadComplete();
+ }
+
+ @Override
+ public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
+ if (instrumentOperation(ctx, () -> ctx.fireUserEventTriggered(evt))) {
+ return;
+ }
+
+ ctx.fireUserEventTriggered(evt);
+ }
+
+ @Override
+ public void channelWritabilityChanged(ChannelHandlerContext ctx) {
+ if (instrumentOperation(ctx, () -> ctx.fireChannelWritabilityChanged())) {
+ return;
+ }
+
+ ctx.fireChannelWritabilityChanged();
+ }
+
+ @Override
+ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
+ if (instrumentOperation(ctx, () -> ctx.fireExceptionCaught(cause))) {
+ return;
+ }
+
+ ctx.fireExceptionCaught(cause);
+ }
+
+ @Override
+ public boolean isSharable() {
+ return false;
+ }
+
+ @Override
+ public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
+ if (instrumentOperation(ctx, () -> {
+ try {
+ super.handlerAdded(ctx);
+ }
+ catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ })) {
+ return;
+ }
+
+ super.handlerAdded(ctx);
+ }
+
+ @Override
+ public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
+ if (instrumentOperation(ctx, () -> {
+ try {
+ super.handlerRemoved(ctx);
+ }
+ catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ })) {
+ return;
+ }
+
+ super.handlerRemoved(ctx);
+ }
+
+ boolean instrumentOperation(ChannelHandlerContext ctx, Runnable operation) {
+ Span span = ctx.channel().attr(SPAN_ATTRIBUTE_KEY).get();
+ if (span != null) {
+ try (CurrentTraceContext.Scope scope = currentTraceContext.maybeScope(span.context())) {
+ operation.run();
+ }
+ return true;
+ }
+ else {
+ Connection conn = Connection.from(ctx.channel());
+ if (conn instanceof ConnectionObserver) {
+ TraceContext parent = ((ConnectionObserver) conn).currentContext().getOrDefault(TraceContext.class,
+ null);
+ if (parent != null) {
+ try (CurrentTraceContext.Scope scope = currentTraceContext.maybeScope(parent)) {
+ operation.run();
+ }
+ return true;
+ }
+ }
+ else {
+ ChannelOperations ops = conn.as(ChannelOperations.class);
+ if (ops instanceof HttpClientResponse) {
+ TraceContext parent = TracingHandlerUtil
+ .traceContext(((HttpClientResponse) ops).currentContextView());
+ if (parent != null) {
+ try (CurrentTraceContext.Scope scope = currentTraceContext.maybeScope(parent)) {
+ operation.run();
+ }
+ return true;
+ }
+ }
+ }
+ }
+ return false;
+ }
+
+}
diff --git a/spring-cloud-sleuth-brave/src/main/java/org/springframework/cloud/sleuth/brave/instrument/reactor/netty/TracingChannelOutboundHandler.java b/spring-cloud-sleuth-brave/src/main/java/org/springframework/cloud/sleuth/brave/instrument/reactor/netty/TracingChannelOutboundHandler.java
new file mode 100644
index 000000000..4f9db8030
--- /dev/null
+++ b/spring-cloud-sleuth-brave/src/main/java/org/springframework/cloud/sleuth/brave/instrument/reactor/netty/TracingChannelOutboundHandler.java
@@ -0,0 +1,202 @@
+/*
+ * Copyright 2013-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.cloud.sleuth.brave.instrument.reactor.netty;
+
+import java.net.SocketAddress;
+
+import brave.Span;
+import brave.propagation.CurrentTraceContext;
+import brave.propagation.TraceContext;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelOutboundHandlerAdapter;
+import io.netty.channel.ChannelPromise;
+import io.netty.util.AttributeKey;
+import reactor.netty.Connection;
+import reactor.netty.ConnectionObserver;
+import reactor.netty.channel.ChannelOperations;
+import reactor.netty.http.client.HttpClientRequest;
+
+/**
+ * {@link ChannelOutboundHandlerAdapter} that wraps all events in scope.
+ *
+ * WARNING: Using this feature can lead to serious performance issues. This should be only
+ * used for debugging purposes.
+ *
+ * @since 3.1.9
+ */
+public class TracingChannelOutboundHandler extends ChannelOutboundHandlerAdapter {
+
+ static final AttributeKey SPAN_ATTRIBUTE_KEY = AttributeKey.valueOf(Span.class.getName());
+
+ final CurrentTraceContext currentTraceContext;
+
+ /**
+ * Creates a new instance of {@link TracingChannelOutboundHandler}.
+ * @param currentTraceContext current trace context
+ */
+ public TracingChannelOutboundHandler(CurrentTraceContext currentTraceContext) {
+ this.currentTraceContext = currentTraceContext;
+ }
+
+ @Override
+ public void bind(ChannelHandlerContext ctx, SocketAddress localAddress, ChannelPromise promise) {
+ if (instrumentOperation(ctx, () -> ctx.bind(localAddress, promise))) {
+ return;
+ }
+
+ ctx.bind(localAddress, promise);
+ }
+
+ @Override
+ public void connect(ChannelHandlerContext ctx, SocketAddress remoteAddress, SocketAddress localAddress,
+ ChannelPromise promise) {
+ if (instrumentOperation(ctx, () -> ctx.connect(remoteAddress, localAddress, promise))) {
+ return;
+ }
+
+ ctx.connect(remoteAddress, localAddress, promise);
+ }
+
+ @Override
+ public void disconnect(ChannelHandlerContext ctx, ChannelPromise promise) {
+ if (instrumentOperation(ctx, () -> ctx.disconnect(promise))) {
+ return;
+ }
+
+ ctx.disconnect(promise);
+ }
+
+ @Override
+ public void close(ChannelHandlerContext ctx, ChannelPromise promise) {
+ if (instrumentOperation(ctx, () -> ctx.close(promise))) {
+ return;
+ }
+
+ ctx.close(promise);
+ }
+
+ @Override
+ public void deregister(ChannelHandlerContext ctx, ChannelPromise promise) {
+ if (instrumentOperation(ctx, () -> ctx.deregister(promise))) {
+ return;
+ }
+
+ ctx.deregister(promise);
+ }
+
+ @Override
+ public void read(ChannelHandlerContext ctx) {
+ if (instrumentOperation(ctx, () -> ctx.read())) {
+ return;
+ }
+
+ ctx.read();
+ }
+
+ @Override
+ public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
+ if (instrumentOperation(ctx, () -> ctx.write(msg, promise))) {
+ return;
+ }
+
+ ctx.write(msg, promise);
+ }
+
+ @Override
+ public void flush(ChannelHandlerContext ctx) {
+ if (instrumentOperation(ctx, () -> ctx.flush())) {
+ return;
+ }
+
+ ctx.flush();
+ }
+
+ @Override
+ public boolean isSharable() {
+ return false;
+ }
+
+ @Override
+ public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
+ if (instrumentOperation(ctx, () -> {
+ try {
+ super.handlerAdded(ctx);
+ }
+ catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ })) {
+ return;
+ }
+
+ super.handlerAdded(ctx);
+ }
+
+ @Override
+ public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
+ if (instrumentOperation(ctx, () -> {
+ try {
+ super.handlerRemoved(ctx);
+ }
+ catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ })) {
+ return;
+ }
+
+ super.handlerRemoved(ctx);
+ }
+
+ boolean instrumentOperation(ChannelHandlerContext ctx, Runnable operation) {
+ Span span = ctx.channel().attr(SPAN_ATTRIBUTE_KEY).get();
+ if (span != null) {
+ try (CurrentTraceContext.Scope scope = currentTraceContext.maybeScope(span.context())) {
+ operation.run();
+ }
+ return true;
+ }
+ else {
+ Connection conn = Connection.from(ctx.channel());
+ if (conn instanceof ConnectionObserver) {
+ TraceContext parent = ((ConnectionObserver) conn).currentContext().getOrDefault(TraceContext.class,
+ null);
+ if (parent != null) {
+ try (CurrentTraceContext.Scope scope = currentTraceContext.maybeScope(parent)) {
+ operation.run();
+ }
+ return true;
+ }
+ }
+ else {
+ ChannelOperations ops = conn.as(ChannelOperations.class);
+ if (ops instanceof HttpClientRequest) {
+ TraceContext traceContext = TracingHandlerUtil
+ .traceContext(((HttpClientRequest) ops).currentContextView());
+ if (traceContext != null) {
+ try (CurrentTraceContext.Scope scope = currentTraceContext.maybeScope(traceContext)) {
+ operation.run();
+ }
+ return true;
+ }
+ }
+ }
+ }
+ return false;
+ }
+
+}
diff --git a/spring-cloud-sleuth-brave/src/main/java/org/springframework/cloud/sleuth/brave/instrument/reactor/netty/TracingHandlerUtil.java b/spring-cloud-sleuth-brave/src/main/java/org/springframework/cloud/sleuth/brave/instrument/reactor/netty/TracingHandlerUtil.java
new file mode 100644
index 000000000..139e72b17
--- /dev/null
+++ b/spring-cloud-sleuth-brave/src/main/java/org/springframework/cloud/sleuth/brave/instrument/reactor/netty/TracingHandlerUtil.java
@@ -0,0 +1,45 @@
+/*
+ * Copyright 2013-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.cloud.sleuth.brave.instrument.reactor.netty;
+
+import java.util.concurrent.atomic.AtomicReference;
+
+import brave.propagation.TraceContext;
+import reactor.util.context.ContextView;
+
+import org.springframework.cloud.sleuth.Span;
+import org.springframework.cloud.sleuth.brave.bridge.BraveTraceContext;
+import org.springframework.cloud.sleuth.instrument.reactor.ReactorSleuth;
+
+final class TracingHandlerUtil {
+
+ private TracingHandlerUtil() {
+ throw new IllegalStateException("Can't instantiate a utility class");
+ }
+
+ static TraceContext traceContext(ContextView ctxView) {
+ AtomicReference pendingSpan = ReactorSleuth.getPendingSpan(ctxView);
+ if (pendingSpan != null) {
+ Span span = pendingSpan.get();
+ if (span != null) {
+ return BraveTraceContext.toBrave(span.context());
+ }
+ }
+ return null;
+ }
+
+}
diff --git a/tests/brave/spring-cloud-sleuth-instrumentation-gateway-tests/pom.xml b/tests/brave/spring-cloud-sleuth-instrumentation-gateway-tests/pom.xml
index 22a70ba37..8194f10d9 100644
--- a/tests/brave/spring-cloud-sleuth-instrumentation-gateway-tests/pom.xml
+++ b/tests/brave/spring-cloud-sleuth-instrumentation-gateway-tests/pom.xml
@@ -76,6 +76,26 @@