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

48.72

today-context/src/main/java/infra/scheduling/concurrent/ConcurrentTaskScheduler.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.Date;
import java.util.concurrent.Callable;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;

import infra.core.task.TaskRejectedException;
import infra.lang.Assert;
import infra.scheduling.SchedulingTaskExecutor;
import infra.scheduling.TaskScheduler;
import infra.scheduling.Trigger;
import infra.scheduling.TriggerContext;
import infra.scheduling.support.TaskUtils;
import infra.util.ClassUtils;
import infra.util.ErrorHandler;
import infra.util.concurrent.Future;
import jakarta.enterprise.concurrent.LastExecution;
import jakarta.enterprise.concurrent.ManagedScheduledExecutorService;

/**
 * Adapter that takes a {@code java.util.concurrent.ScheduledExecutorService} and
 * exposes a Framework {@link TaskScheduler} for it.
 * Extends {@link ConcurrentTaskExecutor} in order to implement the
 * {@link SchedulingTaskExecutor} interface as well.
 *
 * <p>Autodetects a JSR-236 {@link jakarta.enterprise.concurrent.ManagedScheduledExecutorService}
 * in order to use it for trigger-based scheduling if possible, instead of
 * local trigger management which ends up delegating to regular delay-based scheduling
 * against the {@code java.util.concurrent.ScheduledExecutorService} API. For JSR-236 style
 * lookup in a Jakarta EE environment, consider using {@link DefaultManagedTaskScheduler}.
 *
 * <p>Note that there is a pre-built {@link ThreadPoolTaskScheduler} that allows for
 * defining a {@link java.util.concurrent.ScheduledThreadPoolExecutor} in bean style,
 * exposing it as a Framework {@link TaskScheduler} directly.
 * This is a convenient alternative to a raw ScheduledThreadPoolExecutor definition with
 * a separate definition of the present adapter class.
 *
 * @author Juergen Hoeller
 * @author Mark Fisher
 * @author <a href="https://github.com/TAKETODAY">Harry Yang</a>
 * @see ScheduledExecutorService
 * @see java.util.concurrent.ScheduledThreadPoolExecutor
 * @see Executors
 * @see DefaultManagedTaskScheduler
 * @see ThreadPoolTaskScheduler
 * @since 4.0
 */
public class ConcurrentTaskScheduler extends ConcurrentTaskExecutor implements TaskScheduler {

  private static final TimeUnit NANO = TimeUnit.NANOSECONDS;

  @Nullable
  private static final Class<?> managedScheduledExecutorServiceClass = ClassUtils.load(
          "jakarta.enterprise.concurrent.ManagedScheduledExecutorService",
          ConcurrentTaskScheduler.class.getClassLoader()
  );

  @Nullable
  private ScheduledExecutorService scheduledExecutor;

  private boolean enterpriseConcurrentScheduler = false;

  @Nullable
  private ErrorHandler errorHandler;

  private Clock clock = Clock.systemDefaultZone();

  /**
   * Create a new ConcurrentTaskScheduler,
   * using a single thread executor as default.
   *
   * @see java.util.concurrent.Executors#newSingleThreadScheduledExecutor()
   */
  public ConcurrentTaskScheduler() {
    super();
    this.scheduledExecutor = Executors.newSingleThreadScheduledExecutor();
    this.enterpriseConcurrentScheduler = false;
  }

  /**
   * Create a new ConcurrentTaskScheduler, using the given
   * {@link java.util.concurrent.ScheduledExecutorService} as shared delegate.
   * <p>Autodetects a JSR-236 {@link jakarta.enterprise.concurrent.ManagedScheduledExecutorService}
   * in order to use it for trigger-based scheduling if possible,
   * instead of Infra local trigger management.
   *
   * @param scheduledExecutor the {@link java.util.concurrent.ScheduledExecutorService}
   * to delegate to for {@link SchedulingTaskExecutor}
   * as well as {@link TaskScheduler} invocations
   */
  public ConcurrentTaskScheduler(@Nullable ScheduledExecutorService scheduledExecutor) {
    super(scheduledExecutor);
    if (scheduledExecutor != null) {
      initScheduledExecutor(scheduledExecutor);
    }
  }

  /**
   * Create a new ConcurrentTaskScheduler, using the given {@link java.util.concurrent.Executor}
   * and {@link java.util.concurrent.ScheduledExecutorService} as delegates.
   * <p>Autodetects a JSR-236 {@link jakarta.enterprise.concurrent.ManagedScheduledExecutorService}
   * in order to use it for trigger-based scheduling if possible,
   * instead of Infra local trigger management.
   *
   * @param concurrentExecutor the {@link java.util.concurrent.Executor} to delegate to
   * for {@link SchedulingTaskExecutor} invocations
   * @param scheduledExecutor the {@link java.util.concurrent.ScheduledExecutorService}
   * to delegate to for {@link TaskScheduler} invocations
   */
  public ConcurrentTaskScheduler(Executor concurrentExecutor, ScheduledExecutorService scheduledExecutor) {
    super(concurrentExecutor);
    initScheduledExecutor(scheduledExecutor);
  }

  private void initScheduledExecutor(ScheduledExecutorService scheduledExecutor) {
    this.scheduledExecutor = scheduledExecutor;
    this.enterpriseConcurrentScheduler = (managedScheduledExecutorServiceClass != null &&
            managedScheduledExecutorServiceClass.isInstance(scheduledExecutor));
  }

  /**
   * Specify the {@link java.util.concurrent.ScheduledExecutorService} to delegate to.
   * <p>Autodetects a JSR-236 {@link jakarta.enterprise.concurrent.ManagedScheduledExecutorService}
   * in order to use it for trigger-based scheduling if possible,
   * instead of Infra local trigger management.
   * <p>Note: This will only apply to {@link TaskScheduler} invocations.
   * If you want the given executor to apply to
   * {@link SchedulingTaskExecutor} invocations
   * as well, pass the same executor reference to {@link #setConcurrentExecutor}.
   *
   * @see #setConcurrentExecutor
   */
  public void setScheduledExecutor(ScheduledExecutorService scheduledExecutor) {
    initScheduledExecutor(scheduledExecutor);
  }

  private ScheduledExecutorService getScheduledExecutor() {
    if (this.scheduledExecutor == null) {
      throw new IllegalStateException("No ScheduledExecutor is configured");
    }
    return this.scheduledExecutor;
  }

  /**
   * 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;
  }

  @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) {
    ScheduledExecutorService scheduleExecutorToUse = getScheduledExecutor();
    try {
      if (this.enterpriseConcurrentScheduler) {
        return new EnterpriseConcurrentTriggerScheduler().schedule(decorateTask(task, true), trigger);
      }
      else {
        ErrorHandler errorHandler =
                (this.errorHandler != null ? this.errorHandler : TaskUtils.getDefaultErrorHandler(true));
        return new ReschedulingRunnable(
                decorateTaskIfNecessary(task), trigger, this.clock, scheduleExecutorToUse, errorHandler)
                .schedule();
      }
    }
    catch (RejectedExecutionException ex) {
      throw new TaskRejectedException(scheduleExecutorToUse, task, ex);
    }
  }

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

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

  @Override
  public ScheduledFuture<?> scheduleAtFixedRate(Runnable task, Duration period) {
    ScheduledExecutorService scheduleExecutorToUse = getScheduledExecutor();
    try {
      return scheduleExecutorToUse.scheduleAtFixedRate(decorateTask(task, true),
              0, NANO.convert(period), NANO);
    }
    catch (RejectedExecutionException ex) {
      throw new TaskRejectedException(scheduleExecutorToUse, task, ex);
    }
  }

  @Override
  public ScheduledFuture<?> scheduleWithFixedDelay(Runnable task, Instant startTime, Duration delay) {
    ScheduledExecutorService scheduleExecutorToUse = getScheduledExecutor();
    Duration initialDelay = Duration.between(this.clock.instant(), startTime);
    try {
      return scheduleExecutorToUse.scheduleWithFixedDelay(decorateTask(task, true),
              NANO.convert(initialDelay), NANO.convert(delay), NANO);
    }
    catch (RejectedExecutionException ex) {
      throw new TaskRejectedException(scheduleExecutorToUse, task, ex);
    }
  }

  @Override
  public ScheduledFuture<?> scheduleWithFixedDelay(Runnable task, Duration delay) {
    ScheduledExecutorService scheduleExecutorToUse = getScheduledExecutor();
    try {
      return scheduleExecutorToUse.scheduleWithFixedDelay(decorateTask(task, true),
              0, NANO.convert(delay), NANO);
    }
    catch (RejectedExecutionException ex) {
      throw new TaskRejectedException(scheduleExecutorToUse, task, ex);
    }
  }

  private Runnable decorateTask(Runnable task, boolean isRepeatingTask) {
    Runnable result = TaskUtils.decorateTaskWithErrorHandler(task, this.errorHandler, isRepeatingTask);
    result = decorateTaskIfNecessary(result);
    if (this.enterpriseConcurrentScheduler) {
      result = ManagedTaskBuilder.buildManagedTask(result, task.toString());
    }
    return result;
  }

  /**
   * Delegate that adapts an Infra Trigger to a JSR-236 Trigger.
   * Separated into an inner class in order to avoid a hard dependency on the JSR-236 API.
   */
  private final class EnterpriseConcurrentTriggerScheduler {

    public ScheduledFuture<?> schedule(Runnable task, Trigger trigger) {
      ManagedScheduledExecutorService executor = (ManagedScheduledExecutorService) getScheduledExecutor();
      return executor.schedule(task, new TriggerAdapter(trigger));
    }

    private static class TriggerAdapter implements jakarta.enterprise.concurrent.Trigger {

      private final Trigger adaptee;

      public TriggerAdapter(Trigger adaptee) {
        this.adaptee = adaptee;
      }

      @Override
      @Nullable
      public Date getNextRunTime(@Nullable LastExecution le, Date taskScheduledTime) {
        Instant instant = this.adaptee.nextExecution(new LastExecutionAdapter(le));
        return (instant != null ? Date.from(instant) : null);
      }

      @Override
      public boolean skipRun(LastExecution lastExecutionInfo, Date scheduledRunTime) {
        return false;
      }

      private static class LastExecutionAdapter implements TriggerContext {

        @Nullable
        private final LastExecution le;

        public LastExecutionAdapter(@Nullable LastExecution le) {
          this.le = le;
        }

        @Nullable
        @Override
        public Instant lastScheduledExecution() {
          return this.le != null ? toInstant(this.le.getScheduledStart()) : null;
        }

        @Nullable
        @Override
        public Instant lastActualExecution() {
          return (this.le != null ? toInstant(this.le.getRunStart()) : null);
        }

        @Nullable
        @Override
        public Instant lastCompletion() {
          return (this.le != null ? toInstant(this.le.getRunEnd()) : null);
        }

