Create spring-boot-reactor module

spring-boot-project/spring-boot-reactor/build.gradle

@@ -0,0 +1,22 @@
+plugins {
+	id "java-library"
+	id "org.springframework.boot.auto-configuration"
+	id "org.springframework.boot.configuration-properties"
+	id "org.springframework.boot.deployed"
+	id "org.springframework.boot.optional-dependencies"
+}
+
+description = "Spring Boot Reactor"
+
+dependencies {
+	api(project(":spring-boot-project:spring-boot"))
+	api("io.projectreactor:reactor-core")
+
+	optional(project(":spring-boot-project:spring-boot-autoconfigure"))
+
+	testImplementation(project(":spring-boot-project:spring-boot-test"))
+	testImplementation(project(":spring-boot-project:spring-boot-tools:spring-boot-test-support"))
+	testImplementation("io.micrometer:context-propagation")
+
+	testRuntimeOnly("ch.qos.logback:logback-classic")
+}

spring-boot-project/spring-boot-reactor/src/main/java/org/springframework/boot/reactor/ReactorEnvironmentPostProcessor.java

@@ -0,0 +1,70 @@
+/*
+ * Copyright 2012-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.boot.reactor;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.env.EnvironmentPostProcessor;
+import org.springframework.boot.system.JavaVersion;
+import org.springframework.core.Ordered;
+import org.springframework.core.env.ConfigurableEnvironment;
+import org.springframework.util.ClassUtils;
+
+/**
+ * {@link EnvironmentPostProcessor} to enable the Reactor global features as early as
+ * possible in the startup process.
+ * <p>
+ * If the "reactor-tools" dependency is available, the debug agent is enabled by default,
+ * unless the {@code "spring.reactor.debug-agent.enabled"} configuration property is set
+ * to false.
+ * <p>
+ * If the {@code "spring.threads.virtual.enabled"} property is enabled and the current JVM
+ * is 21 or later, then the Reactor System property is set to configure the Bounded
+ * Elastic Scheduler to use Virtual Threads globally.
+ *
+ * @author Brian Clozel
+ * @since 3.2.0
+ */
+public class ReactorEnvironmentPostProcessor implements EnvironmentPostProcessor, Ordered {
+
+	private static final String REACTOR_DEBUGAGENT_CLASS = "reactor.tools.agent.ReactorDebugAgent";
+
+	@Override
+	public void postProcessEnvironment(ConfigurableEnvironment environment, SpringApplication application) {
+		if (ClassUtils.isPresent(REACTOR_DEBUGAGENT_CLASS, null)) {
+			Boolean agentEnabled = environment.getProperty("spring.reactor.debug-agent.enabled", Boolean.class);
+			if (agentEnabled != Boolean.FALSE) {
+				try {
+					Class<?> debugAgent = Class.forName(REACTOR_DEBUGAGENT_CLASS);
+					debugAgent.getMethod("init").invoke(null);
+				}
+				catch (Exception ex) {
+					throw new RuntimeException("Failed to init Reactor's debug agent", ex);
+				}
+			}
+		}
+		if (environment.getProperty("spring.threads.virtual.enabled", boolean.class, false)
+				&& JavaVersion.getJavaVersion().isEqualOrNewerThan(JavaVersion.TWENTY_ONE)) {
+			System.setProperty("reactor.schedulers.defaultBoundedElasticOnVirtualThreads", "true");
+		}
+	}
+
+	@Override
+	public int getOrder() {
+		return Ordered.LOWEST_PRECEDENCE;
+	}
+
+}

spring-boot-project/spring-boot-reactor/src/main/java/org/springframework/boot/reactor/autoconfigure/ReactorAutoConfiguration.java

@@ -0,0 +1,43 @@
+/*
+ * Copyright 2012-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.boot.reactor.autoconfigure;
+
+import reactor.core.publisher.Hooks;
+
+import org.springframework.boot.autoconfigure.AutoConfiguration;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+
+/**
+ * {@link EnableAutoConfiguration Auto-configuration} for Reactor.
+ *
+ * @author Brian Clozel
+ * @since 4.0.0
+ */
+@AutoConfiguration
+@ConditionalOnClass(Hooks.class)
+@EnableConfigurationProperties(ReactorProperties.class)
+public class ReactorAutoConfiguration {
+
+	ReactorAutoConfiguration(ReactorProperties properties) {
+		if (properties.getContextPropagation() == ReactorProperties.ContextPropagationMode.AUTO) {
+			Hooks.enableAutomaticContextPropagation();
+		}
+	}
+
+}

