Merge branch '6.1.x'

Author: bclozel

spring-webflux/spring-webflux.gradle

@@ -40,6 +40,7 @@ dependencies {
 	testImplementation(testFixtures(project(":spring-web")))
 	testImplementation("com.fasterxml:aalto-xml")
 	testImplementation("com.squareup.okhttp3:mockwebserver")
+	testImplementation("io.micrometer:context-propagation")
 	testImplementation("io.micrometer:micrometer-observation-test")
 	testImplementation("io.projectreactor:reactor-test")
 	testImplementation("io.reactivex.rxjava3:rxjava")

spring-webflux/src/main/java/org/springframework/web/reactive/function/client/DefaultWebClient.java

@@ -463,8 +463,9 @@ public Mono<ClientResponse> exchange() {
 				ClientRequest request = requestBuilder.build();
 				observationContext.setUriTemplate((String) request.attribute(URI_TEMPLATE_ATTRIBUTE).orElse(null));
 				observationContext.setRequest(request);
-				Mono<ClientResponse> responseMono = filterFunction.apply(exchangeFunction)
-						.exchange(request)
+				final ExchangeFilterFunction finalFilterFunction = filterFunction;
+				Mono<ClientResponse> responseMono = Mono.defer(
+								() -> finalFilterFunction.apply(exchangeFunction).exchange(request))
 						.checkpoint("Request to " +
 								WebClientUtils.getRequestDescription(request.method(), request.url()) +
 								" [DefaultWebClient]")

spring-webflux/src/test/java/org/springframework/web/reactive/function/client/WebClientObservationTests.java

@@ -27,10 +27,12 @@
 import io.micrometer.observation.contextpropagation.ObservationThreadLocalAccessor;
 import io.micrometer.observation.tck.TestObservationRegistry;
 import io.micrometer.observation.tck.TestObservationRegistryAssert;
+import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.mockito.ArgumentCaptor;
 import reactor.core.publisher.Flux;
+import reactor.core.publisher.Hooks;
 import reactor.core.publisher.Mono;
 import reactor.test.StepVerifier;
 
@@ -63,6 +65,7 @@ class WebClientObservationTests {
 
 	@BeforeEach
 	void setup() {
+		Hooks.enableAutomaticContextPropagation();
 		ClientResponse mockResponse = mock();
 		when(mockResponse.statusCode()).thenReturn(HttpStatus.OK);
 		when(mockResponse.headers()).thenReturn(new MockClientHeaders());
@@ -74,6 +77,11 @@ void setup() {
 		this.observationRegistry.observationConfig().observationHandler(new HeaderInjectingHandler());
 	}
 
+	@AfterEach
+	void cleanUp() {
+		Hooks.disableAutomaticContextPropagation();
+	}
+
 	@Test
 	void recordsObservationForSuccessfulExchange() {
 		this.builder.build().get().uri("/resource/{id}", 42)
@@ -148,6 +156,19 @@ void setsCurrentObservationInReactorContext() {
 		verifyAndGetRequest();
 	}
 
+	@Test
+	void setsCurrentObservationInScope() {
+		ExchangeFilterFunction assertionFilter = (request, chain) -> {
+			Observation currentObservation = observationRegistry.getCurrentObservation();
+			assertThat(currentObservation).isNotNull();
+			assertThat(currentObservation.getContext()).isInstanceOf(ClientRequestObservationContext.class);
+			return chain.exchange(request);
+		};
+		this.builder.filter(assertionFilter).build().get().uri("/resource/{id}", 42)
+				.retrieve().bodyToMono(Void.class).block(Duration.ofSeconds(5));
+		verifyAndGetRequest();
+	}
+
 	@Test
 	void recordsObservationWithResponseDetailsWhenFilterFunctionErrors() {
 		ExchangeFilterFunction errorFunction = (req, next) -> next.exchange(req).then(Mono.error(new IllegalStateException()));