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

87.1

today-context/src/main/java/infra/scheduling/concurrent/ConcurrentTaskExecutor.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.util.HashMap;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;

import infra.core.task.AsyncTaskExecutor;
import infra.core.task.TaskDecorator;
import infra.core.task.TaskExecutor;
import infra.core.task.support.TaskExecutorAdapter;
import infra.scheduling.SchedulingAwareRunnable;
import infra.scheduling.SchedulingTaskExecutor;
import infra.util.ClassUtils;
import infra.util.concurrent.Future;
import jakarta.enterprise.concurrent.ManagedExecutorService;
import jakarta.enterprise.concurrent.ManagedExecutors;
import jakarta.enterprise.concurrent.ManagedTask;

/**
 * Adapter that takes a {@code java.util.concurrent.Executor} and exposes
 * a Framework {@link TaskExecutor} for it.
 * Also detects an extended {@code java.util.concurrent.ExecutorService}, adapting
 * the {@link AsyncTaskExecutor} interface accordingly.
 *
 * <p>Autodetects a JSR-236 {@link ManagedExecutorService}
 * in order to expose {@link jakarta.enterprise.concurrent.ManagedTask} adapters for it,
 * exposing a long-running hint based on {@link SchedulingAwareRunnable} and an identity
 * name based on the given Runnable/Callable's {@code toString()}. For JSR-236 style
 * lookup in a Jakarta EE environment, consider using {@link DefaultManagedTaskExecutor}.
 *
 * <p>Note that there is a pre-built {@link ThreadPoolTaskExecutor} that allows
 * for defining a {@link java.util.concurrent.ThreadPoolExecutor} in bean style,
 * exposing it as a Framework {@link TaskExecutor} directly.
 * This is a convenient alternative to a raw ThreadPoolExecutor definition with
 * a separate definition of the present adapter class.
 *
 * @author Juergen Hoeller
 * @author <a href="https://github.com/TAKETODAY">Harry Yang</a>
 * @see Executor
 * @see java.util.concurrent.ExecutorService
 * @see java.util.concurrent.ThreadPoolExecutor
 * @see Executors
 * @see DefaultManagedTaskExecutor
 * @see ThreadPoolTaskExecutor
 * @since 4.0
 */
public class ConcurrentTaskExecutor implements AsyncTaskExecutor, SchedulingTaskExecutor {

  private static final Executor STUB_EXECUTOR = task -> {
    throw new IllegalStateException("Executor not configured");
  };

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

  private Executor concurrentExecutor = STUB_EXECUTOR;

  private TaskExecutorAdapter adaptedExecutor = new TaskExecutorAdapter(STUB_EXECUTOR);

  @Nullable
  private TaskDecorator taskDecorator;

  /**
   * Create a new ConcurrentTaskExecutor, using a single thread executor as default.
   *
   * @see java.util.concurrent.Executors#newSingleThreadExecutor()
   */
  public ConcurrentTaskExecutor() {
    this.concurrentExecutor = Executors.newSingleThreadExecutor();
    this.adaptedExecutor = new TaskExecutorAdapter(this.concurrentExecutor);
  }

  /**
   * Create a new ConcurrentTaskExecutor, using the given {@link Executor}.
   * <p>Autodetects a JSR-236 {@link ManagedExecutorService}
   * in order to expose {@link jakarta.enterprise.concurrent.ManagedTask} adapters for it.
   *
   * @param executor the {@link Executor} to delegate to
   */
  public ConcurrentTaskExecutor(@Nullable Executor executor) {
    if (executor != null) {
      setConcurrentExecutor(executor);
    }
  }

  /**
   * Specify the {@link Executor} to delegate to.
   * <p>Autodetects a JSR-236 {@link ManagedExecutorService}
   * in order to expose {@link jakarta.enterprise.concurrent.ManagedTask} adapters for it.
   */
  public final void setConcurrentExecutor(Executor executor) {
    this.concurrentExecutor = executor;
    this.adaptedExecutor = getAdaptedExecutor(this.concurrentExecutor);
  }

  /**
   * Return the {@link Executor} that this adapter delegates to.
   */
  public final Executor getConcurrentExecutor() {
    return this.concurrentExecutor;
  }