spring-boot-project/spring-boot-reactor/src/main/java/org/springframework/boot/reactor/autoconfigure/ReactorProperties.java

@@ -0,0 +1,57 @@
+/*
+ * Copyright 2012-2025 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.boot.reactor.autoconfigure;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+
+/**
+ * Configuration properties for Reactor.
+ *
+ * @author Brian Clozel
+ * @since 4.0.0
+ */
+@ConfigurationProperties("spring.reactor")
+public class ReactorProperties {
+
+	/**
+	 * Context Propagation support mode for Reactor operators.
+	 */
+	private ContextPropagationMode contextPropagation = ContextPropagationMode.LIMITED;
+
+	public ContextPropagationMode getContextPropagation() {
+		return this.contextPropagation;
+	}
+
+	public void setContextPropagation(ContextPropagationMode contextPropagation) {
+		this.contextPropagation = contextPropagation;
+	}
+
+	public enum ContextPropagationMode {
+
+		/**
+		 * Context Propagation is applied to all Reactor operators.
+		 */
+		AUTO,
+
+		/**
+		 * Context Propagation is only applied to "tap" and "handle" Reactor operators.
+		 */
+		LIMITED
+
+	}
+
+}

spring-boot-project/spring-boot-reactor/src/main/java/org/springframework/boot/reactor/autoconfigure/package-info.java

@@ -0,0 +1,20 @@
+/*
+ * Copyright 2012-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.
+ */
+
+/**
+ * Auto-configuration for Reactor.
+ */
+package org.springframework.boot.reactor.autoconfigure;

spring-boot-project/spring-boot-reactor/src/main/java/org/springframework/boot/reactor/package-info.java

@@ -0,0 +1,20 @@
+/*
+ * Copyright 2012-2019 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.
+ */
+
+/**
+ * Support classes for Reactor integration.
+ */
+package org.springframework.boot.reactor;

spring-boot-project/spring-boot-reactor/src/main/resources/META-INF/additional-spring-configuration-metadata.json

@@ -0,0 +1,13 @@
+{
+  "groups": [],
+  "properties": [
+      {
+      "name": "spring.reactor.stacktrace-mode.enabled",
+      "description": "Whether Reactor should collect stacktrace information at runtime.",
+      "defaultValue": false,
+      "deprecation": {
+        "replacement": "spring.reactor.debug-agent.enabled"
+      }
+    }
+  ]
+}

spring-boot-project/spring-boot-reactor/src/main/resources/META-INF/spring.factories

@@ -0,0 +1,3 @@
+# Environment Post Processors
+org.springframework.boot.env.EnvironmentPostProcessor=\
+org.springframework.boot.reactor.ReactorEnvironmentPostProcessor

spring-boot-project/spring-boot-reactor/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports

@@ -0,0 +1 @@
+org.springframework.boot.reactor.autoconfigure.ReactorAutoConfiguration

spring-boot-project/spring-boot-reactor/src/test/java/org/springframework/boot/reactor/InstrumentedFluxProvider.java

@@ -0,0 +1,33 @@
+/*
+ * Copyright 2012-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.boot.reactor;
+
+import reactor.core.publisher.Flux;
+
+/**
+ * Utility class that should be instrumented by the reactor debug agent.
+ *
+ * @author Brian Clozel
+ * @see ReactorEnvironmentPostProcessorTests
+ */
+class InstrumentedFluxProvider {
+
+	Flux<Integer> newFluxJust() {
+		return Flux.just(1);
+	}
+
+}

spring-boot-project/spring-boot-reactor/src/test/java/org/springframework/boot/reactor/ReactorEnvironmentPostProcessorTests.java

