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

59.02

today-context/src/main/java/infra/scheduling/concurrent/ReschedulingRunnable.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.Delayed;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import infra.lang.Assert;
import infra.scheduling.Trigger;
import infra.scheduling.support.DelegatingErrorHandlingRunnable;
import infra.scheduling.support.SimpleTriggerContext;
import infra.util.ErrorHandler;

/**
 * Internal adapter that reschedules an underlying {@link Runnable} according
 * to the next execution time suggested by a given {@link Trigger}.
 *
 * <p>Necessary because a native {@link ScheduledExecutorService} supports
 * delay-driven execution only. The flexibility of the {@link Trigger} interface
 * will be translated onto a delay for the next execution time (repeatedly).
 *
 * @author Juergen Hoeller
 * @author Mark Fisher
 * @author <a href="https://github.com/TAKETODAY">海子 Yang</a>
 * @since 4.0
 */
class ReschedulingRunnable extends DelegatingErrorHandlingRunnable implements ScheduledFuture<Object> {

  private final Trigger trigger;

  private final SimpleTriggerContext triggerContext;

  private final ScheduledExecutorService executor;

  @Nullable
  private ScheduledFuture<?> currentFuture;

  @Nullable
  private Instant scheduledExecutionTime;

  private final Object triggerContextMonitor = new Object();

  public ReschedulingRunnable(Runnable delegate, Trigger trigger, Clock clock,
          ScheduledExecutorService executor, ErrorHandler errorHandler) {

    super(delegate, errorHandler);
    this.trigger = trigger;
    this.triggerContext = new SimpleTriggerContext(clock);
    this.executor = executor;
  }

  @Nullable
  public ScheduledFuture<?> schedule() {
    synchronized(this.triggerContextMonitor) {
      this.scheduledExecutionTime = trigger.nextExecution(this.triggerContext);
      if (this.scheduledExecutionTime == null) {
        return null;
      }
      Duration initialDelay = Duration.between(triggerContext.getClock().instant(), scheduledExecutionTime);
      this.currentFuture = executor.schedule(this, initialDelay.toNanos(), TimeUnit.NANOSECONDS);
      return this;
    }
  }

  private ScheduledFuture<?> obtainCurrentFuture() {
    Assert.state(this.currentFuture != null, "No scheduled future");
    return this.currentFuture;
  }

  @Override
  public void run() {
    Instant actualExecutionTime = triggerContext.getClock().instant();
    super.run();
    Instant completionTime = triggerContext.getClock().instant();
    synchronized(this.triggerContextMonitor) {
      Assert.state(this.scheduledExecutionTime != null, "No scheduled execution");
      this.triggerContext.update(this.scheduledExecutionTime, actualExecutionTime, completionTime);
      if (!obtainCurrentFuture().isCancelled()) {
        schedule();
      }
    }
  }

  @Override
  public boolean cancel(boolean mayInterruptIfRunning) {
    synchronized(this.triggerContextMonitor) {
      return obtainCurrentFuture().cancel(mayInterruptIfRunning);
    }
  }

  @Override
  public boolean isCancelled() {
    synchronized(this.triggerContextMonitor) {
      return obtainCurrentFuture().isCancelled();
    }
  }

  @Override
  public boolean isDone() {
    synchronized(this.triggerContextMonitor) {
      return obtainCurrentFuture().isDone();
    }
  }

  @Override
  public Object get() throws InterruptedException, ExecutionException {
    ScheduledFuture<?> curr;
    synchronized(this.triggerContextMonitor) {
      curr = obtainCurrentFuture();
    }
    return curr.get();
  }