  /**
   * Specify a custom {@link TaskDecorator} to be applied to any {@link Runnable}
   * about to be executed.
   * <p>Note that such a decorator is not necessarily being applied to the
   * user-supplied {@code Runnable}/{@code Callable} but rather to the actual
   * execution callback (which may be a wrapper around the user-supplied task).
   * <p>The primary use case is to set some execution context around the task's
   * invocation, or to provide some monitoring/statistics for task execution.
   */
  public final void setTaskDecorator(TaskDecorator taskDecorator) {
    this.taskDecorator = taskDecorator;
    this.adaptedExecutor.setTaskDecorator(taskDecorator);
  }

  @Override
  public void execute(Runnable task) {
    this.adaptedExecutor.execute(task);
  }

  @Override
  public void execute(Runnable task, long startTimeout) {
    this.adaptedExecutor.execute(task, startTimeout);
  }

  @Override
  public Future<Void> submit(Runnable task) {
    return this.adaptedExecutor.submit(task);
  }

  @Override
  public <T> Future<T> submit(Callable<T> task) {
    return this.adaptedExecutor.submit(task);
  }

  private TaskExecutorAdapter getAdaptedExecutor(Executor originalExecutor) {
    var adapter = managedExecutorServiceClass != null && managedExecutorServiceClass.isInstance(originalExecutor)
            ? new ManagedTaskExecutorAdapter(originalExecutor)
            : new TaskExecutorAdapter(originalExecutor);
    if (this.taskDecorator != null) {
      adapter.setTaskDecorator(this.taskDecorator);
    }
    return adapter;
  }

  Runnable decorateTaskIfNecessary(Runnable task) {
    return this.taskDecorator != null ? this.taskDecorator.decorate(task) : task;
  }

  /**
   * TaskExecutorAdapter subclass that wraps all provided Runnables and Callables
   * with a JSR-236 ManagedTask, exposing a long-running hint based on
   * {@link SchedulingAwareRunnable} and an identity name based on the task's
   * {@code toString()} representation.
   */
  private static class ManagedTaskExecutorAdapter extends TaskExecutorAdapter {

    public ManagedTaskExecutorAdapter(Executor concurrentExecutor) {
      super(concurrentExecutor);
    }

    @Override
    public void execute(Runnable task) {
      super.execute(ManagedTaskBuilder.buildManagedTask(task, task.toString()));
    }

    @Override
    public Future<Void> submit(Runnable task) {
      return super.submit(ManagedTaskBuilder.buildManagedTask(task, task.toString()));
    }

    @Override
    public <T> Future<T> submit(Callable<T> task) {
      return super.submit(ManagedTaskBuilder.buildManagedTask(task, task.toString()));
    }

  }

  /**
   * Delegate that wraps a given Runnable/Callable  with a JSR-236 ManagedTask,
   * exposing a long-running hint based on {@link SchedulingAwareRunnable}
   * and a given identity name.
   */
  protected static class ManagedTaskBuilder {

    public static Runnable buildManagedTask(Runnable task, String identityName) {
      Map<String, String> properties;
      if (task instanceof SchedulingAwareRunnable sar) {
        properties = new HashMap<>(4);
        properties.put(ManagedTask.LONGRUNNING_HINT, Boolean.toString(sar.isLongLived()));
      }
      else {
        properties = new HashMap<>(2);
      }
      properties.put(ManagedTask.IDENTITY_NAME, identityName);
      return ManagedExecutors.managedTask(task, properties, null);
    }

