Polishing

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

@@ -656,7 +656,7 @@ public class ScheduledAnnotationBeanPostProcessor
 		}
 		if (liveSubscriptions != null) {
 			for (Runnable subscription : liveSubscriptions) {
-				subscription.run(); // equivalent to cancelling the subscription
+				subscription.run();  // equivalent to cancelling the subscription
 			}
 		}
 	}
@@ -664,7 +664,7 @@ public class ScheduledAnnotationBeanPostProcessor
 	@Override
 	public boolean requiresDestruction(Object bean) {
 		synchronized (this.scheduledTasks) {
-			return this.scheduledTasks.containsKey(bean) || this.reactiveSubscriptions.containsKey(bean);
+			return (this.scheduledTasks.containsKey(bean) || this.reactiveSubscriptions.containsKey(bean));
 		}
 	}
 
@@ -681,7 +681,7 @@ public class ScheduledAnnotationBeanPostProcessor
 			Collection<List<Runnable>> allLiveSubscriptions = this.reactiveSubscriptions.values();
 			for (List<Runnable> liveSubscriptions : allLiveSubscriptions) {
 				for (Runnable liveSubscription : liveSubscriptions) {
-					liveSubscription.run(); //equivalent to cancelling the subscription
+					liveSubscription.run();  // equivalent to cancelling the subscription
 				}
 			}
 		}

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

@@ -44,7 +44,7 @@ import org.springframework.util.StringUtils;
  * cases without a dependency on optional classes.
  *
  * @author Simon Baslé
