18154768944

Committed 01 Oct 2025 07:26AM UTC coverage: 81.882% (-0.005%) from 81.887%

Build # 18154768944

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #290 from TAKETODAY/dev/jspecify

jspecify

Coverage Stats

59788 of 78013 branches covered (76.64%)

Branch coverage included in aggregate %.

141239 of 167496 relevant lines covered (84.32%)

3.6 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

65.35

today-context/src/main/java/infra/scheduling/concurrent/SimpleAsyncTaskScheduler.java

/*
 * Copyright 2017 - 2025 the original author or authors.
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program. If not, see [https://www.gnu.org/licenses/]
 */

package infra.scheduling.concurrent;

import org.jspecify.annotations.Nullable;

import java.time.Clock;
import java.time.Duration;
import java.time.Instant;
import java.util.concurrent.Callable;
import java.util.concurrent.Executor;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import infra.context.ApplicationContext;
import infra.context.ApplicationContextAware;
import infra.context.ApplicationListener;
import infra.context.SmartLifecycle;
import infra.context.event.ContextClosedEvent;
import infra.core.task.SimpleAsyncTaskExecutor;
import infra.core.task.TaskExecutor;
import infra.core.task.TaskRejectedException;
import infra.lang.Assert;
import infra.logging.LoggerFactory;
import infra.scheduling.TaskScheduler;
import infra.scheduling.Trigger;
import infra.scheduling.support.DelegatingErrorHandlingRunnable;
import infra.scheduling.support.TaskUtils;
import infra.util.ErrorHandler;
import infra.util.concurrent.Future;

/**
 * A simple implementation of Infra {@link TaskScheduler} interface, using
 * a single scheduler thread and executing every scheduled task in an individual
 * separate thread. This is an attractive choice with virtual threads on JDK 21,
 * expecting common usage with {@link #setVirtualThreads setVirtualThreads(true)}.
 *
 * <p><b>NOTE: Scheduling with a fixed delay enforces execution on the single
 * scheduler thread, in order to provide traditional fixed-delay semantics!</b>
 * Prefer the use of fixed rates or cron triggers instead which are a better fit
 * with this thread-per-task scheduler variant.
 *
 * <p>Supports a graceful shutdown through {@link #setTaskTerminationTimeout},
 * at the expense of task tracking overhead per execution thread at runtime.
 * Supports limiting concurrent threads through {@link #setConcurrencyLimit}.
 * By default, the number of concurrent task executions is unlimited.
 * This allows for dynamic concurrency of scheduled task executions, in contrast
 * to {@link ThreadPoolTaskScheduler} which requires a fixed pool size.
 *
 * <p><b>NOTE: This implementation does not reuse threads!</b> Consider a
 * thread-pooling TaskScheduler implementation instead, in particular for
 * scheduling a large number of short-lived tasks. Alternatively, on JDK 21,
 * consider setting {@link #setVirtualThreads} to {@code true}.
 *
 * <p>Extends {@link SimpleAsyncTaskExecutor} and can serve as a fully capable
 * replacement for it, e.g. as a single shared instance serving as a
 * {@link TaskExecutor} as well as a {@link TaskScheduler}.
 * This is generally not the case with other executor/scheduler implementations
 * which tend to have specific constraints for the scheduler thread pool,
 * requiring a separate thread pool for general executor purposes in practice.
 *
 * <p><b>NOTE: This scheduler variant does not track the actual completion of tasks
 * but rather just the hand-off to an execution thread.</b> As a consequence,
 * a {@link ScheduledFuture} handle (for example, from {@link #schedule(Runnable, Instant)})
 * represents that hand-off rather than the actual completion of the provided task
 * (or series of repeated tasks). Also, this scheduler participates in lifecycle
 * management to a limited degree only, stopping trigger firing and fixed-delay
 * task execution but not stopping the execution of handed-off tasks.
 *
 * <p>As an alternative to the built-in thread-per-task capability, this scheduler
 * can also be configured with a separate target executor for scheduled task
 * execution through {@link #setTargetTaskExecutor}: e.g. pointing to a shared
 * {@link ThreadPoolTaskExecutor} bean. This is still rather different from a
 * {@link ThreadPoolTaskScheduler} setup since it always uses a single scheduler
 * thread while dynamically dispatching to the target thread pool which may have
 * a dynamic core/max pool size range, participating in a shared concurrency limit.
 *
 * @author Juergen Hoeller
 * @author <a href="https://github.com/TAKETODAY">Harry Yang</a>
 * @see #setVirtualThreads
 * @see #setTaskTerminationTimeout
 * @see #setConcurrencyLimit
 * @see SimpleAsyncTaskExecutor
 * @see ThreadPoolTaskScheduler
 * @since 4.0
 */
