Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Optional io.micrometer:context-propagation (#8556)

Browse Source 
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`
This commit is contained in:
Artem Bilan
2023-02-21 12:02:15 -05:00
committed by GitHub
parent b482b002cb
commit ac577e9ef7
4 changed files with 34 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
build.gradle
View File

@@ -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'

26
spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java
View File

@@ -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());
}

17
spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java
View File

@@ -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));
}

29
spring-integration-webflux/src/main/java/org/springframework/integration/webflux/inbound/WebFluxInboundEndpoint.java
View File

@@ -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);
}