- * @since 6.1.0
+ * @since 6.1
  */
 abstract class ScheduledAnnotationReactiveSupport {
 
@@ -54,7 +54,8 @@ abstract class ScheduledAnnotationReactiveSupport {
 	static final boolean coroutinesReactorPresent = ClassUtils.isPresent(
 			"kotlinx.coroutines.reactor.MonoKt", ScheduledAnnotationReactiveSupport.class.getClassLoader());
 
-	private static final Log LOGGER = LogFactory.getLog(ScheduledAnnotationReactiveSupport.class);
+	private static final Log logger = LogFactory.getLog(ScheduledAnnotationReactiveSupport.class);
+
 
 	/**
 	 * Checks that if the method is reactive, it can be scheduled. Methods are considered
@@ -72,10 +73,10 @@ abstract class ScheduledAnnotationReactiveSupport {
 		if (KotlinDetector.isKotlinPresent() && KotlinDetector.isSuspendingFunction(method)) {
 			// Note that suspending functions declared without args have a single Continuation
 			// parameter in reflective inspection
-			Assert.isTrue(method.getParameterCount() == 1,"Kotlin suspending functions may only be "
+			Assert.isTrue(method.getParameterCount() == 1,
-					+ "annotated with @Scheduled if declared without arguments");
+					"Kotlin suspending functions may only be annotated with @Scheduled if declared without arguments");
-			Assert.isTrue(coroutinesReactorPresent, "Kotlin suspending functions may only be annotated with "
+			Assert.isTrue(coroutinesReactorPresent, "Kotlin suspending functions may only be annotated with " +
-					+ "@Scheduled if the Coroutine-Reactor bridge (kotlinx.coroutines.reactor) is present at runtime");
+					"@Scheduled if the Coroutine-Reactor bridge (kotlinx.coroutines.reactor) is present at runtime");
 			return true;
 		}
 		ReactiveAdapterRegistry registry = ReactiveAdapterRegistry.getSharedInstance();
@@ -87,10 +88,10 @@ abstract class ScheduledAnnotationReactiveSupport {
 		if (candidateAdapter == null) {
 			return false;
 		}
-		Assert.isTrue(method.getParameterCount() == 0, "Reactive methods may only be annotated with "
+		Assert.isTrue(method.getParameterCount() == 0,
-				+ "@Scheduled if declared without arguments");
+				"Reactive methods may only be annotated with @Scheduled if declared without arguments");
-		Assert.isTrue(candidateAdapter.getDescriptor().isDeferred(), "Reactive methods may only be annotated with "
+		Assert.isTrue(candidateAdapter.getDescriptor().isDeferred(),
-				+ "@Scheduled if the return type supports deferred execution");
+				"Reactive methods may only be annotated with @Scheduled if the return type supports deferred execution");
 		return true;
 	}
 
@@ -112,23 +113,23 @@ abstract class ScheduledAnnotationReactiveSupport {
 		Class<?> returnType = method.getReturnType();
 		ReactiveAdapter adapter = registry.getAdapter(returnType);
 		if (adapter == null) {
-			throw new IllegalArgumentException("Cannot convert the @Scheduled reactive method return type to Publisher");
+			throw new IllegalArgumentException("Cannot convert @Scheduled reactive method return type to Publisher");
 		}
 		if (!adapter.getDescriptor().isDeferred()) {
-			throw new IllegalArgumentException("Cannot convert the @Scheduled reactive method return type to Publisher: "
+			throw new IllegalArgumentException("Cannot convert @Scheduled reactive method return type to Publisher: " +
-					+ returnType.getSimpleName() + " is not a deferred reactive type");
+					returnType.getSimpleName() + " is not a deferred reactive type");
 		}
+
 		Method invocableMethod = AopUtils.selectInvocableMethod(method, bean.getClass());
 		try {
 			ReflectionUtils.makeAccessible(invocableMethod);
-			Object r = invocableMethod.invoke(bean);
+			Object returnValue = invocableMethod.invoke(bean);
 
-			Publisher<?> publisher = adapter.toPublisher(r);
+			Publisher<?> publisher = adapter.toPublisher(returnValue);
 			// If Reactor is on the classpath, we could benefit from having a checkpoint for debuggability
 			if (reactorPresent) {
-				final String checkpoint = "@Scheduled '"+ method.getName() + "()' in bean '"
+				return Flux.from(publisher).checkpoint(
-						+ method.getDeclaringClass().getName() + "'";
+						"@Scheduled '"+ method.getName() + "()' in '" + method.getDeclaringClass().getName() + "'");
-				return Flux.from(publisher).checkpoint(checkpoint);
 			}
 			else {
 				return publisher;
@@ -136,12 +137,12 @@ abstract class ScheduledAnnotationReactiveSupport {
 		}
 		catch (InvocationTargetException ex) {
 			throw new IllegalArgumentException(
-				"Cannot obtain a Publisher-convertible value from the @Scheduled reactive method",
+					"Cannot obtain a Publisher-convertible value from the @Scheduled reactive method",
-				ex.getTargetException());
+					ex.getTargetException());
 		}
 		catch (IllegalAccessException ex) {
 			throw new IllegalArgumentException(
-				"Cannot obtain a Publisher-convertible value from the @Scheduled reactive method", ex);
+					"Cannot obtain a Publisher-convertible value from the @Scheduled reactive method", ex);
 		}
 	}
 
@@ -157,20 +158,24 @@ abstract class ScheduledAnnotationReactiveSupport {
 	 */
 	static Runnable createSubscriptionRunnable(Method method, Object targetBean, Scheduled scheduled,
 			List<Runnable> subscriptionTrackerRegistry) {
-		boolean shouldBlock = scheduled.fixedDelay() > 0 || StringUtils.hasText(scheduled.fixedDelayString());
+
-		final Publisher<?> publisher = getPublisherFor(method, targetBean);
+		boolean shouldBlock = (scheduled.fixedDelay() > 0 || StringUtils.hasText(scheduled.fixedDelayString()));
+		Publisher<?> publisher = getPublisherFor(method, targetBean);
 		return new SubscribingRunnable(publisher, shouldBlock, subscriptionTrackerRegistry);
 	}
 
+
 	/**
 	 * 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 {
 
-		final Publisher<?> publisher;
+		private final Publisher<?> publisher;
+
 		final boolean shouldBlock;
-		final List<Runnable> subscriptionTrackerRegistry;
+
+		private final List<Runnable> subscriptionTrackerRegistry;
 
 		SubscribingRunnable(Publisher<?> publisher, boolean shouldBlock, List<Runnable> subscriptionTrackerRegistry) {
 			this.publisher = publisher;
@@ -181,7 +186,7 @@ abstract class ScheduledAnnotationReactiveSupport {
 		@Override
 		public void run() {
 			if (this.shouldBlock) {
-				final CountDownLatch latch = new CountDownLatch(1);
+				CountDownLatch latch = new CountDownLatch(1);
 				TrackingSubscriber subscriber = new TrackingSubscriber(this.subscriptionTrackerRegistry, latch);
 				this.subscriptionTrackerRegistry.add(subscriber);
 				this.publisher.subscribe(subscriber);
@@ -193,13 +198,14 @@ abstract class ScheduledAnnotationReactiveSupport {
 				}
 			}
 			else {
-				final TrackingSubscriber subscriber = new TrackingSubscriber(this.subscriptionTrackerRegistry);
+				TrackingSubscriber subscriber = new TrackingSubscriber(this.subscriptionTrackerRegistry);
 				this.subscriptionTrackerRegistry.add(subscriber);
 				this.publisher.subscribe(subscriber);
 			}
 		}
 	}
 
+
 	/**
 	 * A {@code Subscriber} which keeps track of its {@code Subscription} and exposes the
 	 * capacity to cancel the subscription as a {@code Runnable}. Can optionally support
@@ -246,13 +252,13 @@ abstract class ScheduledAnnotationReactiveSupport {
 
 		@Override
 		public void onNext(Object obj) {
-			// NO-OP
+			// no-op
 		}
 
 		@Override
 		public void onError(Throwable ex) {
 			this.subscriptionTrackerRegistry.remove(this);
-			LOGGER.warn("Unexpected error occurred in scheduled reactive task", ex);
+			logger.warn("Unexpected error occurred in scheduled reactive task", ex);
 			if (this.blockingLatch != null) {
 				this.blockingLatch.countDown();
 			}

spring-context/src/test/java/org/springframework/scheduling/annotation/ScheduledAnnotationReactiveSupportTests.java

@@ -41,6 +41,10 @@ import static org.springframework.scheduling.annotation.ScheduledAnnotationReact
 import static org.springframework.scheduling.annotation.ScheduledAnnotationReactiveSupport.getPublisherFor;
 import static org.springframework.scheduling.annotation.ScheduledAnnotationReactiveSupport.isReactive;
 
+/**
+ * @author Simon Baslé
+ * @since 6.1
+ */
 class ScheduledAnnotationReactiveSupportTests {
 
 	@Test
@@ -76,7 +80,7 @@ class ScheduledAnnotationReactiveSupportTests {
 	void isReactiveRejectsWithParams() {
 		Method m = ReflectionUtils.findMethod(ReactiveMethods.class, "monoWithParam", String.class);
 
-		//isReactive rejects with context
+		// isReactive rejects with context
 		assertThatIllegalArgumentException().isThrownBy(() -> isReactive(m))
 				.withMessage("Reactive methods may only be annotated with @Scheduled if declared without arguments")
 				.withNoCause();
@@ -87,7 +91,7 @@ class ScheduledAnnotationReactiveSupportTests {
 		ReactiveMethods target = new ReactiveMethods();
 		Method m = ReflectionUtils.findMethod(ReactiveMethods.class, "monoThrows");
 
-		//static helper method
+		// static helper method
 		assertThatIllegalArgumentException().isThrownBy(() -> getPublisherFor(m, target))
 				.withMessage("Cannot obtain a Publisher-convertible value from the @Scheduled reactive method")
 				.withCause(new IllegalStateException("expected"));
@@ -98,7 +102,7 @@ class ScheduledAnnotationReactiveSupportTests {
 		ReactiveMethods target = new ReactiveMethods();
 		Method m = ReflectionUtils.findMethod(ReactiveMethods.class, "monoThrowsIllegalAccess");
 
-		//static helper method
+		// static helper method
 		assertThatIllegalArgumentException().isThrownBy(() -> getPublisherFor(m, target))
 				.withMessage("Cannot obtain a Publisher-convertible value from the @Scheduled reactive method")
 				.withCause(new IllegalAccessException("expected"));
@@ -165,9 +169,10 @@ class ScheduledAnnotationReactiveSupportTests {
 				.as("checkpoint class")
 				.isEqualTo("reactor.core.publisher.FluxOnAssembly");
 
-		assertThat(p).hasToString("checkpoint(\"@Scheduled 'mono()' in bean 'org.springframework.scheduling.annotation.ScheduledAnnotationReactiveSupportTests$ReactiveMethods'\")");
+		assertThat(p).hasToString("checkpoint(\"@Scheduled 'mono()' in 'org.springframework.scheduling.annotation.ScheduledAnnotationReactiveSupportTests$ReactiveMethods'\")");
 	}
 
+
 	static class ReactiveMethods {
 
 		public String oops() {
@@ -211,7 +216,7 @@ class ScheduledAnnotationReactiveSupportTests {
 		}
 
 		public Mono<Void> monoThrowsIllegalAccess() throws IllegalAccessException {
-			//simulate a reflection issue
+			// simulate a reflection issue
 			throw new IllegalAccessException("expected");
 		}
 
@@ -226,14 +231,12 @@ class ScheduledAnnotationReactiveSupportTests {
 		AtomicInteger subscription = new AtomicInteger();
 
 		public Mono<Void> trackingMono() {
-			return Mono.<Void>empty()
+			return Mono.<Void>empty().doOnSubscribe(s -> subscription.incrementAndGet());
-					.doOnSubscribe(s -> subscription.incrementAndGet());
 		}
 
 		public Mono<Void> monoError() {
 			return Mono.error(new IllegalStateException("expected"));
 		}
-
 	}
 
 }

spring-context/src/test/kotlin/org/springframework/scheduling/annotation/KotlinScheduledAnnotationReactiveSupportTests.kt

@@ -92,7 +92,7 @@ class KotlinScheduledAnnotationReactiveSupportTests {
 
 		//static helper method
 		Assertions.assertThatIllegalArgumentException().isThrownBy { getPublisherFor(m!!, target!!) }
-				.withMessage("Cannot convert the @Scheduled reactive method return type to Publisher")
+				.withMessage("Cannot convert @Scheduled reactive method return type to Publisher")
 				.withNoCause()
 	}