@SuppressWarnings("serial")
public class SimpleAsyncTaskScheduler extends SimpleAsyncTaskExecutor implements TaskScheduler,
        ApplicationContextAware, SmartLifecycle, ApplicationListener<ContextClosedEvent> {

  /**
   * The default phase for an executor {@link SmartLifecycle}: {@code Integer.MAX_VALUE / 2}.
   *
   * @see #getPhase()
   * @see ExecutorConfigurationSupport#DEFAULT_PHASE
   */
  public static final int DEFAULT_PHASE = ExecutorConfigurationSupport.DEFAULT_PHASE;

  private static final TimeUnit NANO = TimeUnit.NANOSECONDS;

  private final ScheduledExecutorService triggerExecutor = createScheduledExecutor();

  private final ExecutorLifecycleDelegate triggerLifecycle = new ExecutorLifecycleDelegate(this.triggerExecutor);

  private final ScheduledExecutorService fixedDelayExecutor = createFixedDelayExecutor();

  private final ExecutorLifecycleDelegate fixedDelayLifecycle = new ExecutorLifecycleDelegate(this.fixedDelayExecutor);

  @Nullable
  private ErrorHandler errorHandler;

  private Clock clock = Clock.systemDefaultZone();

  private int phase = DEFAULT_PHASE;

  @Nullable
  private Executor targetTaskExecutor;

  @Nullable
  private ApplicationContext applicationContext;

  /**
   * Provide an {@link ErrorHandler} strategy.
   */
  public void setErrorHandler(ErrorHandler errorHandler) {
    Assert.notNull(errorHandler, "ErrorHandler is required");
    this.errorHandler = errorHandler;
  }

  /**
   * Set the clock to use for scheduling purposes.
   * <p>The default clock is the system clock for the default time zone.
   *
   * @see Clock#systemDefaultZone()
   */
  public void setClock(Clock clock) {
    Assert.notNull(clock, "Clock is required");
    this.clock = clock;
  }

  @Override
  public Clock getClock() {
    return this.clock;
  }

  /**
   * Specify the lifecycle phase for pausing and resuming this executor.
   * The default is {@link #DEFAULT_PHASE}.
   *
   * @see SmartLifecycle#getPhase()
   */
  public void setPhase(int phase) {
    this.phase = phase;
  }

  /**
   * Return the lifecycle phase for pausing and resuming this executor.
   *
   * @see #setPhase
   */
  @Override
  public int getPhase() {
    return this.phase;
  }

  /**
   * Specify a custom target {@link Executor} to delegate to for
   * the individual execution of scheduled tasks. This can for example
   * be set to a separate thread pool for executing scheduled tasks,
   * whereas this scheduler keeps using its single scheduler thread.
   * <p>If not set, the regular {@link SimpleAsyncTaskExecutor}
   * arrangements kicks in with a new thread per task.
   */
  public void setTargetTaskExecutor(Executor targetTaskExecutor) {
    this.targetTaskExecutor = (targetTaskExecutor == this ? null : targetTaskExecutor);
  }

  @Override
  public void setApplicationContext(ApplicationContext applicationContext) {
    this.applicationContext = applicationContext;
  }

  private ScheduledExecutorService createScheduledExecutor() {
    return new ScheduledThreadPoolExecutor(1, this::newThread) {
      @Override
      protected void beforeExecute(Thread thread, Runnable task) {
        triggerLifecycle.beforeExecute(thread);
      }

      @Override
      protected void afterExecute(Runnable task, Throwable ex) {
        triggerLifecycle.afterExecute();
      }
    };
  }

  private ScheduledExecutorService createFixedDelayExecutor() {
    return new ScheduledThreadPoolExecutor(1, this::newThread) {
      @Override
      protected void beforeExecute(Thread thread, Runnable task) {
        fixedDelayLifecycle.beforeExecute(thread);
      }

      @Override
      protected void afterExecute(Runnable task, Throwable ex) {
        fixedDelayLifecycle.afterExecute();
      }
    };
  }

  @Override
  protected void doExecute(Runnable task) {
    if (this.targetTaskExecutor != null) {
      this.targetTaskExecutor.execute(task);
    }
    else {
      super.doExecute(task);
    }
  }

  private Runnable taskOnSchedulerThread(Runnable task) {
    return new DelegatingErrorHandlingRunnable(task,
            (this.errorHandler != null ? this.errorHandler : TaskUtils.getDefaultErrorHandler(true)));
  }

  private Runnable scheduledTask(Runnable task) {
    return () -> execute(new DelegatingErrorHandlingRunnable(task, this::shutdownAwareErrorHandler));
  }

  private void shutdownAwareErrorHandler(Throwable ex) {
    if (this.errorHandler != null) {
      this.errorHandler.handleError(ex);
    }
    else if (this.triggerExecutor.isShutdown()) {
      LoggerFactory.getLogger(getClass()).debug("Ignoring scheduled task exception after shutdown", ex);
    }
    else {
      TaskUtils.getDefaultErrorHandler(true).handleError(ex);
    }
  }

  @Override
  public void execute(Runnable task) {
    super.execute(TaskUtils.decorateTaskWithErrorHandler(task, this.errorHandler, false));
  }

  @Override
  public Future<Void> submit(Runnable task) {
    return super.submit(TaskUtils.decorateTaskWithErrorHandler(task, this.errorHandler, false));
  }

  @Override
  public <T> Future<T> submit(Callable<T> task) {
    return super.submit(new DelegatingErrorHandlingCallable<>(task, this.errorHandler));
  }

  @Override
  @Nullable
  public ScheduledFuture<?> schedule(Runnable task, Trigger trigger) {
    try {
      Runnable delegate = scheduledTask(task);
      ErrorHandler errorHandler =
              (this.errorHandler != null ? this.errorHandler : TaskUtils.getDefaultErrorHandler(true));
      return new ReschedulingRunnable(
              delegate, trigger, this.clock, this.triggerExecutor, errorHandler).schedule();
    }
    catch (RejectedExecutionException ex) {
      throw new TaskRejectedException(this.triggerExecutor, task, ex);
    }
  }

  @Override
  public ScheduledFuture<?> schedule(Runnable task, Instant startTime) {
    Duration delay = Duration.between(this.clock.instant(), startTime);
    try {
      return this.triggerExecutor.schedule(scheduledTask(task), NANO.convert(delay), NANO);
    }
    catch (RejectedExecutionException ex) {
      throw new TaskRejectedException(this.triggerExecutor, task, ex);
    }
  }

  @Override
  public ScheduledFuture<?> scheduleAtFixedRate(Runnable task, Instant startTime, Duration period) {
    Duration initialDelay = Duration.between(this.clock.instant(), startTime);
    try {
      return this.triggerExecutor.scheduleAtFixedRate(scheduledTask(task),
              NANO.convert(initialDelay), NANO.convert(period), NANO);
    }
    catch (RejectedExecutionException ex) {
      throw new TaskRejectedException(this.triggerExecutor, task, ex);
    }
  }

  @Override
  public ScheduledFuture<?> scheduleAtFixedRate(Runnable task, Duration period) {
    try {
      return this.triggerExecutor.scheduleAtFixedRate(scheduledTask(task),
              0, NANO.convert(period), NANO);
    }
    catch (RejectedExecutionException ex) {
      throw new TaskRejectedException(this.triggerExecutor, task, ex);
    }
  }

  @Override
  public ScheduledFuture<?> scheduleWithFixedDelay(Runnable task, Instant startTime, Duration delay) {
    Duration initialDelay = Duration.between(this.clock.instant(), startTime);
    try {
      // Blocking task on scheduler thread for fixed delay semantics
      return this.fixedDelayExecutor.scheduleWithFixedDelay(taskOnSchedulerThread(task),
              NANO.convert(initialDelay), NANO.convert(delay), NANO);
    }
    catch (RejectedExecutionException ex) {
      throw new TaskRejectedException(this.fixedDelayExecutor, task, ex);
    }
  }

  @Override
  public ScheduledFuture<?> scheduleWithFixedDelay(Runnable task, Duration delay) {
    try {
      // Blocking task on scheduler thread for fixed delay semantics
      return this.fixedDelayExecutor.scheduleWithFixedDelay(taskOnSchedulerThread(task),
              0, NANO.convert(delay), NANO);
    }
    catch (RejectedExecutionException ex) {
      throw new TaskRejectedException(this.fixedDelayExecutor, task, ex);
    }
  }

  @Override
  public void start() {
    this.triggerLifecycle.start();
    this.fixedDelayLifecycle.start();
  }

  @Override
  public void stop() {
    this.triggerLifecycle.stop();
    this.fixedDelayLifecycle.stop();
  }

  @Override
  public void stop(Runnable callback) {
    this.triggerLifecycle.stop();  // no callback necessary since it's just triggers with hand-offs
    this.fixedDelayLifecycle.stop(callback);  // callback for currently executing fixed-delay tasks
  }

  @Override
  public boolean isRunning() {
    return triggerLifecycle.isRunning() || fixedDelayLifecycle.isRunning();
  }

  @Override
  public void onApplicationEvent(ContextClosedEvent event) {
    if (event.getApplicationContext() == this.applicationContext) {
      this.triggerExecutor.shutdown();
      this.fixedDelayExecutor.shutdown();
    }
  }

  @Override
  public void close() {
    for (Runnable remainingTask : this.triggerExecutor.shutdownNow()) {
      if (remainingTask instanceof java.util.concurrent.Future<?> future) {
        future.cancel(true);
      }
    }
    for (Runnable remainingTask : this.fixedDelayExecutor.shutdownNow()) {
      if (remainingTask instanceof Future<?> future) {
        future.cancel(true);
      }
    }
    super.close();
  }

}