        @Nullable
        private static Instant toInstant(@Nullable Date date) {
          return (date != null ? date.toInstant() : null);
        }
      }
    }
  }

}

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.Date;
26	import java.util.concurrent.Callable;
27	import java.util.concurrent.Executor;
28	import java.util.concurrent.Executors;
29	import java.util.concurrent.RejectedExecutionException;
30	import java.util.concurrent.ScheduledExecutorService;
31	import java.util.concurrent.ScheduledFuture;
32	import java.util.concurrent.TimeUnit;
33
34	import infra.core.task.TaskRejectedException;
35	import infra.lang.Assert;
36	import infra.scheduling.SchedulingTaskExecutor;
37	import infra.scheduling.TaskScheduler;
38	import infra.scheduling.Trigger;
39	import infra.scheduling.TriggerContext;
40	import infra.scheduling.support.TaskUtils;
41	import infra.util.ClassUtils;
42	import infra.util.ErrorHandler;
43	import infra.util.concurrent.Future;
44	import jakarta.enterprise.concurrent.LastExecution;
45	import jakarta.enterprise.concurrent.ManagedScheduledExecutorService;
46
47	/**
48	* Adapter that takes a {@code java.util.concurrent.ScheduledExecutorService} and
49	* exposes a Framework {@link TaskScheduler} for it.
50	* Extends {@link ConcurrentTaskExecutor} in order to implement the
51	* {@link SchedulingTaskExecutor} interface as well.
52	*
53	* <p>Autodetects a JSR-236 {@link jakarta.enterprise.concurrent.ManagedScheduledExecutorService}
54	* in order to use it for trigger-based scheduling if possible, instead of
55	* local trigger management which ends up delegating to regular delay-based scheduling
56	* against the {@code java.util.concurrent.ScheduledExecutorService} API. For JSR-236 style
57	* lookup in a Jakarta EE environment, consider using {@link DefaultManagedTaskScheduler}.
58	*
59	* <p>Note that there is a pre-built {@link ThreadPoolTaskScheduler} that allows for
60	* defining a {@link java.util.concurrent.ScheduledThreadPoolExecutor} in bean style,
61	* exposing it as a Framework {@link TaskScheduler} directly.
62	* This is a convenient alternative to a raw ScheduledThreadPoolExecutor definition with
63	* a separate definition of the present adapter class.
64	*
65	* @author Juergen Hoeller
66	* @author Mark Fisher
67	* @author <a href="https://github.com/TAKETODAY">Harry Yang</a>
68	* @see ScheduledExecutorService
69	* @see java.util.concurrent.ScheduledThreadPoolExecutor
70	* @see Executors
71	* @see DefaultManagedTaskScheduler
72	* @see ThreadPoolTaskScheduler
73	* @since 4.0
74	*/
75	public class ConcurrentTaskScheduler extends ConcurrentTaskExecutor implements TaskScheduler {
76
77	private static final TimeUnit NANO = TimeUnit.NANOSECONDS;	2✔
78
79	@Nullable
80	private static final Class<?> managedScheduledExecutorServiceClass = ClassUtils.load(	5✔
81	"jakarta.enterprise.concurrent.ManagedScheduledExecutorService",
82	ConcurrentTaskScheduler.class.getClassLoader()	1✔
83	);
84
85	@Nullable
86	private ScheduledExecutorService scheduledExecutor;
87
88	private boolean enterpriseConcurrentScheduler = false;	3✔
89
90	@Nullable
91	private ErrorHandler errorHandler;
92
93	private Clock clock = Clock.systemDefaultZone();	3✔
94
95	/**
96	* Create a new ConcurrentTaskScheduler,
97	* using a single thread executor as default.
98	*
99	* @see java.util.concurrent.Executors#newSingleThreadScheduledExecutor()
100	*/
101	public ConcurrentTaskScheduler() {
102	super();	×
103	this.scheduledExecutor = Executors.newSingleThreadScheduledExecutor();	×
104	this.enterpriseConcurrentScheduler = false;	×
105	}	×
106
107	/**
108	* Create a new ConcurrentTaskScheduler, using the given
109	* {@link java.util.concurrent.ScheduledExecutorService} as shared delegate.
110	* <p>Autodetects a JSR-236 {@link jakarta.enterprise.concurrent.ManagedScheduledExecutorService}
111	* in order to use it for trigger-based scheduling if possible,
112	* instead of Infra local trigger management.
113	*
114	* @param scheduledExecutor the {@link java.util.concurrent.ScheduledExecutorService}
115	* to delegate to for {@link SchedulingTaskExecutor}
116	* as well as {@link TaskScheduler} invocations
117	*/
118	public ConcurrentTaskScheduler(@Nullable ScheduledExecutorService scheduledExecutor) {
119	super(scheduledExecutor);	3✔
120	if (scheduledExecutor != null) {	2✔
121	initScheduledExecutor(scheduledExecutor);	3✔
122	}
123	}	1✔
124
125	/**
126	* Create a new ConcurrentTaskScheduler, using the given {@link java.util.concurrent.Executor}
127	* and {@link java.util.concurrent.ScheduledExecutorService} as delegates.
128	* <p>Autodetects a JSR-236 {@link jakarta.enterprise.concurrent.ManagedScheduledExecutorService}
129	* in order to use it for trigger-based scheduling if possible,
130	* instead of Infra local trigger management.
131	*
132	* @param concurrentExecutor the {@link java.util.concurrent.Executor} to delegate to
133	* for {@link SchedulingTaskExecutor} invocations
134	* @param scheduledExecutor the {@link java.util.concurrent.ScheduledExecutorService}
135	* to delegate to for {@link TaskScheduler} invocations
136	*/
137	public ConcurrentTaskScheduler(Executor concurrentExecutor, ScheduledExecutorService scheduledExecutor) {
138	super(concurrentExecutor);	×
139	initScheduledExecutor(scheduledExecutor);	×
140	}	×
141
142	private void initScheduledExecutor(ScheduledExecutorService scheduledExecutor) {
143	this.scheduledExecutor = scheduledExecutor;	3✔
144	this.enterpriseConcurrentScheduler = (managedScheduledExecutorServiceClass != null &&	5✔
145	managedScheduledExecutorServiceClass.isInstance(scheduledExecutor));	4!
146	}	1✔
147
148	/**
149	* Specify the {@link java.util.concurrent.ScheduledExecutorService} to delegate to.
150	* <p>Autodetects a JSR-236 {@link jakarta.enterprise.concurrent.ManagedScheduledExecutorService}
151	* in order to use it for trigger-based scheduling if possible,
152	* instead of Infra local trigger management.
153	* <p>Note: This will only apply to {@link TaskScheduler} invocations.
154	* If you want the given executor to apply to
155	* {@link SchedulingTaskExecutor} invocations
156	* as well, pass the same executor reference to {@link #setConcurrentExecutor}.
157	*
158	* @see #setConcurrentExecutor
159	*/
160	public void setScheduledExecutor(ScheduledExecutorService scheduledExecutor) {
161	initScheduledExecutor(scheduledExecutor);	×
162	}	×
163
164	private ScheduledExecutorService getScheduledExecutor() {
165	if (this.scheduledExecutor == null) {	3✔
166	throw new IllegalStateException("No ScheduledExecutor is configured");	5✔
167	}
168	return this.scheduledExecutor;	3✔
169	}
170
171	/**
172	* Provide an {@link ErrorHandler} strategy.
173	*/
174	public void setErrorHandler(ErrorHandler errorHandler) {
175	Assert.notNull(errorHandler, "ErrorHandler is required");	3✔
176	this.errorHandler = errorHandler;	3✔
177	}	1✔
178
179	/**
180	* Set the clock to use for scheduling purposes.
181	* <p>The default clock is the system clock for the default time zone.
182	*
183	* @see Clock#systemDefaultZone()
184	*/
185	public void setClock(Clock clock) {
186	Assert.notNull(clock, "Clock is required");	×
187	this.clock = clock;	×
188	}	×
189
190	@Override
191	public Clock getClock() {
192	return this.clock;	×
193	}
194
195	@Override
196	public void execute(Runnable task) {
197	super.execute(TaskUtils.decorateTaskWithErrorHandler(task, this.errorHandler, false));	7✔
198	}	1✔
199
200	@Override
201	public Future<Void> submit(Runnable task) {
202	return super.submit(TaskUtils.decorateTaskWithErrorHandler(task, this.errorHandler, false));	8✔
203	}
204
205	@Override
206	public <T> Future<T> submit(Callable<T> task) {
207	return super.submit(new DelegatingErrorHandlingCallable<>(task, this.errorHandler));	9✔
208	}
209
210	@Override
211	@Nullable
212	public ScheduledFuture<?> schedule(Runnable task, Trigger trigger) {
213	ScheduledExecutorService scheduleExecutorToUse = getScheduledExecutor();	3✔
214	try {
215	if (this.enterpriseConcurrentScheduler) {	3!
216	return new EnterpriseConcurrentTriggerScheduler().schedule(decorateTask(task, true), trigger);	×
217	}
218	else {
219	ErrorHandler errorHandler =
220	(this.errorHandler != null ? this.errorHandler : TaskUtils.getDefaultErrorHandler(true));	6!
221	return new ReschedulingRunnable(	5✔
222	decorateTaskIfNecessary(task), trigger, this.clock, scheduleExecutorToUse, errorHandler)	7✔
223	.schedule();	1✔
224	}
225	}
226	catch (RejectedExecutionException ex) {	×
227	throw new TaskRejectedException(scheduleExecutorToUse, task, ex);	×
228	}
229	}
230
231	@Override
232	public ScheduledFuture<?> schedule(Runnable task, Instant startTime) {
233	ScheduledExecutorService scheduleExecutorToUse = getScheduledExecutor();	3✔
234	Duration delay = Duration.between(this.clock.instant(), startTime);	6✔
235	try {
236	return scheduleExecutorToUse.schedule(decorateTask(task, false), NANO.convert(delay), NANO);	11✔
237	}
238	catch (RejectedExecutionException ex) {	×
239	throw new TaskRejectedException(scheduleExecutorToUse, task, ex);	×
240	}
241	}
242
243	@Override
244	public ScheduledFuture<?> scheduleAtFixedRate(Runnable task, Instant startTime, Duration period) {
245	ScheduledExecutorService scheduleExecutorToUse = getScheduledExecutor();	3✔
246	Duration initialDelay = Duration.between(this.clock.instant(), startTime);	6✔
247	try {
248	return scheduleExecutorToUse.scheduleAtFixedRate(decorateTask(task, true),	9✔
249	NANO.convert(initialDelay), NANO.convert(period), NANO);	5✔
250	}
251	catch (RejectedExecutionException ex) {	×
252	throw new TaskRejectedException(scheduleExecutorToUse, task, ex);	×
253	}
254	}
255
256	@Override
257	public ScheduledFuture<?> scheduleAtFixedRate(Runnable task, Duration period) {
258	ScheduledExecutorService scheduleExecutorToUse = getScheduledExecutor();	3✔
259	try {
260	return scheduleExecutorToUse.scheduleAtFixedRate(decorateTask(task, true),	10✔
261	0, NANO.convert(period), NANO);	2✔
262	}
263	catch (RejectedExecutionException ex) {	×
264	throw new TaskRejectedException(scheduleExecutorToUse, task, ex);	×
265	}
266	}
267
268	@Override
269	public ScheduledFuture<?> scheduleWithFixedDelay(Runnable task, Instant startTime, Duration delay) {
270	ScheduledExecutorService scheduleExecutorToUse = getScheduledExecutor();	3✔
271	Duration initialDelay = Duration.between(this.clock.instant(), startTime);	6✔
272	try {
273	return scheduleExecutorToUse.scheduleWithFixedDelay(decorateTask(task, true),	9✔
274	NANO.convert(initialDelay), NANO.convert(delay), NANO);	5✔
275	}
276	catch (RejectedExecutionException ex) {	×
277	throw new TaskRejectedException(scheduleExecutorToUse, task, ex);	×
278	}
279	}
280
281	@Override
282	public ScheduledFuture<?> scheduleWithFixedDelay(Runnable task, Duration delay) {
283	ScheduledExecutorService scheduleExecutorToUse = getScheduledExecutor();	×
284	try {
285	return scheduleExecutorToUse.scheduleWithFixedDelay(decorateTask(task, true),	×
286	0, NANO.convert(delay), NANO);	×
287	}
288	catch (RejectedExecutionException ex) {	×
289	throw new TaskRejectedException(scheduleExecutorToUse, task, ex);	×
290	}
291	}
292
293	private Runnable decorateTask(Runnable task, boolean isRepeatingTask) {
294	Runnable result = TaskUtils.decorateTaskWithErrorHandler(task, this.errorHandler, isRepeatingTask);	6✔
295	result = decorateTaskIfNecessary(result);	4✔
296	if (this.enterpriseConcurrentScheduler) {	3!
297	result = ManagedTaskBuilder.buildManagedTask(result, task.toString());	×
298	}
299	return result;	2✔
300	}
301
302	/**
303	* Delegate that adapts an Infra Trigger to a JSR-236 Trigger.
304	* Separated into an inner class in order to avoid a hard dependency on the JSR-236 API.
305	*/
306	private final class EnterpriseConcurrentTriggerScheduler {	×
307
308	public ScheduledFuture<?> schedule(Runnable task, Trigger trigger) {
309	ManagedScheduledExecutorService executor = (ManagedScheduledExecutorService) getScheduledExecutor();	×
310	return executor.schedule(task, new TriggerAdapter(trigger));	×
311	}
312
313	private static class TriggerAdapter implements jakarta.enterprise.concurrent.Trigger {
314
315	private final Trigger adaptee;
316
317	public TriggerAdapter(Trigger adaptee) {	×
318	this.adaptee = adaptee;	×
319	}	×
320
321	@Override
322	@Nullable
323	public Date getNextRunTime(@Nullable LastExecution le, Date taskScheduledTime) {
324	Instant instant = this.adaptee.nextExecution(new LastExecutionAdapter(le));	×
325	return (instant != null ? Date.from(instant) : null);	×
326	}
327
328	@Override
329	public boolean skipRun(LastExecution lastExecutionInfo, Date scheduledRunTime) {
330	return false;	×
331	}
332
333	private static class LastExecutionAdapter implements TriggerContext {
334
335	@Nullable
336	private final LastExecution le;
337
338	public LastExecutionAdapter(@Nullable LastExecution le) {	×
339	this.le = le;	×
340	}	×
341
342	@Nullable
343	@Override
344	public Instant lastScheduledExecution() {
345	return this.le != null ? toInstant(this.le.getScheduledStart()) : null;	×
346	}
347
348	@Nullable
349	@Override
350	public Instant lastActualExecution() {
351	return (this.le != null ? toInstant(this.le.getRunStart()) : null);	×
352	}
353
354	@Nullable
355	@Override
356	public Instant lastCompletion() {
357	return (this.le != null ? toInstant(this.le.getRunEnd()) : null);	×
358	}
359
360	@Nullable
361	private static Instant toInstant(@Nullable Date date) {
362	return (date != null ? date.toInstant() : null);	×
363	}
364	}
365	}
366	}
367
368	}

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