Support scheduler qualifier for reactive SubscribingRunnable as well

See gh-20818
spring-projects · Jul 8, 2023 · 5243c22 · 5243c22
1 parent a0c80ff
commit 5243c22
Showing 1 changed file with 23 additions and 7 deletions.
diff --git a/...in/java/org/springframework/scheduling/annotation/ScheduledAnnotationReactiveSupport.java b/...in/java/org/springframework/scheduling/annotation/ScheduledAnnotationReactiveSupport.java
@@ -38,6 +38,7 @@
 import org.springframework.core.ReactiveAdapter;
 import org.springframework.core.ReactiveAdapterRegistry;
 import org.springframework.lang.Nullable;
+import org.springframework.scheduling.SchedulingAwareRunnable;
 import org.springframework.scheduling.support.DefaultScheduledTaskObservationConvention;
 import org.springframework.scheduling.support.ScheduledTaskObservationContext;
 import org.springframework.scheduling.support.ScheduledTaskObservationConvention;
@@ -120,8 +121,10 @@ public static Runnable createSubscriptionRunnable(Method method, Object targetBe
 
 		boolean shouldBlock = (scheduled.fixedDelay() > 0 || StringUtils.hasText(scheduled.fixedDelayString()));
 		Publisher<?> publisher = getPublisherFor(method, targetBean);
-		Supplier<ScheduledTaskObservationContext> contextSupplier = () -> new ScheduledTaskObservationContext(targetBean, method);
-		return new SubscribingRunnable(publisher, shouldBlock, subscriptionTrackerRegistry, observationRegistrySupplier, contextSupplier);
+		Supplier<ScheduledTaskObservationContext> contextSupplier =
+				() -> new ScheduledTaskObservationContext(targetBean, method);
+		return new SubscribingRunnable(publisher, shouldBlock, scheduled.scheduler(),
+				subscriptionTrackerRegistry, observationRegistrySupplier, contextSupplier);
 	}
 
 	/**
@@ -180,30 +183,43 @@ static Publisher<?> getPublisherFor(Method method, Object bean) {
 	 * Utility implementation of {@code Runnable} that subscribes to a {@code Publisher}
 	 * or subscribes-then-blocks if {@code shouldBlock} is set to {@code true}.
 	 */
-	static final class SubscribingRunnable implements Runnable {
+	static final class SubscribingRunnable implements SchedulingAwareRunnable {
 
-		private final Publisher<?> publisher;
+		private static final ScheduledTaskObservationConvention DEFAULT_CONVENTION =
+				new DefaultScheduledTaskObservationConvention();
 
-		private static final ScheduledTaskObservationConvention DEFAULT_CONVENTION = new DefaultScheduledTaskObservationConvention();
+		private final Publisher<?> publisher;
 
 		final boolean shouldBlock;
 
+		@Nullable
+		private final String qualifier;
+
 		private final List<Runnable> subscriptionTrackerRegistry;
 
 		final Supplier<ObservationRegistry> observationRegistrySupplier;
 
 		final Supplier<ScheduledTaskObservationContext> contextSupplier;
 
-		SubscribingRunnable(Publisher<?> publisher, boolean shouldBlock, List<Runnable> subscriptionTrackerRegistry,
-				Supplier<ObservationRegistry> observationRegistrySupplier, Supplier<ScheduledTaskObservationContext> contextSupplier) {
+		SubscribingRunnable(Publisher<?> publisher, boolean shouldBlock,
+				@Nullable String qualifier, List<Runnable> subscriptionTrackerRegistry,
+				Supplier<ObservationRegistry> observationRegistrySupplier,
+				Supplier<ScheduledTaskObservationContext> contextSupplier) {
 
 			this.publisher = publisher;
 			this.shouldBlock = shouldBlock;
+			this.qualifier = qualifier;
 			this.subscriptionTrackerRegistry = subscriptionTrackerRegistry;
 			this.observationRegistrySupplier = observationRegistrySupplier;
 			this.contextSupplier = contextSupplier;
 		}
 
+		@Override
+		@Nullable
+		public String getQualifier() {
+			return this.qualifier;
+		}
+
 		@Override
 		public void run() {
 			Observation observation = TASKS_SCHEDULED_EXECUTION.observation(null, DEFAULT_CONVENTION,