1	/*
2	* Copyright 2017 - 2025 the original author or authors.
3	*
4	* This program is free software: you can redistribute it and/or modify
5	* it under the terms of the GNU General Public License as published by
6	* the Free Software Foundation, either version 3 of the License, or
7	* (at your option) any later version.
8	*
9	* This program is distributed in the hope that it will be useful,
10	* but WITHOUT ANY WARRANTY; without even the implied warranty of
11	* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12	* GNU General Public License for more details.
13	*
14	* You should have received a copy of the GNU General Public License
15	* along with this program. If not, see [https://www.gnu.org/licenses/]
16	*/
17
18	package infra.scheduling.concurrent;
19
20	import org.jspecify.annotations.Nullable;
21
22	import java.time.Clock;
23	import java.time.Duration;
24	import java.time.Instant;
25	import java.util.concurrent.Callable;
26	import java.util.concurrent.Executor;
27	import java.util.concurrent.RejectedExecutionException;
28	import java.util.concurrent.ScheduledExecutorService;
29	import java.util.concurrent.ScheduledFuture;
30	import java.util.concurrent.ScheduledThreadPoolExecutor;
31	import java.util.concurrent.TimeUnit;
32
33	import infra.context.ApplicationContext;
34	import infra.context.ApplicationContextAware;
35	import infra.context.ApplicationListener;
36	import infra.context.SmartLifecycle;
37	import infra.context.event.ContextClosedEvent;
38	import infra.core.task.SimpleAsyncTaskExecutor;
39	import infra.core.task.TaskExecutor;
40	import infra.core.task.TaskRejectedException;
41	import infra.lang.Assert;
42	import infra.logging.LoggerFactory;
43	import infra.scheduling.TaskScheduler;
44	import infra.scheduling.Trigger;
45	import infra.scheduling.support.DelegatingErrorHandlingRunnable;
46	import infra.scheduling.support.TaskUtils;
47	import infra.util.ErrorHandler;
48	import infra.util.concurrent.Future;
49
50	/**
51	* A simple implementation of Infra {@link TaskScheduler} interface, using
52	* a single scheduler thread and executing every scheduled task in an individual
53	* separate thread. This is an attractive choice with virtual threads on JDK 21,
54	* expecting common usage with {@link #setVirtualThreads setVirtualThreads(true)}.
55	*
56	* <p><b>NOTE: Scheduling with a fixed delay enforces execution on the single
57	* scheduler thread, in order to provide traditional fixed-delay semantics!</b>
58	* Prefer the use of fixed rates or cron triggers instead which are a better fit
59	* with this thread-per-task scheduler variant.
60	*
61	* <p>Supports a graceful shutdown through {@link #setTaskTerminationTimeout},
62	* at the expense of task tracking overhead per execution thread at runtime.
63	* Supports limiting concurrent threads through {@link #setConcurrencyLimit}.
64	* By default, the number of concurrent task executions is unlimited.
65	* This allows for dynamic concurrency of scheduled task executions, in contrast
66	* to {@link ThreadPoolTaskScheduler} which requires a fixed pool size.
67	*
68	* <p><b>NOTE: This implementation does not reuse threads!</b> Consider a
69	* thread-pooling TaskScheduler implementation instead, in particular for
70	* scheduling a large number of short-lived tasks. Alternatively, on JDK 21,
71	* consider setting {@link #setVirtualThreads} to {@code true}.
72	*
73	* <p>Extends {@link SimpleAsyncTaskExecutor} and can serve as a fully capable
74	* replacement for it, e.g. as a single shared instance serving as a
75	* {@link TaskExecutor} as well as a {@link TaskScheduler}.
76	* This is generally not the case with other executor/scheduler implementations
77	* which tend to have specific constraints for the scheduler thread pool,
78	* requiring a separate thread pool for general executor purposes in practice.
79	*
80	* <p><b>NOTE: This scheduler variant does not track the actual completion of tasks
81	* but rather just the hand-off to an execution thread.</b> As a consequence,
82	* a {@link ScheduledFuture} handle (for example, from {@link #schedule(Runnable, Instant)})
83	* represents that hand-off rather than the actual completion of the provided task
84	* (or series of repeated tasks). Also, this scheduler participates in lifecycle
85	* management to a limited degree only, stopping trigger firing and fixed-delay
86	* task execution but not stopping the execution of handed-off tasks.
87	*
88	* <p>As an alternative to the built-in thread-per-task capability, this scheduler
89	* can also be configured with a separate target executor for scheduled task
90	* execution through {@link #setTargetTaskExecutor}: e.g. pointing to a shared
91	* {@link ThreadPoolTaskExecutor} bean. This is still rather different from a
92	* {@link ThreadPoolTaskScheduler} setup since it always uses a single scheduler
93	* thread while dynamically dispatching to the target thread pool which may have
94	* a dynamic core/max pool size range, participating in a shared concurrency limit.
95	*
96	* @author Juergen Hoeller
97	* @author <a href="https://github.com/TAKETODAY">Harry Yang</a>
98	* @see #setVirtualThreads
99	* @see #setTaskTerminationTimeout
100	* @see #setConcurrencyLimit
101	* @see SimpleAsyncTaskExecutor
102	* @see ThreadPoolTaskScheduler
103	* @since 4.0
104	*/
105	@SuppressWarnings("serial")
106	public class SimpleAsyncTaskScheduler extends SimpleAsyncTaskExecutor implements TaskScheduler,	2✔
107	ApplicationContextAware, SmartLifecycle, ApplicationListener<ContextClosedEvent> {
108
109	/**
110	* The default phase for an executor {@link SmartLifecycle}: {@code Integer.MAX_VALUE / 2}.
111	*
112	* @see #getPhase()
113	* @see ExecutorConfigurationSupport#DEFAULT_PHASE
114	*/
115	public static final int DEFAULT_PHASE = ExecutorConfigurationSupport.DEFAULT_PHASE;
116
117	private static final TimeUnit NANO = TimeUnit.NANOSECONDS;	3✔
118
119	private final ScheduledExecutorService triggerExecutor = createScheduledExecutor();	4✔
120
121	private final ExecutorLifecycleDelegate triggerLifecycle = new ExecutorLifecycleDelegate(this.triggerExecutor);	7✔
122
123	private final ScheduledExecutorService fixedDelayExecutor = createFixedDelayExecutor();	4✔
124
125	private final ExecutorLifecycleDelegate fixedDelayLifecycle = new ExecutorLifecycleDelegate(this.fixedDelayExecutor);	7✔
126
127	@Nullable
128	private ErrorHandler errorHandler;
129
130	private Clock clock = Clock.systemDefaultZone();	3✔
131
132	private int phase = DEFAULT_PHASE;	4✔
133
134	@Nullable
135	private Executor targetTaskExecutor;
136
137	@Nullable
138	private ApplicationContext applicationContext;
139
140	/**
141	* Provide an {@link ErrorHandler} strategy.
142	*/
143	public void setErrorHandler(ErrorHandler errorHandler) {
144	Assert.notNull(errorHandler, "ErrorHandler is required");	3✔
145	this.errorHandler = errorHandler;	3✔
146	}	1✔
147
148	/**
149	* Set the clock to use for scheduling purposes.
150	* <p>The default clock is the system clock for the default time zone.
151	*
152	* @see Clock#systemDefaultZone()
153	*/
154	public void setClock(Clock clock) {
155	Assert.notNull(clock, "Clock is required");	3✔
156	this.clock = clock;	3✔
157	}	1✔
158
159	@Override
160	public Clock getClock() {
161	return this.clock;	3✔
162	}
163
164	/**
165	* Specify the lifecycle phase for pausing and resuming this executor.
166	* The default is {@link #DEFAULT_PHASE}.
167	*
168	* @see SmartLifecycle#getPhase()
169	*/
170	public void setPhase(int phase) {
171	this.phase = phase;	×
172	}	×
173
174	/**
175	* Return the lifecycle phase for pausing and resuming this executor.
176	*
177	* @see #setPhase
178	*/
179	@Override
180	public int getPhase() {
181	return this.phase;	3✔
182	}
183
184	/**
185	* Specify a custom target {@link Executor} to delegate to for
186	* the individual execution of scheduled tasks. This can for example
187	* be set to a separate thread pool for executing scheduled tasks,
188	* whereas this scheduler keeps using its single scheduler thread.
189	* <p>If not set, the regular {@link SimpleAsyncTaskExecutor}
190	* arrangements kicks in with a new thread per task.
191	*/
192	public void setTargetTaskExecutor(Executor targetTaskExecutor) {
193	this.targetTaskExecutor = (targetTaskExecutor == this ? null : targetTaskExecutor);	6!
194	}	1✔
195
196	@Override
197	public void setApplicationContext(ApplicationContext applicationContext) {
198	this.applicationContext = applicationContext;	3✔
199	}	1✔
200
201	private ScheduledExecutorService createScheduledExecutor() {
202	return new ScheduledThreadPoolExecutor(1, this::newThread) {	20✔
203	@Override
204	protected void beforeExecute(Thread thread, Runnable task) {
205	triggerLifecycle.beforeExecute(thread);	5✔
206	}	1✔
207
208	@Override
209	protected void afterExecute(Runnable task, Throwable ex) {
210	triggerLifecycle.afterExecute();	4✔
211	}	1✔
212	};
213	}
214
215	private ScheduledExecutorService createFixedDelayExecutor() {
216	return new ScheduledThreadPoolExecutor(1, this::newThread) {	20✔
217	@Override
218	protected void beforeExecute(Thread thread, Runnable task) {
219	fixedDelayLifecycle.beforeExecute(thread);	×
220	}	×
221
222	@Override
223	protected void afterExecute(Runnable task, Throwable ex) {
224	fixedDelayLifecycle.afterExecute();	×
225	}	×
226	};
227	}
228
229	@Override
230	protected void doExecute(Runnable task) {
231	if (this.targetTaskExecutor != null) {	3✔
232	this.targetTaskExecutor.execute(task);	5✔
233	}
234	else {
235	super.doExecute(task);	3✔
236	}
237	}	1✔
238
239	private Runnable taskOnSchedulerThread(Runnable task) {
240	return new DelegatingErrorHandlingRunnable(task,	4✔
241	(this.errorHandler != null ? this.errorHandler : TaskUtils.getDefaultErrorHandler(true)));	6!
242	}
243
244	private Runnable scheduledTask(Runnable task) {
245	return () -> execute(new DelegatingErrorHandlingRunnable(task, this::shutdownAwareErrorHandler));	13✔
246	}
247
248	private void shutdownAwareErrorHandler(Throwable ex) {
249	if (this.errorHandler != null) {	3✔
250	this.errorHandler.handleError(ex);	5✔
251	}
252	else if (this.triggerExecutor.isShutdown()) {	4!
253	LoggerFactory.getLogger(getClass()).debug("Ignoring scheduled task exception after shutdown", ex);	×
254	}
255	else {
256	TaskUtils.getDefaultErrorHandler(true).handleError(ex);	4✔
257	}
258	}	1✔
259
260	@Override
261	public void execute(Runnable task) {
262	super.execute(TaskUtils.decorateTaskWithErrorHandler(task, this.errorHandler, false));	7✔
263	}	1✔
264
265	@Override
266	public Future<Void> submit(Runnable task) {
267	return super.submit(TaskUtils.decorateTaskWithErrorHandler(task, this.errorHandler, false));	8✔
268	}
269
270	@Override
271	public <T> Future<T> submit(Callable<T> task) {
272	return super.submit(new DelegatingErrorHandlingCallable<>(task, this.errorHandler));	9✔
273	}
274
275	@Override
276	@Nullable
277	public ScheduledFuture<?> schedule(Runnable task, Trigger trigger) {
278	try {
279	Runnable delegate = scheduledTask(task);	4✔
280	ErrorHandler errorHandler =
281	(this.errorHandler != null ? this.errorHandler : TaskUtils.getDefaultErrorHandler(true));	9✔
282	return new ReschedulingRunnable(	11✔
283	delegate, trigger, this.clock, this.triggerExecutor, errorHandler).schedule();	1✔
284	}
285	catch (RejectedExecutionException ex) {	×
286	throw new TaskRejectedException(this.triggerExecutor, task, ex);	×
287	}
288	}
289
290	@Override
291	public ScheduledFuture<?> schedule(Runnable task, Instant startTime) {
292	Duration delay = Duration.between(this.clock.instant(), startTime);	6✔
293	try {
294	return this.triggerExecutor.schedule(scheduledTask(task), NANO.convert(delay), NANO);	11✔
295	}
296	catch (RejectedExecutionException ex) {	×
297	throw new TaskRejectedException(this.triggerExecutor, task, ex);	×
298	}
299	}
300
301	@Override
302	public ScheduledFuture<?> scheduleAtFixedRate(Runnable task, Instant startTime, Duration period) {
303	Duration initialDelay = Duration.between(this.clock.instant(), startTime);	×
304	try {
305	return this.triggerExecutor.scheduleAtFixedRate(scheduledTask(task),	×
306	NANO.convert(initialDelay), NANO.convert(period), NANO);	×
307	}
308	catch (RejectedExecutionException ex) {	×
309	throw new TaskRejectedException(this.triggerExecutor, task, ex);	×
310	}
311	}
312
313	@Override
314	public ScheduledFuture<?> scheduleAtFixedRate(Runnable task, Duration period) {
315	try {
316	return this.triggerExecutor.scheduleAtFixedRate(scheduledTask(task),	10✔
317	0, NANO.convert(period), NANO);	2✔
318	}
319	catch (RejectedExecutionException ex) {	×
320	throw new TaskRejectedException(this.triggerExecutor, task, ex);	×
321	}
322	}
323
324	@Override
325	public ScheduledFuture<?> scheduleWithFixedDelay(Runnable task, Instant startTime, Duration delay) {
326	Duration initialDelay = Duration.between(this.clock.instant(), startTime);	6✔
327	try {
328	// Blocking task on scheduler thread for fixed delay semantics
329	return this.fixedDelayExecutor.scheduleWithFixedDelay(taskOnSchedulerThread(task),	9✔
330	NANO.convert(initialDelay), NANO.convert(delay), NANO);	5✔
331	}
332	catch (RejectedExecutionException ex) {	×
333	throw new TaskRejectedException(this.fixedDelayExecutor, task, ex);	×
334	}
335	}
336
337	@Override
338	public ScheduledFuture<?> scheduleWithFixedDelay(Runnable task, Duration delay) {
339	try {
340	// Blocking task on scheduler thread for fixed delay semantics
341	return this.fixedDelayExecutor.scheduleWithFixedDelay(taskOnSchedulerThread(task),	×
342	0, NANO.convert(delay), NANO);	×
343	}
344	catch (RejectedExecutionException ex) {	×
345	throw new TaskRejectedException(this.fixedDelayExecutor, task, ex);	×
346	}
347	}
348
349	@Override
350	public void start() {
351	this.triggerLifecycle.start();	3✔
352	this.fixedDelayLifecycle.start();	3✔
353	}	1✔
354
355	@Override
356	public void stop() {
357	this.triggerLifecycle.stop();	3✔
358	this.fixedDelayLifecycle.stop();	3✔
359	}	1✔
360
361	@Override
362	public void stop(Runnable callback) {
363	this.triggerLifecycle.stop(); // no callback necessary since it's just triggers with hand-offs	×
364	this.fixedDelayLifecycle.stop(callback); // callback for currently executing fixed-delay tasks	×
365	}	×
366
367	@Override
368	public boolean isRunning() {
369	return triggerLifecycle.isRunning() \|\| fixedDelayLifecycle.isRunning();	12!
370	}
371
372	@Override
373	public void onApplicationEvent(ContextClosedEvent event) {
374	if (event.getApplicationContext() == this.applicationContext) {	5!
375	this.triggerExecutor.shutdown();	3✔
376	this.fixedDelayExecutor.shutdown();	3✔
377	}
378	}	1✔
379
380	@Override
381	public void close() {
382	for (Runnable remainingTask : this.triggerExecutor.shutdownNow()) {	8!
383	if (remainingTask instanceof java.util.concurrent.Future<?> future) {	×
384	future.cancel(true);	×
385	}
386	}	×
387	for (Runnable remainingTask : this.fixedDelayExecutor.shutdownNow()) {	8!
388	if (remainingTask instanceof Future<?> future) {	×
389	future.cancel(true);	×
390	}
391	}	×
392	super.close();	2✔
393	}	1✔
394
395	}

TAKETODAY / today-infrastructure / 18154768944

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous