8308118446

Committed 16 Mar 2024 01:26PM UTC coverage: 78.393% (+0.02%) from 78.374%

Build # 8308118446

Build Type

push

github

Committed by

TAKETODAY

Commit Message

:white_check_mark:

Run Details

63592 of 86119 branches covered (73.84%)

Branch coverage included in aggregate %.

157000 of 195273 relevant lines covered (80.4%)

3.41 hits per line

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

64.62

today-context/src/main/java/cn/taketoday/scheduling/concurrent/ConcurrentTaskExecutor.java

/*
 * Copyright 2017 - 2024 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 cn.taketoday.scheduling.concurrent;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

import cn.taketoday.core.task.AsyncListenableTaskExecutor;
import cn.taketoday.core.task.TaskDecorator;
import cn.taketoday.core.task.support.TaskExecutorAdapter;
import cn.taketoday.lang.Nullable;
import cn.taketoday.scheduling.SchedulingAwareRunnable;
import cn.taketoday.scheduling.SchedulingTaskExecutor;
import cn.taketoday.util.ClassUtils;
import cn.taketoday.util.concurrent.ListenableFuture;
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 cn.taketoday.core.task.TaskExecutor} for it.
 * Also detects an extended {@code java.util.concurrent.ExecutorService}, adapting
 * the {@link cn.taketoday.core.task.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 cn.taketoday.core.task.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 AsyncListenableTaskExecutor, 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<?> submit(Runnable task) {
    return this.adaptedExecutor.submit(task);
  }

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

  @Override
  public ListenableFuture<?> submitListenable(Runnable task) {
    return this.adaptedExecutor.submitListenable(task);
  }

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

  private TaskExecutorAdapter getAdaptedExecutor(Executor originalExecutor) {
    TaskExecutorAdapter 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<?> 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()));
    }

    @Override
    public ListenableFuture<?> submitListenable(Runnable task) {
      return super.submitListenable(ManagedTaskBuilder.buildManagedTask(task, task.toString()));
    }

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

TAKETODAY / today-infrastructure / 8308118446

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