mirror of
https://github.com/spring-projects/spring-framework.git
synced 2026-09-17 08:24:13 +00:00
Merge branch '7.0.x'
This commit is contained in:
+22
-18
@@ -317,24 +317,24 @@ public class ScheduledAnnotationBeanPostProcessor
|
||||
}
|
||||
|
||||
/**
|
||||
* Process the given {@code @Scheduled} method declaration on the given bean,
|
||||
* attempting to distinguish {@linkplain #processScheduledAsync(Scheduled, Method, Object)
|
||||
* reactive} methods from {@linkplain #processScheduledSync(Scheduled, Method, Object)
|
||||
* synchronous} methods.
|
||||
* Process the given {@code @Scheduled} method declaration on the given bean.
|
||||
* @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 #processScheduledAsync(Scheduled, Method, Object)
|
||||
*/
|
||||
protected void processScheduled(Scheduled scheduled, Method method, Object bean) {
|
||||
Object key = AopProxyUtils.getSingletonTarget(bean);
|
||||
if (key == null) {
|
||||
key = bean;
|
||||
}
|
||||
|
||||
// Is the method a Kotlin suspending function? Throws if true and the reactor bridge isn't on the classpath.
|
||||
// Does the method return a reactive type? Throws if true and it isn't a deferred Publisher type.
|
||||
if (REACTIVE_STREAMS_PRESENT && ScheduledAnnotationReactiveSupport.isReactive(method)) {
|
||||
processScheduledAsync(scheduled, method, bean);
|
||||
processScheduledAsync(scheduled, method, bean, key);
|
||||
return;
|
||||
}
|
||||
processScheduledSync(scheduled, method, bean);
|
||||
processScheduledSync(scheduled, method, bean, key);
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -345,8 +345,9 @@ public class ScheduledAnnotationBeanPostProcessor
|
||||
* @param scheduled the {@code @Scheduled} annotation
|
||||
* @param method the method that the annotation has been declared on
|
||||
* @param bean the target bean instance
|
||||
* @param key a cache key for the target bean instance
|
||||
*/
|
||||
private void processScheduledSync(Scheduled scheduled, Method method, Object bean) {
|
||||
private void processScheduledSync(Scheduled scheduled, Method method, Object bean, Object key) {
|
||||
Runnable task;
|
||||
try {
|
||||
task = createRunnable(bean, method, scheduled.scheduler());
|
||||
@@ -355,7 +356,8 @@ public class ScheduledAnnotationBeanPostProcessor
|
||||
throw new IllegalStateException("Could not create recurring task for @Scheduled method '" +
|
||||
method.getName() + "': " + ex.getMessage());
|
||||
}
|
||||
processScheduledTask(scheduled, task, method, bean);
|
||||
|
||||
processScheduledTask(scheduled, task, method, key);
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -370,33 +372,35 @@ public class ScheduledAnnotationBeanPostProcessor
|
||||
* @param method the method that the annotation has been declared on, which
|
||||
* must either return a Publisher-adaptable type or be a Kotlin suspending function
|
||||
* @param bean the target bean instance
|
||||
* @param key a cache key for the target bean (potentially the raw singleton target)
|
||||
* @see ScheduledAnnotationReactiveSupport
|
||||
*/
|
||||
private void processScheduledAsync(Scheduled scheduled, Method method, Object bean) {
|
||||
private void processScheduledAsync(Scheduled scheduled, Method method, Object bean, Object key) {
|
||||
Runnable task;
|
||||
try {
|
||||
task = ScheduledAnnotationReactiveSupport.createSubscriptionRunnable(method, bean, scheduled,
|
||||
this.registrar::getObservationRegistry,
|
||||
this.reactiveSubscriptions.computeIfAbsent(bean, key -> new CopyOnWriteArrayList<>()));
|
||||
this.reactiveSubscriptions.computeIfAbsent(key, k -> new CopyOnWriteArrayList<>()));
|
||||
}
|
||||
catch (IllegalArgumentException ex) {
|
||||
throw new IllegalStateException("Could not create recurring task for @Scheduled method '" +
|
||||
method.getName() + "': " + ex.getMessage());
|
||||
}
|
||||
processScheduledTask(scheduled, task, method, bean);
|
||||
|
||||
processScheduledTask(scheduled, task, method, key);
|
||||
}
|
||||
|
||||
/**
|
||||
* 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)}).
|
||||
* (see {@link #processScheduledSync}) or an asynchronous one (see
|
||||
* {@link #processScheduledAsync}).
|
||||
* @param scheduled the {@code @Scheduled} annotation
|
||||
* @param runnable the runnable to be scheduled
|
||||
* @param method the method that the annotation has been declared on
|
||||
* @param bean the target bean instance
|
||||
* @param key a cache key for the target bean (potentially the raw singleton target)
|
||||
*/
|
||||
private void processScheduledTask(Scheduled scheduled, Runnable runnable, Method method, Object bean) {
|
||||
private void processScheduledTask(Scheduled scheduled, Runnable runnable, Method method, Object key) {
|
||||
try {
|
||||
boolean processedSchedule = false;
|
||||
String errorMessage = "Exactly one of the 'cron', 'fixedDelay' or 'fixedRate' attributes is required";
|
||||
@@ -510,7 +514,7 @@ public class ScheduledAnnotationBeanPostProcessor
|
||||
|
||||
// Finally register the scheduled tasks
|
||||
synchronized (this.scheduledTasks) {
|
||||
Set<ScheduledTask> regTasks = this.scheduledTasks.computeIfAbsent(bean, key -> new LinkedHashSet<>(4));
|
||||
Set<ScheduledTask> regTasks = this.scheduledTasks.computeIfAbsent(key, k -> new LinkedHashSet<>(4));
|
||||
regTasks.addAll(tasks);
|
||||
}
|
||||
}
|
||||
|
||||
+4
-4
@@ -117,14 +117,14 @@ abstract class ScheduledAnnotationReactiveSupport {
|
||||
* semantics (i.e. the task blocks until completion of the Publisher, and the
|
||||
* delay is applied until the next iteration).
|
||||
*/
|
||||
public static Runnable createSubscriptionRunnable(Method method, Object targetBean, Scheduled scheduled,
|
||||
public static Runnable createSubscriptionRunnable(Method method, Object bean, Scheduled scheduled,
|
||||
Supplier<ObservationRegistry> observationRegistrySupplier, List<Runnable> subscriptionTrackerRegistry) {
|
||||
|
||||
boolean shouldBlock = (scheduled.fixedDelay() > 0 || StringUtils.hasText(scheduled.fixedDelayString()));
|
||||
Publisher<?> publisher = getPublisherFor(method, targetBean);
|
||||
Publisher<?> publisher = getPublisherFor(method, bean);
|
||||
Supplier<ScheduledTaskObservationContext> contextSupplier =
|
||||
() -> new ScheduledTaskObservationContext(targetBean, method);
|
||||
String displayName = targetBean.getClass().getName() + "." + method.getName();
|
||||
() -> new ScheduledTaskObservationContext(bean, method);
|
||||
String displayName = bean.getClass().getName() + "." + method.getName();
|
||||
return new SubscribingRunnable(publisher, shouldBlock, scheduled.scheduler(),
|
||||
subscriptionTrackerRegistry, displayName, observationRegistrySupplier, contextSupplier);
|
||||
}
|
||||
|
||||
+40
@@ -42,6 +42,8 @@ import org.springframework.context.annotation.Scope;
|
||||
import org.springframework.context.support.PropertySourcesPlaceholderConfigurer;
|
||||
import org.springframework.core.task.TaskExecutor;
|
||||
import org.springframework.core.testfixture.EnabledForTestGroups;
|
||||
import org.springframework.resilience.annotation.ConcurrencyLimit;
|
||||
import org.springframework.resilience.annotation.EnableResilientMethods;
|
||||
import org.springframework.scheduling.TaskScheduler;
|
||||
import org.springframework.scheduling.concurrent.SimpleAsyncTaskScheduler;
|
||||
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
|
||||
@@ -320,6 +322,20 @@ class EnableSchedulingTests {
|
||||
assertThat(ctx.getBean(AtomicInteger.class).get()).isGreaterThan(1);
|
||||
}
|
||||
|
||||
@Test
|
||||
void withPlainTaskToDestroy() {
|
||||
ctx = new AnnotationConfigApplicationContext(PlainTaskConfig.class);
|
||||
ctx.getDefaultListableBeanFactory().destroySingleton("config");
|
||||
assertThat(ctx.getBean(ScheduledAnnotationBeanPostProcessor.class).getScheduledTasks().isEmpty()).isTrue();
|
||||
}
|
||||
|
||||
@Test
|
||||
void withProxiedTaskToDestroy() {
|
||||
ctx = new AnnotationConfigApplicationContext(ProxiedTaskConfig.class);
|
||||
ctx.getDefaultListableBeanFactory().destroySingleton("config");
|
||||
assertThat(ctx.getBean(ScheduledAnnotationBeanPostProcessor.class).getScheduledTasks().isEmpty()).isTrue();
|
||||
}
|
||||
|
||||
|
||||
@Configuration
|
||||
@EnableScheduling
|
||||
@@ -891,4 +907,28 @@ class EnableSchedulingTests {
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@Configuration("config")
|
||||
@EnableScheduling
|
||||
static class PlainTaskConfig {
|
||||
|
||||
@Scheduled(fixedDelay = 100)
|
||||
public void task() throws InterruptedException {
|
||||
Thread.sleep(100);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@Configuration("config")
|
||||
@EnableScheduling
|
||||
@EnableResilientMethods(proxyTargetClass = true)
|
||||
static class ProxiedTaskConfig {
|
||||
|
||||
@Scheduled(fixedDelay = 100)
|
||||
@ConcurrencyLimit(1)
|
||||
public void task() throws InterruptedException {
|
||||
Thread.sleep(100);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
+4
-4
@@ -43,7 +43,7 @@ class ClassFileTransformerAdapter implements ClassFileTransformer {
|
||||
|
||||
private final ClassTransformer classTransformer;
|
||||
|
||||
private boolean currentlyTransforming = false;
|
||||
private final ThreadLocal<@Nullable Boolean> currentlyTransforming = new ThreadLocal<>();
|
||||
|
||||
|
||||
public ClassFileTransformerAdapter(ClassTransformer classTransformer) {
|
||||
@@ -58,13 +58,13 @@ class ClassFileTransformerAdapter implements ClassFileTransformer {
|
||||
ProtectionDomain protectionDomain, byte[] classfileBuffer) {
|
||||
|
||||
synchronized (this) {
|
||||
if (this.currentlyTransforming) {
|
||||
if (this.currentlyTransforming.get() == Boolean.TRUE) {
|
||||
// Defensively back out when called from within the transform delegate below:
|
||||
// in particular, for the over-eager transformer implementation in Hibernate.
|
||||
return null;
|
||||
}
|
||||
|
||||
this.currentlyTransforming = true;
|
||||
this.currentlyTransforming.set(Boolean.TRUE);
|
||||
try {
|
||||
byte[] transformed = this.classTransformer.transform(
|
||||
loader, className, classBeingRedefined, protectionDomain, classfileBuffer);
|
||||
@@ -91,7 +91,7 @@ class ClassFileTransformerAdapter implements ClassFileTransformer {
|
||||
throw new IllegalStateException("Could not weave class [" + className + "]", ex);
|
||||
}
|
||||
finally {
|
||||
this.currentlyTransforming = false;
|
||||
this.currentlyTransforming.remove();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user