  @Override
  public Object get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
    ScheduledFuture<?> curr;
    synchronized(this.triggerContextMonitor) {
      curr = obtainCurrentFuture();
    }
    return curr.get(timeout, unit);
  }

  @Override
  public long getDelay(TimeUnit unit) {
    ScheduledFuture<?> curr;
    synchronized(this.triggerContextMonitor) {
      curr = obtainCurrentFuture();
    }
    return curr.getDelay(unit);
  }

  @Override
  public int compareTo(Delayed other) {
    if (this == other) {
      return 0;
    }
    long diff = getDelay(TimeUnit.NANOSECONDS) - other.getDelay(TimeUnit.NANOSECONDS);
    return (diff == 0 ? 0 : (diff < 0 ? -1 : 1));
  }

}

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.Delayed;
26	import java.util.concurrent.ExecutionException;
27	import java.util.concurrent.ScheduledExecutorService;
28	import java.util.concurrent.ScheduledFuture;
29	import java.util.concurrent.TimeUnit;
30	import java.util.concurrent.TimeoutException;
31
32	import infra.lang.Assert;
33	import infra.scheduling.Trigger;
34	import infra.scheduling.support.DelegatingErrorHandlingRunnable;
35	import infra.scheduling.support.SimpleTriggerContext;
36	import infra.util.ErrorHandler;
37
38	/**
39	* Internal adapter that reschedules an underlying {@link Runnable} according
40	* to the next execution time suggested by a given {@link Trigger}.
41	*
42	* <p>Necessary because a native {@link ScheduledExecutorService} supports
43	* delay-driven execution only. The flexibility of the {@link Trigger} interface
44	* will be translated onto a delay for the next execution time (repeatedly).
45	*
46	* @author Juergen Hoeller
47	* @author Mark Fisher
48	* @author <a href="https://github.com/TAKETODAY">海子 Yang</a>
49	* @since 4.0
50	*/
51	class ReschedulingRunnable extends DelegatingErrorHandlingRunnable implements ScheduledFuture<Object> {
52
53	private final Trigger trigger;
54
55	private final SimpleTriggerContext triggerContext;
56
57	private final ScheduledExecutorService executor;
58
59	@Nullable
60	private ScheduledFuture<?> currentFuture;
61
62	@Nullable
63	private Instant scheduledExecutionTime;
64
65	private final Object triggerContextMonitor = new Object();	5✔
66
67	public ReschedulingRunnable(Runnable delegate, Trigger trigger, Clock clock,
68	ScheduledExecutorService executor, ErrorHandler errorHandler) {
69
70	super(delegate, errorHandler);	4✔
71	this.trigger = trigger;	3✔
72	this.triggerContext = new SimpleTriggerContext(clock);	6✔
73	this.executor = executor;	3✔
74	}	1✔
75
76	@Nullable
77	public ScheduledFuture<?> schedule() {
78	synchronized(this.triggerContextMonitor) {	5✔
79	this.scheduledExecutionTime = trigger.nextExecution(this.triggerContext);	7✔
80	if (this.scheduledExecutionTime == null) {	3✔
81	return null;	4✔
82	}
83	Duration initialDelay = Duration.between(triggerContext.getClock().instant(), scheduledExecutionTime);	8✔
84	this.currentFuture = executor.schedule(this, initialDelay.toNanos(), TimeUnit.NANOSECONDS);	9✔
85	return this;	4✔
86	}
87	}
88
89	private ScheduledFuture<?> obtainCurrentFuture() {
90	Assert.state(this.currentFuture != null, "No scheduled future");	7!
91	return this.currentFuture;	3✔
92	}
93
94	@Override
95	public void run() {
96	Instant actualExecutionTime = triggerContext.getClock().instant();	5✔
97	super.run();	2✔
98	Instant completionTime = triggerContext.getClock().instant();	5✔
99	synchronized(this.triggerContextMonitor) {	5✔
100	Assert.state(this.scheduledExecutionTime != null, "No scheduled execution");	7!
101	this.triggerContext.update(this.scheduledExecutionTime, actualExecutionTime, completionTime);	7✔
102	if (!obtainCurrentFuture().isCancelled()) {	4!
103	schedule();	3✔
104	}
105	}	3✔
106	}	1✔
107
108	@Override
109	public boolean cancel(boolean mayInterruptIfRunning) {
110	synchronized(this.triggerContextMonitor) {	5✔
111	return obtainCurrentFuture().cancel(mayInterruptIfRunning);	7✔
112	}
113	}
114
115	@Override
116	public boolean isCancelled() {
117	synchronized(this.triggerContextMonitor) {	×
118	return obtainCurrentFuture().isCancelled();	×
119	}
120	}
121
122	@Override
123	public boolean isDone() {
124	synchronized(this.triggerContextMonitor) {	×
125	return obtainCurrentFuture().isDone();	×
126	}
127	}
128
129	@Override
130	public Object get() throws InterruptedException, ExecutionException {
131	ScheduledFuture<?> curr;
132	synchronized(this.triggerContextMonitor) {	×
133	curr = obtainCurrentFuture();	×
134	}	×
135	return curr.get();	×
136	}
137
138	@Override
139	public Object get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
140	ScheduledFuture<?> curr;
141	synchronized(this.triggerContextMonitor) {	5✔
142	curr = obtainCurrentFuture();	3✔
143	}	3✔
144	return curr.get(timeout, unit);	5✔
145	}
146
147	@Override
148	public long getDelay(TimeUnit unit) {
149	ScheduledFuture<?> curr;
150	synchronized(this.triggerContextMonitor) {	×
151	curr = obtainCurrentFuture();	×
152	}	×
153	return curr.getDelay(unit);	×
154	}
155
156	@Override
157	public int compareTo(Delayed other) {
158	if (this == other) {	×
159	return 0;	×
160	}
161	long diff = getDelay(TimeUnit.NANOSECONDS) - other.getDelay(TimeUnit.NANOSECONDS);	×
162	return (diff == 0 ? 0 : (diff < 0 ? -1 : 1));	×
163	}
164
165	}

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