From 3dfd6838c06833965424be3631b8d69b1dafe14c Mon Sep 17 00:00:00 2001 From: Sigurd Gerke Date: Fri, 17 Apr 2026 12:18:01 +0200 Subject: [PATCH 1/3] Fix a regression on value class parameter handling This commit fixes a regression introduced by gh-36449 for nullable value class with an non-null value. Closes gh-36665 Signed-off-by: Sigurd Gerke --- .../springframework/core/CoroutinesUtils.java | 10 ++++++---- .../core/CoroutinesUtilsTests.kt | 7 +++++++ .../method/support/InvocableHandlerMethod.java | 10 ++++++---- .../InvocableHandlerMethodKotlinTests.kt | 15 +++++++++++++++ .../result/method/InvocableHandlerMethod.java | 10 ++++++---- .../InvocableHandlerMethodKotlinTests.kt | 18 +++++++++++++++++- 6 files changed, 57 insertions(+), 13 deletions(-) diff --git a/spring-core/src/main/java/org/springframework/core/CoroutinesUtils.java b/spring-core/src/main/java/org/springframework/core/CoroutinesUtils.java index 568d94b6c03..020a00064c9 100644 --- a/spring-core/src/main/java/org/springframework/core/CoroutinesUtils.java +++ b/spring-core/src/main/java/org/springframework/core/CoroutinesUtils.java @@ -134,9 +134,10 @@ public abstract class CoroutinesUtils { Object arg = args[index]; if (!(parameter.isOptional() && arg == null)) { KType type = parameter.getType(); - if (!type.isMarkedNullable() && + if (!(type.isMarkedNullable() && arg == null) && type.getClassifier() instanceof KClass kClass && - KotlinDetector.isInlineClass(JvmClassMappingKt.getJavaClass(kClass))) { + KotlinDetector.isInlineClass(JvmClassMappingKt.getJavaClass(kClass)) && + !JvmClassMappingKt.getJavaClass(kClass).isInstance(arg)) { arg = box(kClass, arg); } argMap.put(parameter, arg); @@ -166,9 +167,10 @@ public abstract class CoroutinesUtils { private static Object box(KClass kClass, @Nullable Object arg) { KFunction constructor = Objects.requireNonNull(KClasses.getPrimaryConstructor(kClass)); KType type = constructor.getParameters().get(0).getType(); - if (!type.isMarkedNullable() && + if (!(type.isMarkedNullable() && arg == null) && type.getClassifier() instanceof KClass parameterClass && - KotlinDetector.isInlineClass(JvmClassMappingKt.getJavaClass(parameterClass))) { + KotlinDetector.isInlineClass(JvmClassMappingKt.getJavaClass(parameterClass)) && + !JvmClassMappingKt.getJavaClass(parameterClass).isInstance(arg)) { arg = box(parameterClass, arg); } if (!KCallablesJvm.isAccessible(constructor)) { diff --git a/spring-core/src/test/kotlin/org/springframework/core/CoroutinesUtilsTests.kt b/spring-core/src/test/kotlin/org/springframework/core/CoroutinesUtilsTests.kt index 4cc2fdeefdb..0f3f4312d2c 100644 --- a/spring-core/src/test/kotlin/org/springframework/core/CoroutinesUtilsTests.kt +++ b/spring-core/src/test/kotlin/org/springframework/core/CoroutinesUtilsTests.kt @@ -236,6 +236,13 @@ class CoroutinesUtilsTests { Assertions.assertThat(mono.awaitSingleOrNull()).isEqualTo("foo") } + @Test + suspend fun invokeSuspendingFunctionWithNullableValueClassParameterAndUnderlyingValue() { + val method = CoroutinesUtilsTests::class.java.declaredMethods.first { it.name.startsWith("suspendingFunctionWithNullableValueClass") } + val mono = CoroutinesUtils.invokeSuspendingFunction(method, this, "foo", null) as Mono + Assertions.assertThat(mono.awaitSingleOrNull()).isEqualTo("foo") + } + @Test suspend fun invokeSuspendingFunctionWithNullableValueClassParameter() { val method = CoroutinesUtilsTests::class.java.declaredMethods.first { it.name.startsWith("suspendingFunctionWithNullableValueClass") } diff --git a/spring-web/src/main/java/org/springframework/web/method/support/InvocableHandlerMethod.java b/spring-web/src/main/java/org/springframework/web/method/support/InvocableHandlerMethod.java index 6dca729698e..1a1ef3a011a 100644 --- a/spring-web/src/main/java/org/springframework/web/method/support/InvocableHandlerMethod.java +++ b/spring-web/src/main/java/org/springframework/web/method/support/InvocableHandlerMethod.java @@ -316,9 +316,10 @@ public class InvocableHandlerMethod extends HandlerMethod { Object arg = args[index]; if (!(parameter.isOptional() && arg == null)) { KType type = parameter.getType(); - if (!type.isMarkedNullable() && + if (!(type.isMarkedNullable() && arg == null) && type.getClassifier() instanceof KClass kClass && - KotlinDetector.isInlineClass(JvmClassMappingKt.getJavaClass(kClass))) { + KotlinDetector.isInlineClass(JvmClassMappingKt.getJavaClass(kClass)) && + !JvmClassMappingKt.getJavaClass(kClass).isInstance(arg)) { arg = box(kClass, arg); } argMap.put(parameter, arg); @@ -337,9 +338,10 @@ public class InvocableHandlerMethod extends HandlerMethod { private static Object box(KClass kClass, @Nullable Object arg) { KFunction constructor = Objects.requireNonNull(KClasses.getPrimaryConstructor(kClass)); KType type = constructor.getParameters().get(0).getType(); - if (!type.isMarkedNullable() && + if (!(type.isMarkedNullable() && arg == null) && type.getClassifier() instanceof KClass parameterClass && - KotlinDetector.isInlineClass(JvmClassMappingKt.getJavaClass(parameterClass))) { + KotlinDetector.isInlineClass(JvmClassMappingKt.getJavaClass(parameterClass)) && + !JvmClassMappingKt.getJavaClass(parameterClass).isInstance(arg)) { arg = box(parameterClass, arg); } if (!KCallablesJvm.isAccessible(constructor)) { diff --git a/spring-web/src/test/kotlin/org/springframework/web/method/support/InvocableHandlerMethodKotlinTests.kt b/spring-web/src/test/kotlin/org/springframework/web/method/support/InvocableHandlerMethodKotlinTests.kt index b7da2b669f2..a3dbd8f91a4 100644 --- a/spring-web/src/test/kotlin/org/springframework/web/method/support/InvocableHandlerMethodKotlinTests.kt +++ b/spring-web/src/test/kotlin/org/springframework/web/method/support/InvocableHandlerMethodKotlinTests.kt @@ -155,6 +155,13 @@ class InvocableHandlerMethodKotlinTests { Assertions.assertThat(value).isEqualTo(1L) } + @Test + fun valueClassWithNullableAndUnderlyingValue() { + composite.addResolver(StubArgumentResolver(LongValueClass::class.java, 1L)) + val value = getInvocable(ValueClassHandler::valueClassWithNullable.javaMethod!!).invokeForRequest(request, null) + Assertions.assertThat(value).isEqualTo(1L) + } + @Test fun valueClassWithNullable() { composite.addResolver(StubArgumentResolver(LongValueClass::class.java, null)) @@ -215,6 +222,14 @@ class InvocableHandlerMethodKotlinTests { StepVerifier.create(value as Mono).verifyComplete() } + @Test + fun suspendingValueClassWithNullableAndUnderlyingValue() { + composite.addResolver(ContinuationHandlerMethodArgumentResolver()) + composite.addResolver(StubArgumentResolver(LongValueClass::class.java, 1L)) + val value = getInvocable(SuspendingValueClassHandler::valueClassWithNullable.javaMethod!!).invokeForRequest(request, null) + StepVerifier.create(value as Mono).expectNext(1L).verifyComplete() + } + @Test fun suspendingValueClassWithPrivateConstructor() { composite.addResolver(ContinuationHandlerMethodArgumentResolver()) diff --git a/spring-webflux/src/main/java/org/springframework/web/reactive/result/method/InvocableHandlerMethod.java b/spring-webflux/src/main/java/org/springframework/web/reactive/result/method/InvocableHandlerMethod.java index c06ea62a2aa..11e312bbe8b 100644 --- a/spring-webflux/src/main/java/org/springframework/web/reactive/result/method/InvocableHandlerMethod.java +++ b/spring-webflux/src/main/java/org/springframework/web/reactive/result/method/InvocableHandlerMethod.java @@ -356,9 +356,10 @@ public class InvocableHandlerMethod extends HandlerMethod { Object arg = args[index]; if (!(parameter.isOptional() && arg == null)) { KType type = parameter.getType(); - if (!type.isMarkedNullable() && + if (!(type.isMarkedNullable() && arg == null) && type.getClassifier() instanceof KClass kClass && - KotlinDetector.isInlineClass(JvmClassMappingKt.getJavaClass(kClass))) { + KotlinDetector.isInlineClass(JvmClassMappingKt.getJavaClass(kClass)) && + !JvmClassMappingKt.getJavaClass(kClass).isInstance(arg)) { arg = box(kClass, arg); } argMap.put(parameter, arg); @@ -378,9 +379,10 @@ public class InvocableHandlerMethod extends HandlerMethod { private static Object box(KClass kClass, @Nullable Object arg) { KFunction constructor = Objects.requireNonNull(KClasses.getPrimaryConstructor(kClass)); KType type = constructor.getParameters().get(0).getType(); - if (!type.isMarkedNullable() && + if (!(type.isMarkedNullable() && arg == null) && type.getClassifier() instanceof KClass parameterClass && - KotlinDetector.isInlineClass(JvmClassMappingKt.getJavaClass(parameterClass))) { + KotlinDetector.isInlineClass(JvmClassMappingKt.getJavaClass(parameterClass)) && + !JvmClassMappingKt.getJavaClass(parameterClass).isInstance(arg)) { arg = box(parameterClass, arg); } if (!KCallablesJvm.isAccessible(constructor)) { diff --git a/spring-webflux/src/test/kotlin/org/springframework/web/reactive/result/InvocableHandlerMethodKotlinTests.kt b/spring-webflux/src/test/kotlin/org/springframework/web/reactive/result/InvocableHandlerMethodKotlinTests.kt index 9e41a222399..f8041148619 100644 --- a/spring-webflux/src/test/kotlin/org/springframework/web/reactive/result/InvocableHandlerMethodKotlinTests.kt +++ b/spring-webflux/src/test/kotlin/org/springframework/web/reactive/result/InvocableHandlerMethodKotlinTests.kt @@ -258,6 +258,14 @@ class InvocableHandlerMethodKotlinTests { assertHandlerResultValue(result, "1") } + @Test + fun valueClassWithNullableAndUnderlyingValue() { + this.resolvers.add(stubResolver(1L, LongValueClass::class.java)) + val method = ValueClassController::valueClassWithNullable.javaMethod!! + val result = invoke(ValueClassController(), method) + assertHandlerResultValue(result, "1") + } + @Test fun valueClassWithNullable() { this.resolvers.add(stubResolver(null, LongValueClass::class.java)) @@ -320,6 +328,14 @@ class InvocableHandlerMethodKotlinTests { assertHandlerResultValue(result, "null") } + @Test + fun suspendingValueClassWithNullableAndUnderlyingValue() { + this.resolvers.add(stubResolver(1L, LongValueClass::class.java)) + val method = SuspendingValueClassController::valueClassWithNullable.javaMethod!! + val result = invoke(SuspendingValueClassController(), method) + assertHandlerResultValue(result, "1") + } + @Test fun suspendingValueClassWithPrivateConstructor() { this.resolvers.add(stubResolver(1L, Long::class.java)) @@ -590,4 +606,4 @@ class InvocableHandlerMethodKotlinTests { } class CustomException(message: String) : Throwable(message) -} \ No newline at end of file +} From 29a7402adf2a02b194e844c0200fd899888f8c37 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?S=C3=A9bastien=20Deleuze?= Date: Fri, 27 Mar 2026 17:34:44 +0100 Subject: [PATCH 2/3] Fix WebClient context propagation in Kotlin Coroutines Prior to this commit, thread-local variables like Trace ID were not automatically propagated into the Reactor Context when making requests using Kotlin coroutine WebClient extensions like `awaitExchange`. This commit updates `CoroutineContext.toReactorContext()` to capture thread-local values via Micrometer Context Propagation when available, ensuring observations and traces are properly reused. Closes gh-36182 --- spring-webflux/spring-webflux.gradle | 1 + .../function/client/WebClientExtensions.kt | 24 +++++- .../client/WebClientExtensionsTests.kt | 84 +++++++++++++++++++ 3 files changed, 107 insertions(+), 2 deletions(-) diff --git a/spring-webflux/spring-webflux.gradle b/spring-webflux/spring-webflux.gradle index dad853c5430..16c9be3a271 100644 --- a/spring-webflux/spring-webflux.gradle +++ b/spring-webflux/spring-webflux.gradle @@ -32,6 +32,7 @@ dependencies { optional("org.jetbrains.kotlin:kotlin-reflect") optional("org.jetbrains.kotlin:kotlin-stdlib") optional("org.jetbrains.kotlinx:kotlinx-coroutines-reactor") + optional("io.micrometer:context-propagation") optional("org.webjars:webjars-locator-lite") optional("tools.jackson.core:jackson-databind") optional("tools.jackson.dataformat:jackson-dataformat-smile") diff --git a/spring-webflux/src/main/kotlin/org/springframework/web/reactive/function/client/WebClientExtensions.kt b/spring-webflux/src/main/kotlin/org/springframework/web/reactive/function/client/WebClientExtensions.kt index d993ea1eacc..c426db2aabc 100644 --- a/spring-webflux/src/main/kotlin/org/springframework/web/reactive/function/client/WebClientExtensions.kt +++ b/spring-webflux/src/main/kotlin/org/springframework/web/reactive/function/client/WebClientExtensions.kt @@ -16,6 +16,8 @@ package org.springframework.web.reactive.function.client +import io.micrometer.context.ContextRegistry +import io.micrometer.context.ContextSnapshotFactory import kotlinx.coroutines.Job import kotlinx.coroutines.currentCoroutineContext import kotlinx.coroutines.flow.Flow @@ -25,6 +27,7 @@ import kotlinx.coroutines.withContext import org.reactivestreams.Publisher import org.springframework.core.ParameterizedTypeReference import org.springframework.http.ResponseEntity +import org.springframework.util.ClassUtils import org.springframework.web.reactive.function.client.CoExchangeFilterFunction.Companion.COROUTINE_CONTEXT_ATTRIBUTE import org.springframework.web.reactive.function.client.WebClient.RequestBodySpec import org.springframework.web.reactive.function.client.WebClient.RequestHeadersSpec @@ -237,8 +240,25 @@ suspend inline fun WebClient.ResponseSpec.awaitEntity(): Respo } } +private val contextPropagationPresent = ClassUtils.isPresent("io.micrometer.context.ContextSnapshotFactory", + WebClient::class.java.classLoader) + @PublishedApi internal fun CoroutineContext.toReactorContext(): ReactorContext { - val context = Context.of(COROUTINE_CONTEXT_ATTRIBUTE, this).readOnly() - return (this[ReactorContext.Key]?.context?.putAll(context) ?: context).asCoroutineContext() + var context = Context.of(COROUTINE_CONTEXT_ATTRIBUTE, this) + if (contextPropagationPresent) { + context = ContextPropagationDelegate.captureThreadLocalsInto(context) + } + val readOnlyContext = context.readOnly() + return (this[ReactorContext.Key]?.context?.putAll(readOnlyContext) ?: readOnlyContext).asCoroutineContext() +} + +private object ContextPropagationDelegate { + + private val contextSnapshotFactory = ContextSnapshotFactory.builder() + .contextRegistry(ContextRegistry.getInstance()).build() + + fun captureThreadLocalsInto(context: Context): Context { + return contextSnapshotFactory.captureAll().updateContext(context) + } } diff --git a/spring-webflux/src/test/kotlin/org/springframework/web/reactive/function/client/WebClientExtensionsTests.kt b/spring-webflux/src/test/kotlin/org/springframework/web/reactive/function/client/WebClientExtensionsTests.kt index 514deffc602..78943f1013c 100644 --- a/spring-webflux/src/test/kotlin/org/springframework/web/reactive/function/client/WebClientExtensionsTests.kt +++ b/spring-webflux/src/test/kotlin/org/springframework/web/reactive/function/client/WebClientExtensionsTests.kt @@ -16,6 +16,9 @@ package org.springframework.web.reactive.function.client +import io.micrometer.observation.Observation +import io.micrometer.observation.ObservationHandler +import io.micrometer.observation.ObservationRegistry import io.mockk.every import io.mockk.mockk import io.mockk.slot @@ -30,11 +33,13 @@ import org.assertj.core.api.Assertions.assertThat import org.junit.jupiter.api.Test import org.reactivestreams.Publisher import org.springframework.core.ParameterizedTypeReference +import org.springframework.core.PropagationContextElement import org.springframework.http.HttpHeaders import org.springframework.http.HttpStatus import org.springframework.http.ResponseEntity import org.springframework.web.reactive.function.client.CoExchangeFilterFunction.Companion.COROUTINE_CONTEXT_ATTRIBUTE import reactor.core.publisher.Flux +import reactor.core.publisher.Hooks import reactor.core.publisher.Mono import java.time.Duration import java.util.concurrent.CompletableFuture @@ -433,9 +438,88 @@ class WebClientExtensionsTests { } } + @Test + fun `awaitExchange preserves parent observation with automatic context propagation`() { + Hooks.enableAutomaticContextPropagation() + try { + val observationRegistry = ObservationRegistry.create() + val contextObservationHandler = ContextObservationHandler() + observationRegistry.observationConfig().observationHandler(contextObservationHandler) + val exchangeFunction = mockk() + val mockResponse = mockk() + every { exchangeFunction.exchange(any()) } returns Mono.just(mockResponse) + every { mockResponse.statusCode() } returns HttpStatus.OK + every { mockResponse.releaseBody() } returns Mono.empty() + + val parent = Observation.start("parent", observationRegistry) + val scope = parent.openScope() + try { + runBlocking(PropagationContextElement()) { + val webClient = WebClient.builder() + .exchangeFunction(exchangeFunction) + .observationRegistry(observationRegistry) + .build() + webClient.get().uri("/path1").awaitExchange { it.statusCode() } + webClient.get().uri("/path2").awaitExchange { it.statusCode() } + } + } finally { + scope.close() + parent.stop() + } + assertThat(contextObservationHandler.parentObservation).containsExactly(true, true) + } finally { + Hooks.disableAutomaticContextPropagation() + } + } + + @Test + fun `awaitBody preserves parent observation with automatic context propagation`() { + Hooks.enableAutomaticContextPropagation() + try { + val observationRegistry = ObservationRegistry.create() + val contextObservationHandler = ContextObservationHandler() + observationRegistry.observationConfig().observationHandler(contextObservationHandler) + val exchangeFunction = mockk() + val mockResponse = mockk() + every { exchangeFunction.exchange(any()) } returns Mono.just(mockResponse) + every { mockResponse.statusCode() } returns HttpStatus.OK + every { mockResponse.bodyToMono(object : ParameterizedTypeReference() {}) } returns Mono.just("body") + + val parent = Observation.start("parent", observationRegistry) + val scope = parent.openScope() + try { + runBlocking(PropagationContextElement()) { + val webClient = WebClient.builder() + .exchangeFunction(exchangeFunction) + .observationRegistry(observationRegistry) + .build() + webClient.get().uri("/path1").retrieve().awaitBody() + webClient.get().uri("/path2").retrieve().awaitBody() + } + } finally { + scope.close() + parent.stop() + } + assertThat(contextObservationHandler.parentObservation).containsExactly(true, true) + } finally { + Hooks.disableAutomaticContextPropagation() + } + } + class Foo private data class FooContextElement(val foo: Foo) : AbstractCoroutineContextElement(FooContextElement) { companion object Key : CoroutineContext.Key } + + private class ContextObservationHandler : ObservationHandler { + + val parentObservation = mutableListOf() + + override fun onStart(context: ClientRequestObservationContext) { + parentObservation.add(context.parentObservation != null) + } + + override fun supportsContext(context: Observation.Context) = context is ClientRequestObservationContext + } } From 8d936704301e58d62c7beb71681d4be394bcdf32 Mon Sep 17 00:00:00 2001 From: Dmitry Sulman Date: Fri, 17 Apr 2026 20:43:12 +0300 Subject: [PATCH 3/3] Support Micrometer context propagation in Kotlin Flow See gh-36427 Closes gh-36667 Signed-off-by: Dmitry Sulman --- .../core/ReactiveAdapterRegistry.java | 7 +++- .../ReactiveAdapterRegistryKotlinTests.kt | 24 ++++++++++++ .../annotation/CoroutinesIntegrationTests.kt | 37 +++++++++++++++---- 3 files changed, 59 insertions(+), 9 deletions(-) diff --git a/spring-core/src/main/java/org/springframework/core/ReactiveAdapterRegistry.java b/spring-core/src/main/java/org/springframework/core/ReactiveAdapterRegistry.java index 7d66979afb2..5be957ecc2b 100644 --- a/spring-core/src/main/java/org/springframework/core/ReactiveAdapterRegistry.java +++ b/spring-core/src/main/java/org/springframework/core/ReactiveAdapterRegistry.java @@ -71,6 +71,8 @@ public class ReactiveAdapterRegistry { private static final boolean MUTINY_PRESENT; + private static final boolean CONTEXT_PROPAGATION_PRESENT; + static { ClassLoader classLoader = ReactiveAdapterRegistry.class.getClassLoader(); REACTIVE_STREAMS_PRESENT = ClassUtils.isPresent("org.reactivestreams.Publisher", classLoader); @@ -78,6 +80,7 @@ public class ReactiveAdapterRegistry { RXJAVA_3_PRESENT = ClassUtils.isPresent("io.reactivex.rxjava3.core.Flowable", classLoader); COROUTINES_REACTOR_PRESENT = ClassUtils.isPresent("kotlinx.coroutines.reactor.MonoKt", classLoader); MUTINY_PRESENT = ClassUtils.isPresent("io.smallrye.mutiny.Multi", classLoader); + CONTEXT_PROPAGATION_PRESENT = ClassUtils.isPresent("io.micrometer.context.ContextSnapshotFactory", classLoader); } private final List adapters = new ArrayList<>(); @@ -356,7 +359,9 @@ public class ReactiveAdapterRegistry { registry.registerReactiveType( ReactiveTypeDescriptor.multiValue(kotlinx.coroutines.flow.Flow.class, kotlinx.coroutines.flow.FlowKt::emptyFlow), - source -> kotlinx.coroutines.reactor.ReactorFlowKt.asFlux((kotlinx.coroutines.flow.Flow) source), + CONTEXT_PROPAGATION_PRESENT ? + source -> kotlinx.coroutines.reactor.ReactorFlowKt.asFlux((kotlinx.coroutines.flow.Flow) source, new PropagationContextElement()) : + source -> kotlinx.coroutines.reactor.ReactorFlowKt.asFlux((kotlinx.coroutines.flow.Flow) source), kotlinx.coroutines.reactive.ReactiveFlowKt::asFlow); } } diff --git a/spring-core/src/test/kotlin/org/springframework/core/ReactiveAdapterRegistryKotlinTests.kt b/spring-core/src/test/kotlin/org/springframework/core/ReactiveAdapterRegistryKotlinTests.kt index 817fecbed5b..26b244256f6 100644 --- a/spring-core/src/test/kotlin/org/springframework/core/ReactiveAdapterRegistryKotlinTests.kt +++ b/spring-core/src/test/kotlin/org/springframework/core/ReactiveAdapterRegistryKotlinTests.kt @@ -16,13 +16,18 @@ package org.springframework.core +import io.micrometer.observation.Observation +import io.micrometer.observation.tck.TestObservationRegistry import kotlinx.coroutines.Deferred import kotlinx.coroutines.DelicateCoroutinesApi +import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.GlobalScope import kotlinx.coroutines.async import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.flow import kotlinx.coroutines.flow.toList +import kotlinx.coroutines.reactive.awaitSingle +import kotlinx.coroutines.runBlocking import org.assertj.core.api.Assertions.assertThat import org.junit.jupiter.api.Test import org.reactivestreams.Publisher @@ -40,6 +45,8 @@ import kotlin.reflect.KClass @OptIn(DelicateCoroutinesApi::class) class ReactiveAdapterRegistryKotlinTests { + private val observationRegistry = TestObservationRegistry.create() + private val registry = ReactiveAdapterRegistry.getSharedInstance() @Test @@ -82,6 +89,23 @@ class ReactiveAdapterRegistryKotlinTests { assertThat((target as Flow<*>).toList()).contains(1, 2, 3) } + @Test + fun propagateMicrometerContextToFlow() { + val source = flow { + val currentObservation = observationRegistry.currentObservation + assertThat(currentObservation).isNotNull + emit(currentObservation?.context?.name) + } + val observation = Observation.createNotStarted("coroutine", observationRegistry) + observation.observe { + val target: Publisher = getAdapter(Flow::class).toPublisher(source) + val result = runBlocking(Dispatchers.IO) { + target.awaitSingle() + } + assertThat(result).isEqualTo("coroutine") + } + } + private fun getAdapter(reactiveType: KClass<*>): ReactiveAdapter { return this.registry.getAdapter(reactiveType.java)!! } diff --git a/spring-webflux/src/test/kotlin/org/springframework/web/reactive/result/method/annotation/CoroutinesIntegrationTests.kt b/spring-webflux/src/test/kotlin/org/springframework/web/reactive/result/method/annotation/CoroutinesIntegrationTests.kt index b932e5204b4..6850468bb22 100644 --- a/spring-webflux/src/test/kotlin/org/springframework/web/reactive/result/method/annotation/CoroutinesIntegrationTests.kt +++ b/spring-webflux/src/test/kotlin/org/springframework/web/reactive/result/method/annotation/CoroutinesIntegrationTests.kt @@ -16,18 +16,16 @@ package org.springframework.web.reactive.result.method.annotation -import kotlinx.coroutines.Deferred -import kotlinx.coroutines.DelicateCoroutinesApi -import kotlinx.coroutines.GlobalScope -import kotlinx.coroutines.async -import kotlinx.coroutines.delay +import io.micrometer.observation.ObservationRegistry +import io.micrometer.observation.tck.TestObservationRegistry +import kotlinx.coroutines.* import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.flow import org.assertj.core.api.Assertions.assertThat import org.assertj.core.api.Assertions.assertThatExceptionOfType -import org.junit.jupiter.api.Assumptions.assumeFalse import org.springframework.context.ApplicationContext import org.springframework.context.annotation.AnnotationConfigApplicationContext +import org.springframework.context.annotation.Bean import org.springframework.context.annotation.ComponentScan import org.springframework.context.annotation.Configuration import org.springframework.http.HttpHeaders @@ -86,6 +84,15 @@ class CoroutinesIntegrationTests : AbstractRequestMappingIntegrationTests() { assertThat(entity.body).isEqualTo("foobar") } + @ParameterizedHttpServerTest + fun `Handler method returning Flow with observation`(httpServer: HttpServer) { + startServer(httpServer) + + val entity = performGet("/flow-observation", HttpHeaders.EMPTY, String::class.java) + assertThat(entity.statusCode).isEqualTo(HttpStatus.OK) + assertThat(entity.body).isEqualTo("http.server.requests") + } + @ParameterizedHttpServerTest fun `Suspending handler method returning Flow`(httpServer: HttpServer) { startServer(httpServer) @@ -135,11 +142,16 @@ class CoroutinesIntegrationTests : AbstractRequestMappingIntegrationTests() { @Configuration @EnableWebFlux @ComponentScan(resourcePattern = "**/CoroutinesIntegrationTests*") - open class WebConfig + open class WebConfig { + + @Bean + open fun observationRegistry(): ObservationRegistry = TestObservationRegistry.create() + + } @OptIn(DelicateCoroutinesApi::class) @RestController - class CoroutinesController { + class CoroutinesController(private val observationRegistry: ObservationRegistry) { @GetMapping("/suspend") suspend fun suspendingEndpoint(): String { @@ -167,6 +179,15 @@ class CoroutinesIntegrationTests : AbstractRequestMappingIntegrationTests() { delay(1) } + @GetMapping("/flow-observation") + fun flowObservationEndpoint(): Flow { + return flow { + val currentObservation = observationRegistry.currentObservation + assertThat(currentObservation).isNotNull + emit(currentObservation?.context?.name) + } + } + @GetMapping("/suspending-flow") suspend fun suspendingFlowEndpoint(): Flow { delay(1)