From ac577e9ef7e7368643f1f654a9b90af95d612c85 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Tue, 21 Feb 2023 12:02:15 -0500 Subject: [PATCH] Optional io.micrometer:context-propagation (#8556) For better performance by default it is better to not pull a `io.micrometer:context-propagation` a hard dependency. * Remove `io.micrometer:context-propagation` dependency management * It is pulled transitively by the `io.micrometer:micrometer-tracing-integration-test` in test scope * Rework all the `ContextSnapshot` usage in the reactive code to respective recommended `handle()` API in `Flux` and `Mono` --- build.gradle | 2 -- .../channel/FluxMessageChannel.java | 26 ++++++++--------- .../gateway/MessagingGatewaySupport.java | 17 +++++------ .../inbound/WebFluxInboundEndpoint.java | 29 +++++++++---------- 4 files changed, 34 insertions(+), 40 deletions(-) diff --git a/build.gradle b/build.gradle index 0c3603178c..b001979b72 100644 --- a/build.gradle +++ b/build.gradle @@ -87,7 +87,6 @@ ext { lettuceVersion = '6.2.2.RELEASE' log4jVersion = '2.19.0' mailVersion = '1.0.0' - micrometerPropagationVersion = '1.0.2' micrometerTracingVersion = '1.0.2' micrometerVersion = '1.10.4' mockitoVersion = '4.10.0' @@ -533,7 +532,6 @@ project('spring-integration-core') { } api 'io.projectreactor:reactor-core' api 'io.micrometer:micrometer-observation' - api "io.micrometer:context-propagation:$micrometerPropagationVersion" optionalApi 'com.fasterxml.jackson.core:jackson-databind' optionalApi 'com.fasterxml.jackson.datatype:jackson-datatype-jdk8' diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java index 4f1fa5e146..38c543bafd 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java @@ -20,7 +20,6 @@ import java.time.Duration; import java.util.concurrent.TimeUnit; import java.util.concurrent.locks.LockSupport; -import io.micrometer.context.ContextSnapshot; import org.reactivestreams.Publisher; import org.reactivestreams.Subscriber; import reactor.core.Disposable; @@ -112,19 +111,18 @@ public class FluxMessageChannel extends AbstractMessageChannel Flux.from(publisher) .delaySubscription(this.subscribedSignal.asFlux().filter(Boolean::booleanValue).next()) .publishOn(this.scheduler) - .transformDeferredContextual((flux, contextView) -> - flux.doOnNext((message) -> { - var scope = ContextSnapshot.setAllThreadLocalsFrom(contextView); - try (scope) { - if (!send(message)) { - throw new MessageDeliveryException(message, - "Failed to send message to channel '" + this); - } - } - catch (Exception ex) { - logger.warn(ex, () -> "Error during processing event: " + message); - } - })) + .handle((message, synchronousSink) -> { + try { + if (!send(message)) { + logger.warn(new MessageDeliveryException(message, + "Failed to send message to channel '" + this), + "Message was not delivered"); + } + } + catch (Exception ex) { + logger.warn(ex, () -> "Error during processing event: " + message); + } + }) .contextCapture() .subscribe()); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java b/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java index 419334f5c9..112ff0ae92 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java @@ -20,7 +20,6 @@ import java.util.Map; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; -import io.micrometer.context.ContextSnapshot; import io.micrometer.observation.ObservationRegistry; import org.reactivestreams.Publisher; import org.reactivestreams.Subscriber; @@ -721,7 +720,7 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint throw new MessageMappingException("Cannot map to message: " + object, e); } - return Mono.deferContextual(contextView -> { + return Mono.defer(() -> { Object originalReplyChannelHeader = requestMessage.getHeaders().getReplyChannel(); Object originalErrorChannelHeader = requestMessage.getHeaders().getErrorChannel(); @@ -739,13 +738,13 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint .setErrorChannel(replyChan) .build(); - var scope = ContextSnapshot.setAllThreadLocalsFrom(contextView); - try (scope) { - sendMessageForReactiveFlow(requestChannel, messageToSend); - } - - return buildReplyMono(requestMessage, replyChan.replyMono.asMono(), error, - originalReplyChannelHeader, originalErrorChannelHeader); + return Mono.just(messageToSend) + .handle((message, synchronousSink) -> { + sendMessageForReactiveFlow(requestChannel, message); + synchronousSink.complete(); + }) + .then(buildReplyMono(requestMessage, replyChan.replyMono.asMono(), error, + originalReplyChannelHeader, originalErrorChannelHeader)); }) .onErrorResume(t -> error ? Mono.error(t) : handleSendError(requestMessage, t)); } diff --git a/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/inbound/WebFluxInboundEndpoint.java b/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/inbound/WebFluxInboundEndpoint.java index f11368ca22..3be5073289 100644 --- a/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/inbound/WebFluxInboundEndpoint.java +++ b/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/inbound/WebFluxInboundEndpoint.java @@ -27,7 +27,6 @@ import java.util.Set; import java.util.function.Supplier; import java.util.stream.Collectors; -import io.micrometer.context.ContextSnapshot; import org.reactivestreams.Publisher; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; @@ -152,20 +151,20 @@ public class WebFluxInboundEndpoint extends BaseHttpInboundEndpoint implements W new RequestEntity<>(body, exchange.getRequest().getHeaders(), exchange.getRequest().getMethod(), exchange.getRequest().getURI())) .flatMap(entity -> buildMessage(entity, exchange)) - .flatMap(requestTuple -> - Mono.deferContextual(contextView -> { - if (isExpectReply()) { - return sendAndReceiveMessageReactive(requestTuple.getT1()) - .flatMap(replyMessage -> populateResponse(exchange, replyMessage)); - } - else { - var scope = ContextSnapshot.setAllThreadLocalsFrom(contextView); - try (scope) { - send(requestTuple.getT1()); - } - return setStatusCode(exchange, requestTuple.getT2()); - } - })) + .flatMap(requestTuple -> { + if (isExpectReply()) { + return sendAndReceiveMessageReactive(requestTuple.getT1()) + .flatMap(replyMessage -> populateResponse(exchange, replyMessage)); + } + else { + return Mono.just(requestTuple.getT1()) + .handle((objectMessage, synchronousSink) -> { + send(objectMessage); + synchronousSink.complete(); + }) + .then(setStatusCode(exchange, requestTuple.getT2())); + } + }) .doOnTerminate(this.activeCount::decrementAndGet); }