@@ -0,0 +1,72 @@
+/*
+ * Copyright 2012-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.boot.reactor;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.EnabledForJreRange;
+import org.junit.jupiter.api.condition.JRE;
+import reactor.core.Scannable;
+import reactor.core.publisher.Flux;
+
+import org.springframework.mock.env.MockEnvironment;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Tests for {@link ReactorEnvironmentPostProcessor}.
+ *
+ * @author Brian Clozel
+ */
+
+@Disabled("Tests rely on static initialization and are flaky on CI")
+class ReactorEnvironmentPostProcessorTests {
+
+	static {
+		MockEnvironment environment = new MockEnvironment();
+		environment.setProperty("spring.threads.virtual.enabled", "true");
+		ReactorEnvironmentPostProcessor postProcessor = new ReactorEnvironmentPostProcessor();
+		postProcessor.postProcessEnvironment(environment, null);
+	}
+
+	@Test
+	void enablesReactorDebugAgent() {
+		InstrumentedFluxProvider fluxProvider = new InstrumentedFluxProvider();
+		Flux<Integer> flux = fluxProvider.newFluxJust();
+		assertThat(Scannable.from(flux).stepName())
+			.startsWith("Flux.just ⇢ at org.springframework.boot.reactor.InstrumentedFluxProvider.newFluxJust");
+	}
+
+	@Test
+	@EnabledForJreRange(max = JRE.JAVA_20)
+	void shouldNotEnableVirtualThreads() {
+		assertThat(System.getProperty("reactor.schedulers.defaultBoundedElasticOnVirtualThreads")).isNotEqualTo("true");
+	}
+
+	@Test
+	@EnabledForJreRange(min = JRE.JAVA_21)
+	void shouldEnableVirtualThreads() {
+		assertThat(System.getProperty("reactor.schedulers.defaultBoundedElasticOnVirtualThreads")).isEqualTo("true");
+	}
+
+	@AfterEach
+	void cleanup() {
+		System.setProperty("reactor.schedulers.defaultBoundedElasticOnVirtualThreads", "false");
+	}
+
+}

spring-boot-project/spring-boot-reactor/src/test/java/org/springframework/boot/reactor/autoconfigure/ReactorAutoConfigurationTests.java

@@ -0,0 +1,93 @@
+/*
+ * Copyright 2012-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.boot.reactor.autoconfigure;
+
+import java.util.concurrent.atomic.AtomicReference;
+
+import io.micrometer.context.ContextRegistry;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import reactor.core.publisher.Hooks;
+import reactor.core.publisher.Mono;
+import reactor.util.context.Context;
+
+import org.springframework.boot.autoconfigure.AutoConfigurations;
+import org.springframework.boot.test.context.runner.ApplicationContextRunner;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Tests for {@link ReactorAutoConfiguration}.
+ *
+ * @author Brian Clozel
+ * @author Moritz Halbritter
+ */
+class ReactorAutoConfigurationTests {
+
+	private final ApplicationContextRunner contextRunner = new ApplicationContextRunner()
+		.withConfiguration(AutoConfigurations.of(ReactorAutoConfiguration.class));
+
+	private static final String THREADLOCAL_KEY = "ReactorAutoConfigurationTests";
+
+	private static final ThreadLocal<String> THREADLOCAL_VALUE = ThreadLocal.withInitial(() -> "initial");
+
+	@BeforeEach
+	@AfterEach
+	void resetStaticState() {
+		Hooks.disableAutomaticContextPropagation();
+	}
+
+	@BeforeAll
+	static void initializeThreadLocalAccessors() {
+		ContextRegistry globalRegistry = ContextRegistry.getInstance();
+		globalRegistry.registerThreadLocalAccessor(THREADLOCAL_KEY, THREADLOCAL_VALUE);
+	}
+
+	@AfterAll
+	static void removeThreadLocalAccessors() {
+		ContextRegistry globalRegistry = ContextRegistry.getInstance();
+		globalRegistry.removeThreadLocalAccessor(THREADLOCAL_KEY);
+	}
+
+	@Test
+	void shouldNotConfigurePropagationByDefault() {
+		AtomicReference<String> threadLocalValue = new AtomicReference<>();
+		this.contextRunner.run((applicationContext) -> {
+			Mono.just("test")
+				.doOnNext((element) -> threadLocalValue.set(THREADLOCAL_VALUE.get()))
+				.contextWrite(Context.of(THREADLOCAL_KEY, "updated"))
+				.block();
+			assertThat(threadLocalValue.get()).isEqualTo("initial");
+		});
+	}
+
+	@Test
+	void shouldConfigurePropagationIfSetToAuto() {
+		AtomicReference<String> threadLocalValue = new AtomicReference<>();
+		this.contextRunner.withPropertyValues("spring.reactor.context-propagation=auto").run((applicationContext) -> {
+			Mono.just("test")
+				.doOnNext((element) -> threadLocalValue.set(THREADLOCAL_VALUE.get()))
+				.contextWrite(Context.of(THREADLOCAL_KEY, "updated"))
+				.block();
+			assertThat(threadLocalValue.get()).isEqualTo("updated");
+		});
+	}
+
+}