    public static <T> Callable<T> buildManagedTask(Callable<T> task, String identityName) {
      Map<String, String> properties = new HashMap<>(2);
      properties.put(ManagedTask.IDENTITY_NAME, identityName);
      return ManagedExecutors.managedTask(task, properties, 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.util.HashMap;
23	import java.util.Map;
24	import java.util.concurrent.Callable;
25	import java.util.concurrent.Executor;
26	import java.util.concurrent.Executors;
27
28	import infra.core.task.AsyncTaskExecutor;
29	import infra.core.task.TaskDecorator;
30	import infra.core.task.TaskExecutor;
31	import infra.core.task.support.TaskExecutorAdapter;
32	import infra.scheduling.SchedulingAwareRunnable;
33	import infra.scheduling.SchedulingTaskExecutor;
34	import infra.util.ClassUtils;
35	import infra.util.concurrent.Future;
36	import jakarta.enterprise.concurrent.ManagedExecutorService;
37	import jakarta.enterprise.concurrent.ManagedExecutors;
38	import jakarta.enterprise.concurrent.ManagedTask;
39
40	/**
41	* Adapter that takes a {@code java.util.concurrent.Executor} and exposes
42	* a Framework {@link TaskExecutor} for it.
43	* Also detects an extended {@code java.util.concurrent.ExecutorService}, adapting
44	* the {@link AsyncTaskExecutor} interface accordingly.
45	*
46	* <p>Autodetects a JSR-236 {@link ManagedExecutorService}
47	* in order to expose {@link jakarta.enterprise.concurrent.ManagedTask} adapters for it,
48	* exposing a long-running hint based on {@link SchedulingAwareRunnable} and an identity
49	* name based on the given Runnable/Callable's {@code toString()}. For JSR-236 style
50	* lookup in a Jakarta EE environment, consider using {@link DefaultManagedTaskExecutor}.
51	*
52	* <p>Note that there is a pre-built {@link ThreadPoolTaskExecutor} that allows
53	* for defining a {@link java.util.concurrent.ThreadPoolExecutor} in bean style,
54	* exposing it as a Framework {@link TaskExecutor} directly.
55	* This is a convenient alternative to a raw ThreadPoolExecutor definition with
56	* a separate definition of the present adapter class.
57	*
58	* @author Juergen Hoeller
59	* @author <a href="https://github.com/TAKETODAY">Harry Yang</a>
60	* @see Executor
61	* @see java.util.concurrent.ExecutorService
62	* @see java.util.concurrent.ThreadPoolExecutor
63	* @see Executors
64	* @see DefaultManagedTaskExecutor
65	* @see ThreadPoolTaskExecutor
66	* @since 4.0
67	*/
68	public class ConcurrentTaskExecutor implements AsyncTaskExecutor, SchedulingTaskExecutor {
69
70	private static final Executor STUB_EXECUTOR = task -> {	2✔
71	throw new IllegalStateException("Executor not configured");	5✔
72	};
73
74	@Nullable
75	private static final Class<?> managedExecutorServiceClass = ClassUtils.load(	5✔
76	"jakarta.enterprise.concurrent.ManagedExecutorService", ConcurrentTaskScheduler.class.getClassLoader()	1✔
77	);
78
79	private Executor concurrentExecutor = STUB_EXECUTOR;	6✔
80
81	private TaskExecutorAdapter adaptedExecutor = new TaskExecutorAdapter(STUB_EXECUTOR);	12✔
82
83	@Nullable
84	private TaskDecorator taskDecorator;
85
86	/**
87	* Create a new ConcurrentTaskExecutor, using a single thread executor as default.
88	*
89	* @see java.util.concurrent.Executors#newSingleThreadExecutor()
90	*/
91	public ConcurrentTaskExecutor() {	2✔
92	this.concurrentExecutor = Executors.newSingleThreadExecutor();	3✔
93	this.adaptedExecutor = new TaskExecutorAdapter(this.concurrentExecutor);	7✔
94	}	1✔
95
96	/**
97	* Create a new ConcurrentTaskExecutor, using the given {@link Executor}.
98	* <p>Autodetects a JSR-236 {@link ManagedExecutorService}
99	* in order to expose {@link jakarta.enterprise.concurrent.ManagedTask} adapters for it.
100	*
101	* @param executor the {@link Executor} to delegate to
102	*/
103	public ConcurrentTaskExecutor(@Nullable Executor executor) {	2✔
104	if (executor != null) {	2✔
105	setConcurrentExecutor(executor);	3✔
106	}
107	}	1✔
108
109	/**
110	* Specify the {@link Executor} to delegate to.
111	* <p>Autodetects a JSR-236 {@link ManagedExecutorService}
112	* in order to expose {@link jakarta.enterprise.concurrent.ManagedTask} adapters for it.
113	*/
114	public final void setConcurrentExecutor(Executor executor) {
115	this.concurrentExecutor = executor;	3✔
116	this.adaptedExecutor = getAdaptedExecutor(this.concurrentExecutor);	6✔
117	}	1✔
118
119	/**
120	* Return the {@link Executor} that this adapter delegates to.
121	*/
122	public final Executor getConcurrentExecutor() {
123	return this.concurrentExecutor;	3✔
124	}
125
126	/**
127	* Specify a custom {@link TaskDecorator} to be applied to any {@link Runnable}
128	* about to be executed.
129	* <p>Note that such a decorator is not necessarily being applied to the
130	* user-supplied {@code Runnable}/{@code Callable} but rather to the actual
131	* execution callback (which may be a wrapper around the user-supplied task).
132	* <p>The primary use case is to set some execution context around the task's
133	* invocation, or to provide some monitoring/statistics for task execution.
134	*/
135	public final void setTaskDecorator(TaskDecorator taskDecorator) {
136	this.taskDecorator = taskDecorator;	3✔
137	this.adaptedExecutor.setTaskDecorator(taskDecorator);	4✔
138	}	1✔
139
140	@Override
141	public void execute(Runnable task) {
142	this.adaptedExecutor.execute(task);	4✔
143	}	1✔
144
145	@Override
146	public void execute(Runnable task, long startTimeout) {
147	this.adaptedExecutor.execute(task, startTimeout);	5✔
148	}	1✔
149
150	@Override
151	public Future<Void> submit(Runnable task) {
152	return this.adaptedExecutor.submit(task);	5✔
153	}
154
155	@Override
156	public <T> Future<T> submit(Callable<T> task) {
157	return this.adaptedExecutor.submit(task);	5✔
158	}
159
160	private TaskExecutorAdapter getAdaptedExecutor(Executor originalExecutor) {
161	var adapter = managedExecutorServiceClass != null && managedExecutorServiceClass.isInstance(originalExecutor)	6✔
162	? new ManagedTaskExecutorAdapter(originalExecutor)	5✔
163	: new TaskExecutorAdapter(originalExecutor);	5✔
164	if (this.taskDecorator != null) {	3✔
165	adapter.setTaskDecorator(this.taskDecorator);	4✔
166	}
167	return adapter;	2✔
168	}
169
170	Runnable decorateTaskIfNecessary(Runnable task) {
171	return this.taskDecorator != null ? this.taskDecorator.decorate(task) : task;	10✔
172	}
173
174	/**
175	* TaskExecutorAdapter subclass that wraps all provided Runnables and Callables
176	* with a JSR-236 ManagedTask, exposing a long-running hint based on
177	* {@link SchedulingAwareRunnable} and an identity name based on the task's
178	* {@code toString()} representation.
179	*/
180	private static class ManagedTaskExecutorAdapter extends TaskExecutorAdapter {
181
182	public ManagedTaskExecutorAdapter(Executor concurrentExecutor) {
183	super(concurrentExecutor);	3✔
184	}	1✔
185
186	@Override
187	public void execute(Runnable task) {
188	super.execute(ManagedTaskBuilder.buildManagedTask(task, task.toString()));	6✔
189	}	1✔
190
191	@Override
192	public Future<Void> submit(Runnable task) {
193	return super.submit(ManagedTaskBuilder.buildManagedTask(task, task.toString()));	×
194	}
195
196	@Override
197	public <T> Future<T> submit(Callable<T> task) {
198	return super.submit(ManagedTaskBuilder.buildManagedTask(task, task.toString()));	×
199	}
200
201	}
202
203	/**
204	* Delegate that wraps a given Runnable/Callable with a JSR-236 ManagedTask,
205	* exposing a long-running hint based on {@link SchedulingAwareRunnable}
206	* and a given identity name.
207	*/
208	protected static class ManagedTaskBuilder {	×
209
210	public static Runnable buildManagedTask(Runnable task, String identityName) {
211	Map<String, String> properties;
212	if (task instanceof SchedulingAwareRunnable sar) {	6!
213	properties = new HashMap<>(4);	5✔
214	properties.put(ManagedTask.LONGRUNNING_HINT, Boolean.toString(sar.isLongLived()));	8✔
215	}
216	else {
217	properties = new HashMap<>(2);	×
218	}
219	properties.put(ManagedTask.IDENTITY_NAME, identityName);	5✔
220	return ManagedExecutors.managedTask(task, properties, null);	5✔
221	}
222
223	public static <T> Callable<T> buildManagedTask(Callable<T> task, String identityName) {
224	Map<String, String> properties = new HashMap<>(2);	×
225	properties.put(ManagedTask.IDENTITY_NAME, identityName);	×
226	return ManagedExecutors.managedTask(task, properties, null);	×
227	}
228	}
229
230	}

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