Pivot to use the existing Runnable-based infrastructure

Author: simonbasle

framework-docs/modules/ROOT/pages/integration/scheduling.adoc

@@ -452,17 +452,13 @@ registry denotes that by having the `getDescriptor().isDeferred()` method return
 	}
 ----
 
-All these types of methods must be declared without any arguments. Additionally, the `cron`
-configuration elements are not supported for these cases.
-
-Note that the `TaskExecutor`/`TaskScheduler` infrastructure is not used and the scheduling
-is instead performed by the Reactor library, which must be present at runtime.
-In the case of Kotlin suspending functions the `kotlinx.coroutines.reactor` bridge must
-also be present.
+All these types of methods must be declared without any arguments. In the case of Kotlin
+suspending functions the `kotlinx.coroutines.reactor` bridge must also be present to allow
+the framework to invoke a suspending function as a `Publisher`.
 
 The Spring Framework will obtain a `Publisher` out of the annotated method once and will
-create a timed `Flux` which regularly subscribes to said `Publisher`. These inner regular
-subscriptions happen according to the `fixedDelay`/`fixedRate` configuration.
+schedule a `Runnable` in which it subscribes to said `Publisher`. These inner regular
+subscriptions happen according to the `cron`/fixedDelay`/`fixedRate` configuration.
 
 If the `Publisher` emits `onNext` signal(s), these are ignored and discarded (the same way
 return values from synchronous `@Scheduled` methods are ignored).
@@ -502,8 +498,11 @@ seconds:
 
 [NOTE]
 ====
-The Spring Framework subscribes once to the underlying scheduling `Flux`, which can be
-cancelled by destroying the annotated bean or closing the application context.
+When destroying the annotated bean or closing the application context Spring Framework cancels
+scheduled tasks, which includes the next scheduled subscription to the `Publisher`.
+However, there is no tracking of individual subscriptions: once a subscription has started
+it can't be cancelled. For that reason, it is a pre-requisite that the `Publisher` is finite
+in addition to supporting multiple subsequent subscriptions.
 ====
 
 

spring-context/spring-context.gradle

@@ -38,6 +38,7 @@ dependencies {
 	testImplementation("org.apache.commons:commons-pool2")
 	testImplementation("org.awaitility:awaitility")
 	testImplementation("jakarta.inject:jakarta.inject-tck")
+	testImplementation("org.jetbrains.kotlinx:kotlinx-coroutines-core")
 	testImplementation("org.jetbrains.kotlinx:kotlinx-coroutines-reactor")
 	testImplementation("io.reactivex.rxjava3:rxjava")
 	testRuntimeOnly("jakarta.xml.bind:jakarta.xml.bind-api")

spring-context/src/main/java/org/springframework/scheduling/annotation/EnableScheduling.java

@@ -188,11 +188,6 @@
  * application context and any separate {@code DispatcherServlet} application contexts,
  * if you need to apply its behavior at multiple levels.
  *
- * <p><b>Note: {@code @EnableScheduling} and {@code @Scheduled} support of reactive methods
- * and Kotlin suspending functions uses Reactor infrastructure</b> instead of the
- * {@code ScheduledTaskRegistrar} infrastructure, so previously discussed task-related
- * aspects like {@code SchedulingConfigurer} don't apply to these two cases.
- *
  * @author Chris Beams
  * @author Juergen Hoeller
  * @since 3.1

spring-context/src/main/java/org/springframework/scheduling/annotation/Scheduled.java

@@ -37,17 +37,18 @@
  * when called through the scheduler.
  *
  * <p>Methods that return a reactive {@code Publisher} or a type which can be adapted
- * to {@code Publisher} by the default {@code ReactiveAdapterRegistry} are supported
- * provided the Reactor library is present at runtime. Reactor is used to implement
- * the scheduling by repeatedly subscribing to the returned Publisher, which is only
- * produced once.
- * The cron configuration is not supported for this type of method. Values emitted by
- * the publisher are ignored and discarded. Errors are logged at WARN level, which
- * doesn't prevent further iterations.
+ * to {@code Publisher} by the default {@code ReactiveAdapterRegistry} are supported.
+ * The {@code Publisher} MUST be finite and MUST support multiple subsequent subscriptions
+ * (i.e. be cold).
+ * The returned Publisher is only produced once, and the scheduling infrastructure
+ * then periodically {@code subscribe()} to it according to configuration.
+ * Values emitted by the publisher are ignored. Errors are logged at WARN level, which
+ * doesn't prevent further iterations. If a {@code fixed delay} is configured, the
+ * subscription is blocked upon in order to respect the fixed delay semantics.
  *
- * <p>Kotlin suspending functions are also supported, provided the coroutine-Reactor
+ * <p>Kotlin suspending functions are also supported, provided the coroutine-reactor
  * bridge ({@code kotlinx.coroutine.reactor}) is present at runtime. This bridge is
- * used to adapt the suspending function into a Reactor {@code Mono} which is treated
+ * used to adapt the suspending function into a {@code Publisher} which is treated
  * the same way as in the reactive method case (see above).
  *
  * <p>Processing of {@code @Scheduled} annotations is performed by

spring-context/src/main/java/org/springframework/scheduling/annotation/ScheduledAnnotationBeanPostProcessor.java

@@ -143,7 +143,6 @@ public class ScheduledAnnotationBeanPostProcessor
 	private final Set<Class<?>> nonAnnotatedClasses = Collections.newSetFromMap(new ConcurrentHashMap<>(64));
 
 	private final Map<Object, Set<ScheduledTask>> scheduledTasks = new IdentityHashMap<>(16);
-	private final Map<Object, Set<ScheduledAnnotationReactiveSupport.ReactiveTask>> scheduledReactiveTasks = new IdentityHashMap<>(16);
 
 
 	/**
@@ -318,15 +317,6 @@ private void finishRegistration() {
 		}
 
 		this.registrar.afterPropertiesSet();
-		// Start the reactive tasks (we synchronize on the common scheduledTasks on purpose)
-		synchronized (this.scheduledTasks) {
-			for (Set<ScheduledAnnotationReactiveSupport.ReactiveTask> reactiveTasks : this.scheduledReactiveTasks.values()) {
-				for (ScheduledAnnotationReactiveSupport.ReactiveTask reactiveTask : reactiveTasks) {
-					reactiveTask.subscribe();
-				}
-			}
-		}
-
 	}
 
 	private <T> T resolveSchedulerBean(BeanFactory beanFactory, Class<T> schedulerType, boolean byName) {
@@ -397,37 +387,32 @@ public Object postProcessAfterInitialization(Object bean, String beanName) {
 
 	/**
 	 * Process the given {@code @Scheduled} method declaration on the given bean,
-	 * attempting to distinguish {@link #processScheduledReactive(Scheduled, Method, Object) reactive}
-	 * method from {@link #processScheduledSync(Scheduled, Method, Object) synchronous} methods.
+	 * attempting to distinguish {@link #processScheduledAsync(Scheduled, Method, Object) reactive}
+	 * methods from {@link #processScheduledSync(Scheduled, Method, Object) synchronous} methods.
 	 * @param scheduled the {@code @Scheduled} annotation
 	 * @param method the method that the annotation has been declared on
 	 * @param bean the target bean instance
 	 * @see #processScheduledSync(Scheduled, Method, Object)
-	 * @see #processScheduledReactive(Scheduled, Method, Object)
+	 * @see #processScheduledAsync(Scheduled, Method, Object)
 	 */
 	protected void processScheduled(Scheduled scheduled, Method method, Object bean) {
 		// Is method a Kotlin suspending function? Throws if true but reactor bridge isn't on the classpath.
-		// Is method returning a Publisher instance? Throws if true but Reactor isn't on the classpath.
+		// Is method returning a reactive type? Throws if true, but it isn't a deferred Publisher type.
 		if (ScheduledAnnotationReactiveSupport.isReactive(method)) {
-			processScheduledReactive(scheduled, method, bean);
+			processScheduledAsync(scheduled, method, bean);
 			return;
 		}
 		processScheduledSync(scheduled, method, bean);
 	}
 
 	/**
-	 * Process the given {@code @Scheduled} method declaration on the given bean,
-	 * as a synchronous method. The method MUST take no arguments. Its return value
-	 * is ignored (if any) and the scheduled invocations of the method take place
-	 * using the underlying {@link TaskScheduler} infrastructure.
-	 * @param scheduled the {@code @Scheduled} annotation
-	 * @param method the method that the annotation has been declared on
-	 * @param bean the target bean instance
-	 * @see #createRunnable(Object, Method)
+	 * Parse the {@code Scheduled} annotation and schedule the provided {@code Runnable}
+	 * accordingly. The Runnable can represent either a synchronous method invocation
+	 * (see {@link #processScheduledSync(Scheduled, Method, Object)}) or an asynchronous
+	 * one (see {@link #processScheduledAsync(Scheduled, Method, Object)}).
 	 */
-	protected void processScheduledSync(Scheduled scheduled, Method method, Object bean) {
+	protected void processScheduledTask(Scheduled scheduled, Runnable runnable, Method method, Object bean) {
 		try {
-			Runnable runnable = createRunnable(bean, method);
 			boolean processedSchedule = false;
 			String errorMessage =
 					"Exactly one of the 'cron', 'fixedDelay(String)', or 'fixedRate(String)' attributes is required";
@@ -550,118 +535,51 @@ protected void processScheduledSync(Scheduled scheduled, Method method, Object b
 		}
 	}
 
+	/**
+	 * Process the given {@code @Scheduled} method declaration on the given bean,
+	 * as a synchronous method. The method MUST take no arguments. Its return value
+	 * is ignored (if any) and the scheduled invocations of the method take place
+	 * using the underlying {@link TaskScheduler} infrastructure.
+	 * @param scheduled the {@code @Scheduled} annotation
+	 * @param method the method that the annotation has been declared on
+	 * @param bean the target bean instance
+	 * @see #createRunnable(Object, Method)
+	 */
+	protected void processScheduledSync(Scheduled scheduled, Method method, Object bean) {
+		Runnable task;
+		try {
+			task = createRunnable(bean, method);
+		}
+		catch (IllegalArgumentException ex) {
+			throw new IllegalStateException("Could not create recurring task for @Scheduled method '" + method.getName() + "': " + ex.getMessage());
+		}
+		processScheduledTask(scheduled, task, method, bean);
+	}
+
 	/**
 	 * Process the given {@code @Scheduled} bean method declaration which returns
 	 * a {@code Publisher}, or the given Kotlin suspending function converted to a
-	 * Publisher. The publisher is then repeatedly subscribed to, according to the
-	 * fixedDelay/fixedRate configuration. Cron configuration isn't supported,nor
-	 * is non-Publisher return types (even if a {@code ReactiveAdapter} is registered).
+	 * Publisher. A {@code Runnable} which subscribes to that publisher is then repeatedly
+	 * scheduled according to the annotation configuration.
+	 * <p>Note that for fixed delay configuration, the subscription is turned into a blocking
+	 * call instead. Types for which a {@code ReactiveAdapter} is registered but which cannot
+	 * be deferred (i.e. not  a {@code Publisher}) are not supported.
 	 * @param scheduled the {@code @Scheduled} annotation
 	 * @param method the method that the annotation has been declared on, which
-	 * MUST either return a Publisher or be a Kotlin suspending function
+	 * MUST either return a Publisher-adaptable type or be a Kotlin suspending function
 	 * @param bean the target bean instance
 	 * @see ScheduledAnnotationReactiveSupport
 	 */
-	protected void processScheduledReactive(Scheduled scheduled, Method method, Object bean) {
+	protected void processScheduledAsync(Scheduled scheduled, Method method, Object bean) {
+		Runnable task;
 		try {
-			boolean processedSchedule = false;
-			String errorMessage =
-					"Exactly one of the 'fixedDelay(String)' or 'fixedRate(String)' attributes is required";
-
-			Set<ScheduledAnnotationReactiveSupport.ReactiveTask> reactiveTasks = new LinkedHashSet<>(4);
-
-			// Determine initial delay
-			Duration initialDelay = toDuration(scheduled.initialDelay(), scheduled.timeUnit());
-			String initialDelayString = scheduled.initialDelayString();
-			if (StringUtils.hasText(initialDelayString)) {
-				Assert.isTrue(initialDelay.isNegative(), "Specify 'initialDelay' or 'initialDelayString', not both");
-				if (this.embeddedValueResolver != null) {
-					initialDelayString = this.embeddedValueResolver.resolveStringValue(initialDelayString);
-				}
-				if (StringUtils.hasLength(initialDelayString)) {
-					try {
-						initialDelay = toDuration(initialDelayString, scheduled.timeUnit());
-					}
-					catch (RuntimeException ex) {
-						throw new IllegalArgumentException(
-								"Invalid initialDelayString value \"" + initialDelayString + "\" - cannot parse into long");
-					}
-				}
-			}
-
-			// Reject cron expression
-			Assert.state(!StringUtils.hasText(scheduled.cron()), "'cron' not supported for reactive @Scheduled");
-
-			// At this point we don't need to differentiate between initial delay set or not anymore
-			if (initialDelay.isNegative()) {
-				initialDelay = Duration.ZERO;
-			}
-
-			// Check fixed delay
-			Duration fixedDelay = toDuration(scheduled.fixedDelay(), scheduled.timeUnit());
-			if (!fixedDelay.isNegative()) {
-				processedSchedule = true;
-				reactiveTasks.add(new ScheduledAnnotationReactiveSupport.ReactiveTask(method, bean, initialDelay, fixedDelay, false));
-			}
-
-			String fixedDelayString = scheduled.fixedDelayString();
-			if (StringUtils.hasText(fixedDelayString)) {
-				if (this.embeddedValueResolver != null) {
-					fixedDelayString = this.embeddedValueResolver.resolveStringValue(fixedDelayString);
-				}
-				if (StringUtils.hasLength(fixedDelayString)) {
-					Assert.isTrue(!processedSchedule, errorMessage);
-					processedSchedule = true;
-					try {
-						fixedDelay = toDuration(fixedDelayString, scheduled.timeUnit());
-					}
-					catch (RuntimeException ex) {
-						throw new IllegalArgumentException(
-								"Invalid fixedDelayString value \"" + fixedDelayString + "\" - cannot parse into long");
-					}
-					reactiveTasks.add(new ScheduledAnnotationReactiveSupport.ReactiveTask(method, bean, initialDelay, fixedDelay, false));
-				}
-			}
-
-			// Check fixed rate
-			Duration fixedRate = toDuration(scheduled.fixedRate(), scheduled.timeUnit());
-			if (!fixedRate.isNegative()) {
-				Assert.isTrue(!processedSchedule, errorMessage);
-				processedSchedule = true;
-				reactiveTasks.add(new ScheduledAnnotationReactiveSupport.ReactiveTask(method, bean, initialDelay, fixedRate, true));
-			}
-			String fixedRateString = scheduled.fixedRateString();
-			if (StringUtils.hasText(fixedRateString)) {
-				if (this.embeddedValueResolver != null) {
-					fixedRateString = this.embeddedValueResolver.resolveStringValue(fixedRateString);
-				}
-				if (StringUtils.hasLength(fixedRateString)) {
-					Assert.isTrue(!processedSchedule, errorMessage);
-					processedSchedule = true;
-					try {
-						fixedRate = toDuration(fixedRateString, scheduled.timeUnit());
-					}
-					catch (RuntimeException ex) {
-						throw new IllegalArgumentException(
-								"Invalid fixedRateString value \"" + fixedRateString + "\" - cannot parse into long");
-					}
-					reactiveTasks.add(new ScheduledAnnotationReactiveSupport.ReactiveTask(method, bean, initialDelay, fixedRate, true));
-				}
-			}
-
-			// Check whether we had any attribute set
-			Assert.isTrue(processedSchedule, errorMessage);
-
-			// Finally register the scheduled tasks (we synchronize on scheduledTasks on purpose)
-			synchronized (this.scheduledTasks) {
-				Set<ScheduledAnnotationReactiveSupport.ReactiveTask> subscriptionTasks = this.scheduledReactiveTasks.computeIfAbsent(bean, key -> new LinkedHashSet<>(4));
-				subscriptionTasks.addAll(reactiveTasks);
-			}
+			boolean isFixedDelaySpecialCase = scheduled.fixedDelay() > 0 || StringUtils.hasText(scheduled.fixedDelayString());
+			task = ScheduledAnnotationReactiveSupport.createSubscriptionRunnable(method, bean, isFixedDelaySpecialCase);
 		}
 		catch (IllegalArgumentException ex) {
-			throw new IllegalStateException(
-					"Encountered invalid reactive @Scheduled method '" + method.getName() + "': " + ex.getMessage(), ex);
+			throw new IllegalStateException("Could not create recurring task for @Scheduled method '" + method.getName() + "': " + ex.getMessage());
 		}
+		processScheduledTask(scheduled, task, method, bean);
 	}
 
 	/**
@@ -721,27 +639,20 @@ public Set<ScheduledTask> getScheduledTasks() {
 	@Override
 	public void postProcessBeforeDestruction(Object bean, String beanName) {
 		Set<ScheduledTask> tasks;
-		Set<ScheduledAnnotationReactiveSupport.ReactiveTask> reactiveTasks;
 		synchronized (this.scheduledTasks) {
 			tasks = this.scheduledTasks.remove(bean);
-			reactiveTasks = this.scheduledReactiveTasks.remove(bean);
 		}
 		if (tasks != null) {
 			for (ScheduledTask task : tasks) {
 				task.cancel();
 			}
 		}
-		if (reactiveTasks != null) {
-			for (ScheduledAnnotationReactiveSupport.ReactiveTask task : reactiveTasks) {
-				task.cancel();
-			}
-		}
 	}
 
 	@Override
 	public boolean requiresDestruction(Object bean) {
 		synchronized (this.scheduledTasks) {
-			return this.scheduledTasks.containsKey(bean) || this.scheduledReactiveTasks.containsKey(bean);
+			return this.scheduledTasks.containsKey(bean);
 		}
 	}
 
@@ -755,13 +666,6 @@ public void destroy() {
 				}
 			}
 			this.scheduledTasks.clear();
-			Collection<Set<ScheduledAnnotationReactiveSupport.ReactiveTask>> allReactiveTasks = this.scheduledReactiveTasks.values();
-			for (Set<ScheduledAnnotationReactiveSupport.ReactiveTask> tasks : allReactiveTasks) {
-				for (ScheduledAnnotationReactiveSupport.ReactiveTask task : tasks) {
-					task.cancel();
-				}
-			}
-			this.scheduledReactiveTasks.clear();
 		}
 		this.registrar.destroy();
 	}