From 0a84d0ead6a7844478ad3b2ca44fca64f33c4979 Mon Sep 17 00:00:00 2001
From: moqui-industrial <729502+moqui-industrial@users.noreply.github.com>
Date: Fri, 17 Jul 2026 18:03:32 +0200
Subject: [PATCH] Add scheduled service API and distributed executor
configuration
---
.../service/ServiceCallScheduledImpl.groovy | 419 ++++++++++++++++++
.../impl/service/ServiceFacadeImpl.groovy | 93 +++-
.../moqui/service/ServiceCallScheduled.java | 109 +++++
.../java/org/moqui/service/ServiceFacade.java | 11 +-
.../src/main/resources/MoquiDefaultConf.xml | 72 ++-
framework/template/XmlActions.groovy.ftl | 35 ++
framework/xsd/moqui-conf-3.xsd | 23 +-
framework/xsd/xml-actions-3.xsd | 88 +++-
8 files changed, 841 insertions(+), 9 deletions(-)
create mode 100644 framework/src/main/groovy/org/moqui/impl/service/ServiceCallScheduledImpl.groovy
create mode 100644 framework/src/main/java/org/moqui/service/ServiceCallScheduled.java
diff --git a/framework/src/main/groovy/org/moqui/impl/service/ServiceCallScheduledImpl.groovy b/framework/src/main/groovy/org/moqui/impl/service/ServiceCallScheduledImpl.groovy
new file mode 100644
index 000000000..920fc7b8b
--- /dev/null
+++ b/framework/src/main/groovy/org/moqui/impl/service/ServiceCallScheduledImpl.groovy
@@ -0,0 +1,419 @@
+/*
+ * This software is in the public domain under CC0 1.0 Universal plus a
+ * Grant of Patent License.
+ *
+ * To the extent possible under law, the author(s) have dedicated all
+ * copyright and related and neighboring rights to this software to the
+ * public domain worldwide. This software is distributed without any
+ * warranty.
+ *
+ * You should have received a copy of the CC0 Public Domain Dedication
+ * along with this software (see the LICENSE.md file). If not, see
+ * .
+ */
+package org.moqui.impl.service
+
+import groovy.transform.CompileStatic
+import org.moqui.Moqui
+import org.moqui.impl.context.ExecutionContextFactoryImpl
+import org.moqui.impl.context.ExecutionContextImpl
+import org.moqui.service.ServiceCallScheduled
+import org.moqui.service.ServiceException
+import org.moqui.service.ServiceCallSync
+
+import org.slf4j.Logger
+import org.slf4j.LoggerFactory
+
+import java.util.concurrent.Callable
+import java.util.concurrent.ScheduledFuture
+import java.util.concurrent.TimeUnit
+
+@CompileStatic
+class ServiceCallScheduledImpl extends ServiceCallImpl implements ServiceCallScheduled {
+ protected final static Logger logger = LoggerFactory.getLogger(ServiceCallScheduledImpl.class)
+
+ protected String taskName = null
+ protected ScheduledServiceInfo scheduledServiceInfo = null
+ protected ScheduledFuture scheduledFuture = null
+ protected boolean isOneShot = true
+ protected boolean isPeriodic = false
+ protected TimeUnit unit = TimeUnit.MILLISECONDS
+ protected Long initialDelay = 0L
+ protected Integer transactionTimeout = null
+ protected Long period
+ protected Long delay
+ protected Long duration
+
+ ServiceCallScheduledImpl(ServiceFacadeImpl sfi) {
+ super(sfi)
+ }
+
+ ServiceCallScheduledImpl(ServiceFacadeImpl sfi, String taskName) {
+ super(sfi)
+ this.taskName(taskName)
+ }
+
+ @Override
+ ServiceCallScheduled name(String serviceName) { serviceNameInternal(serviceName); return this }
+ @Override
+ ServiceCallScheduled name(String v, String n) { serviceNameInternal(null, v, n); return this }
+ @Override
+ ServiceCallScheduled name(String p, String v, String n) { serviceNameInternal(p, v, n); return this }
+
+ @Override
+ ServiceCallScheduled parameters(Map map) { parameters.putAll(map); return this }
+ @Override
+ ServiceCallScheduled parameter(String name, Object value) { parameters.put(name, value); return this }
+
+ @Override
+ String getTaskName() { return this.taskName }
+
+ @Override
+ ServiceCallScheduled taskName(String taskName) {
+ if (taskName == null) throw new IllegalArgumentException("The argument taskName is null.")
+ this.taskName = taskName
+ // try to get and restore existing scheduledFuture by taskName
+ this.scheduledFuture = sfi.getScheduledFuture(taskName)
+ return this
+ }
+
+ @Override
+ ServiceCallScheduled timeUnit(TimeUnit unit) { this.unit = unit; return this }
+
+ @Override
+ ServiceCallScheduled initialDelay(long delay) { this.initialDelay = delay; return this }
+
+ @Override
+ ServiceCallScheduled transactionTimeout(int transactionTimeout) { this.transactionTimeout = transactionTimeout; return this }
+
+ @Override
+ ServiceCallScheduled atFixedRate(long period) { this.period = period; this.delay = null; this.isOneShot = false; this.isPeriodic = true; return this }
+
+ @Override
+ ServiceCallScheduled withFixedDelay(long delay) { this.delay = delay; this.period = null; this.isOneShot = false; this.isPeriodic = true; return this }
+
+ @Override
+ ServiceCallScheduled duration(long duration) { this.duration = duration; return this }
+
+ @Override
+ ServiceCallScheduled call() throws ServiceException {
+ if (!taskName) throw new IllegalStateException("taskName is required for call()")
+
+ // guard: if we already have a running future for this builder, return
+ if (scheduledFuture != null && this.isRunning()) {
+ logger.warn("A previous scheduled service call [${taskName}] is still running. To terminate the previous scheduling use the cancel() method.")
+ return this
+ }
+ // guard: if another task with the same name is already scheduled, reuse it and return
+ ScheduledFuture existing = sfi.getScheduledFuture(taskName)
+ if (existing != null && !existing.isCancelled() && !existing.isDone()) {
+ logger.warn("A previous scheduled service call [${taskName}] is still running. To terminate the previous scheduling use the cancel() method.")
+ this.scheduledFuture = existing
+ return this
+ }
+
+ ExecutionContextFactoryImpl ecfi = sfi.ecfi
+ ExecutionContextImpl eci = ecfi.getEci()
+ validateCall(eci)
+
+ // runs the scheduled service and gets the scheduleFuture
+ ScheduledServiceRunnable runnable
+ if (transactionTimeout != null) {
+ runnable = new ScheduledServiceRunnable(eci, serviceName, parameters, taskName, transactionTimeout)
+ } else {
+ runnable = new ScheduledServiceRunnable(eci, serviceName, parameters, taskName)
+ }
+ scheduledServiceInfo = runnable
+
+ if (isOneShot) {
+ scheduledFuture = sfi.scheduledExecutor.schedule(runnable, initialDelay, unit)
+ } else if (isPeriodic) {
+ if (period != null && period >= 0) {
+ scheduledFuture = sfi.scheduledExecutor.scheduleAtFixedRate(runnable, initialDelay, period, unit)
+ if (duration > 0) cancel(false, duration, unit)
+ } else if (delay != null && delay >= 0) {
+ scheduledFuture = sfi.scheduledExecutor.scheduleWithFixedDelay(runnable, initialDelay, delay, unit)
+ if (duration > 0) cancel(false, duration, unit)
+ } else {
+ throw new IllegalStateException("Periodic scheduling requested but neither period nor delay was set for task [${taskName}].")
+ }
+ } else {
+ throw new IllegalStateException("Schedule mode not set (one-shot/periodic) for task [${taskName}].")
+ }
+
+ if (scheduledFuture == null) throw new IllegalStateException("Scheduler returned null ScheduledFuture for task [${taskName}].")
+
+ sfi.putScheduledFuture(taskName, scheduledFuture)
+ return this
+ }
+
+ @Override
+ ScheduledFuture> callFuture() throws ServiceException {
+ if (!taskName) throw new IllegalStateException("taskName is required for call()")
+
+ // guard: if we already have a running future for this builder, return
+ if (scheduledFuture != null && this.isRunning()) {
+ logger.warn("A previous scheduled service call [${taskName}] is still running. To terminate the previous scheduling use the cancel() method.")
+ return scheduledFuture
+ }
+ // guard: if another task with the same name is already scheduled, reuse it and return
+ ScheduledFuture existing = sfi.getScheduledFuture(taskName)
+ if (existing != null && !existing.isCancelled() && !existing.isDone()) {
+ logger.warn("A previous scheduled service call [${taskName}] is still running. To terminate the previous scheduling use the cancel() method.")
+ this.scheduledFuture = existing
+ return (ScheduledFuture>) existing
+ }
+
+ if (isPeriodic) throw new IllegalStateException("You cannot call the callFuture method for periodic service, with fixed delay or at fixed rate.")
+
+ ExecutionContextFactoryImpl ecfi = sfi.ecfi
+ ExecutionContextImpl eci = ecfi.getEci()
+ validateCall(eci)
+
+ ScheduledServiceCallable callable
+ if (transactionTimeout != null)
+ callable = new ScheduledServiceCallable(eci, serviceName, parameters, taskName, transactionTimeout)
+ else
+ callable = new ScheduledServiceCallable(eci, serviceName, parameters, taskName)
+
+ scheduledServiceInfo = callable
+
+ scheduledFuture = sfi.scheduledExecutor.schedule(callable, initialDelay, unit)
+ if (duration > 0) cancel(false, duration, unit)
+ return scheduledFuture
+ }
+
+ boolean cancel(boolean mayInterruptIfRunning) {
+ if (!taskName) throw new IllegalStateException("taskName is required for cancel()")
+
+ // resolve future from this builder or by taskName (so cancel works from a new builder)
+ ScheduledFuture sf = scheduledFuture
+ if (sf == null) {
+ if (sf == null) sf = sfi.getScheduledFuture(taskName)
+ if (sf == null) {
+ logger.error("No scheduled task found with name [${taskName}] to cancel")
+ return false
+ }
+ // remember it locally so isRunning() etc keep working
+ scheduledFuture = sf
+ }
+
+ boolean cancelled = sf.cancel(mayInterruptIfRunning)
+ if (cancelled) sfi.removeScheduledFuture(taskName)
+ return cancelled
+ }
+
+ ScheduledFuture cancel(boolean mayInterruptIfRunning, long cancelDelay, TimeUnit unit) {
+ if (!taskName) throw new IllegalStateException("taskName is required for cancel(delay)")
+ if (cancelDelay <= 0) throw new IllegalArgumentException("You cannot call the cancel method with an invalid cancelDelay argument.")
+
+ // Resolve the future eagerly so delayed cancellation does not depend on a later cluster lookup by task name.
+ ScheduledFuture sf = scheduledFuture
+ if (sf == null) {
+ sf = sfi.getScheduledFuture(taskName)
+ if (sf != null) scheduledFuture = sf
+ }
+ final ScheduledFuture sfFinal = sf
+ final String tnFinal = taskName
+
+ Runnable runnable = new Runnable() {
+ @Override
+ void run() {
+ try {
+ boolean cancelled = false
+ if (sfFinal != null) {
+ cancelled = sfFinal.cancel(mayInterruptIfRunning)
+ if (cancelled) sfi.removeScheduledFuture(tnFinal)
+ } else {
+ cancelled = cancel(mayInterruptIfRunning)
+ }
+ if (!cancelled) logger.error("No scheduled task found with name [${tnFinal}] to cancel")
+ } catch (Throwable t) {
+ logger.warn("Error cancelling scheduled service [${tnFinal}] from delayed cancellation runnable: ${t.toString()}")
+ }
+ }
+ }
+ return sfi.scheduledExecutor.schedule(runnable, cancelDelay, unit)
+ }
+
+ @Override
+ ScheduledFuture cancel(boolean mayInterruptIfRunning, long cancelDelay) {
+ return cancel(mayInterruptIfRunning, cancelDelay, TimeUnit.MILLISECONDS)
+ }
+
+ @Override
+ boolean isCancelled() {
+ if (scheduledFuture == null) throw new IllegalStateException("Must call method call() or callFuture() before using ScheduledFuture interface methods")
+ return scheduledFuture.isCancelled()
+ }
+
+ @Override
+ boolean isDone() {
+ if (scheduledFuture == null) throw new IllegalStateException("Must call method call() or callFuture() before using ScheduledFuture interface methods")
+ return scheduledFuture.isDone()
+ }
+
+ @Override
+ boolean isRunning() {
+ if (scheduledFuture == null) throw new IllegalStateException("Must call call() or callFuture() before using ScheduledFuture interface methods")
+ return (!scheduledFuture.isCancelled() && !scheduledFuture.isDone())
+ }
+
+ @Override
+ Runnable getRunnable() {
+ if (this.transactionTimeout != null)
+ return new ScheduledServiceRunnable(sfi.ecfi.getEci(), serviceName, parameters, taskName, transactionTimeout)
+ else
+ return new ScheduledServiceRunnable(sfi.ecfi.getEci(), serviceName, parameters, taskName)
+ }
+
+ @Override
+ Callable> getCallable() {
+ if (this.transactionTimeout != null)
+ return new ScheduledServiceCallable(sfi.ecfi.getEci(), serviceName, parameters, taskName, transactionTimeout)
+ else
+ return new ScheduledServiceCallable(sfi.ecfi.getEci(), serviceName, parameters, taskName)
+ }
+
+ static class ScheduledServiceInfo implements Externalizable {
+ transient ExecutionContextFactoryImpl ecfiLocal
+ String threadUsername
+ String serviceName, taskName
+ Map parameters
+ Integer transactionTimeout = null
+
+ ScheduledServiceInfo() { }
+ ScheduledServiceInfo(ExecutionContextImpl eci, String serviceName, Map parameters, String taskName) {
+ ecfiLocal = eci.ecfi
+ threadUsername = eci.userFacade.username
+ this.serviceName = serviceName
+ this.parameters = new HashMap<>(parameters)
+ this.taskName = taskName
+ }
+ ScheduledServiceInfo(ExecutionContextImpl eci, String serviceName, Map parameters, String taskName, int transactionTimeout) {
+ ecfiLocal = eci.ecfi
+ threadUsername = eci.userFacade.username
+ this.serviceName = serviceName
+ this.parameters = new HashMap<>(parameters)
+ this.taskName = taskName
+ this.transactionTimeout = transactionTimeout
+ }
+ ScheduledServiceInfo(ExecutionContextFactoryImpl ecfi, String username, String serviceName, Map parameters, String taskName) {
+ ecfiLocal = ecfi
+ threadUsername = username
+ this.serviceName = serviceName
+ this.parameters = new HashMap<>(parameters)
+ this.taskName = taskName
+ }
+ ScheduledServiceInfo(ExecutionContextFactoryImpl ecfi, String username, String serviceName, Map parameters, String taskName, int transactionTimeout) {
+ ecfiLocal = ecfi
+ threadUsername = username
+ this.serviceName = serviceName
+ this.parameters = new HashMap<>(parameters)
+ this.taskName = taskName
+ this.transactionTimeout = transactionTimeout
+ }
+
+ @Override
+ void writeExternal(ObjectOutput out) throws IOException {
+ out.writeObject(threadUsername) // might be null
+ out.writeUTF(serviceName) // never null
+ out.writeObject(parameters)
+ out.writeUTF(taskName)
+ out.writeInt(transactionTimeout != null ? transactionTimeout.intValue() : 0)
+ }
+
+ @Override
+ void readExternal(ObjectInput objectInput) throws IOException, ClassNotFoundException {
+ threadUsername = (String) objectInput.readObject()
+ serviceName = objectInput.readUTF()
+ parameters = (Map) objectInput.readObject()
+ taskName = objectInput.readUTF()
+ int tx = objectInput.readInt()
+ transactionTimeout = (tx == 0 ? null : Integer.valueOf(tx))
+ }
+
+ ExecutionContextFactoryImpl getEcfi() {
+ if (ecfiLocal == null) ecfiLocal = (ExecutionContextFactoryImpl) Moqui.getExecutionContextFactory()
+ return ecfiLocal
+ }
+
+ Map runInternal() throws Exception {
+ return runInternal(null, false)
+ }
+ Map runInternal(Map parameters, boolean skipEcCheck) throws Exception {
+ ExecutionContextImpl threadEci = (ExecutionContextImpl) null
+ try {
+ // check for active Transaction
+ if (getEcfi().transactionFacade.isTransactionInPlace()) {
+ logger.error("In ServiceCallScheduled service ${serviceName} a transaction is in place for thread ${Thread.currentThread().getName()}, trying to commit")
+ try {
+ getEcfi().transactionFacade.destroyAllInThread()
+ } catch (Exception e) {
+ logger.error("ServiceCallScheduled commit in place transaction failed for thread ${Thread.currentThread().getName()}", e)
+ }
+ }
+ // check for active ExecutionContext
+ if (!skipEcCheck) {
+ ExecutionContextImpl activeEc = getEcfi().activeContext.get()
+ if (activeEc != null) {
+ logger.error("In ServiceCallScheduled service ${serviceName} there is already an ExecutionContext for user ${activeEc.user.username} (from ${activeEc.forThreadId}:${activeEc.forThreadName}) in this thread ${Thread.currentThread().id}:${Thread.currentThread().name}, destroying")
+ try {
+ activeEc.destroy()
+ } catch (Throwable t) {
+ logger.error("Error destroying ExecutionContext already in place in ServiceCallScheduled in thread ${Thread.currentThread().id}:${Thread.currentThread().name}", t)
+ }
+ }
+ }
+
+ threadEci = getEcfi().getEci()
+ if (threadUsername != null && threadUsername.length() > 0) {
+ threadEci.userFacade.internalLoginUser(threadUsername, false)
+ } else {
+ threadEci.userFacade.loginAnonymousIfNoUser()
+ }
+
+ Map parmsToUse = this.parameters
+ if (parameters != null) {
+ parmsToUse = new HashMap<>(this.parameters)
+ parmsToUse.putAll(parameters)
+ }
+
+ // NOTE: authz is disabled because authz is checked before queueing
+ ServiceCallSync serviceCallSync = threadEci.serviceFacade.sync().name(serviceName).parameters(parmsToUse)
+ if (this.transactionTimeout) serviceCallSync.transactionTimeout(transactionTimeout)
+ Map results = serviceCallSync.disableAuthz().call()
+ return results
+ } catch (Throwable t) {
+ logger.error("Error in scheduling service call", t)
+ throw t
+ } finally {
+ if (threadEci != null) threadEci.destroy()
+ }
+ }
+ }
+
+ static class ScheduledServiceRunnable extends ScheduledServiceInfo implements Runnable, Externalizable {
+ ScheduledServiceRunnable() { super() }
+ ScheduledServiceRunnable(ExecutionContextImpl eci, String serviceName, Map parameters, String taskName) {
+ super(eci, serviceName, parameters, taskName)
+ }
+ ScheduledServiceRunnable(ExecutionContextImpl eci, String serviceName, Map parameters, String taskName, int transactionTimeout) {
+ super(eci, serviceName, parameters, taskName, transactionTimeout)
+ }
+ @Override void run() { runInternal() }
+ }
+
+
+ static class ScheduledServiceCallable extends ScheduledServiceInfo implements Callable>, Externalizable {
+ ScheduledServiceCallable() { super() }
+ ScheduledServiceCallable(ExecutionContextImpl eci, String serviceName, Map parameters, String taskName) {
+ super(eci, serviceName, parameters, taskName)
+ }
+ ScheduledServiceCallable(ExecutionContextImpl eci, String serviceName, Map parameters, String taskName, int transactionTimeout) {
+ super(eci, serviceName, parameters, taskName, transactionTimeout)
+ }
+ @Override Map call() throws Exception { return runInternal() }
+ }
+
+}
diff --git a/framework/src/main/groovy/org/moqui/impl/service/ServiceFacadeImpl.groovy b/framework/src/main/groovy/org/moqui/impl/service/ServiceFacadeImpl.groovy
index ea3e6f3f1..37cefc61d 100644
--- a/framework/src/main/groovy/org/moqui/impl/service/ServiceFacadeImpl.groovy
+++ b/framework/src/main/groovy/org/moqui/impl/service/ServiceFacadeImpl.groovy
@@ -1,12 +1,12 @@
/*
* This software is in the public domain under CC0 1.0 Universal plus a
* Grant of Patent License.
- *
+ *
* To the extent possible under law, the author(s) have dedicated all
* copyright and related and neighboring rights to this software to the
* public domain worldwide. This software is distributed without any
* warranty.
- *
+ *
* You should have received a copy of the CC0 Public Domain Dedication
* along with this software (see the LICENSE.md file). If not, see
* .
@@ -64,6 +64,10 @@ class ServiceFacadeImpl implements ServiceFacade {
/** Distributed ExecutorService for async services, etc */
protected ExecutorService distributedExecutorService = null
+ /** An executor for scheduled services */
+ protected CustomScheduledExecutor scheduledExecutor = null
+ /** Map of all serviceCallScheduled scheduledFuture (not concelled yet), using taskName as the key. */
+ protected final ConcurrentMap> scheduledFutureMap = new ConcurrentHashMap<>()
protected final ConcurrentMap> callbackRegistry = new ConcurrentHashMap<>()
@@ -84,6 +88,7 @@ class ServiceFacadeImpl implements ServiceFacade {
restApi = new RestApi(ecfi)
jobWorkerPool = makeWorkerPool()
+ scheduledExecutor = makeScheduledExecutor()
}
private ThreadPoolExecutor makeWorkerPool() {
@@ -106,6 +111,16 @@ class ServiceFacadeImpl implements ServiceFacade {
workQueue, new ContextJavaUtil.JobThreadFactory())
}
+ private CustomScheduledExecutor makeScheduledExecutor() {
+ MNode serviceFacadeNode = ecfi.confXmlRoot.first("service-facade")
+
+ int coreSize = (serviceFacadeNode.attribute("scheduled-thread-pool-core") ?: "16") as int
+ int maxSize = (serviceFacadeNode.attribute("scheduled-thread-pool-max") ?: "32") as int
+ CustomScheduledExecutor executor = new CustomScheduledExecutor(coreSize)
+ executor.setMaximumPoolSize(maxSize)
+ return executor
+ }
+
void postFacadeInit() {
// load Service ECA rules
loadSecaRulesAll()
@@ -164,12 +179,25 @@ class ServiceFacadeImpl implements ServiceFacade {
void destroy() {
// destroy all service runners
for (ServiceRunner sr in serviceRunners.values()) sr.destroy()
+
+ // shutdown scheduled executor
+ try {
+ logger.info("Shutting scheduled executor")
+ scheduledExecutor.shutdown()
+ scheduledExecutor.awaitTermination(30, TimeUnit.SECONDS)
+ if (scheduledExecutor.isTerminated()) logger.info("Scheduled executor shut down and terminated")
+ else logger.warn("Scheduled executor not yet terminated, waited 30 seconds")
+ } catch (Throwable t) {
+ logger.error("Error in scheduledExecutor shutdown", t)
+ }
}
ServiceRunner getServiceRunner(String type) { serviceRunners.get(type) }
// NOTE: this is used in the ServiceJobList screen
ScheduledJobRunner getJobRunner() { jobRunner }
+ long getJobRunnerRate() { jobRunnerRate }
+
boolean isServiceDefined(String serviceName) {
ServiceDefinition sd = getServiceDefinition(serviceName)
if (sd != null) return true
@@ -568,11 +596,72 @@ class ServiceFacadeImpl implements ServiceFacade {
for (EmailEcaRule eer in emecaRuleList) eer.runIfMatches(message, emailServerId, eci)
}
+ /**
+ * True if a task with this name is registered.
+ *
+ * @param taskName Unique task name.
+ * @return true if a name is registerd
+ */
+ boolean hasScheduledFuture(String taskName) {
+ return taskName && scheduledFutureMap.containsKey(taskName)
+ }
+
+ /** Get the scheduled future for a scheduled service call by taskName.
+ * @param taskName Unique task name with which the specified scheduled future is associated.
+ * @return scheduledFuture - scheduledFuture associated with the specified task name.
+ */
+ ScheduledFuture getScheduledFuture(String taskName) {
+ return (ScheduledFuture) scheduledFutureMap.get(taskName)
+ }
+
+ /** Associates the specified scheduled future for a scheduled service call with the specified taskName.
+ * @param taskName Unique task name with which the specified scheduled future is to be associated.
+ * @param scheduledFuture - scheduledFuture to be associated with the specified task name.
+ */
+ ScheduledFuture putScheduledFuture(String taskName, ScheduledFuture scheduledFuture) {
+ if (taskName == null) throw new IllegalArgumentException("The argument taskName is null.")
+ if (scheduledFuture == null) throw new IllegalArgumentException("The argument scheduledFuture is null.")
+ return scheduledFutureMap.put(taskName, scheduledFuture)
+ }
+
+ /** If the specified taskName is not already associated with a scheduled future associates it
+ * with the given instance and returns null, else returns the current scheduled future.
+ * @param taskName Unique task name with which the specified scheduled future is to be associated.
+ * @param scheduledFuture - scheduledFuture to be associated with the specified task name.
+ */
+ ScheduledFuture putScheduledFutureIfAbsent(String taskName, ScheduledFuture scheduledFuture) {
+ if (taskName == null) throw new IllegalArgumentException("The argument taskName is null.")
+ if (scheduledFuture == null) throw new IllegalArgumentException("The argument scheduledFuture is null.")
+ return (ScheduledFuture) scheduledFutureMap.putIfAbsent(taskName, scheduledFuture)
+ }
+
+ /** Removes the entry for the scheduled future and the associated taskName if it is present.
+ * @param taskName Unique task name for which the specified scheduled future is to be removed.
+ */
+ boolean removeScheduledFuture(String taskName) {
+ if (taskName == null) throw new IllegalArgumentException("The argument taskName is null.")
+ return scheduledFutureMap.remove(taskName)
+ }
+
+ /** Snapshot of current scheduled future task names. */
+ List listScheduledFutureTaskNames() {
+ return new ArrayList<>(scheduledFutureMap.keySet())
+ }
+
+ /** Current scheduled future count. */
+ int scheduledFutureCount() {
+ return scheduledFutureMap.size()
+ }
+
@Override
ServiceCallSync sync() { return new ServiceCallSyncImpl(this) }
@Override
ServiceCallAsync async() { return new ServiceCallAsyncImpl(this) }
@Override
+ ServiceCallScheduled schedule(String taskName) { return new ServiceCallScheduledImpl(this, taskName) }
+ @Override
+ ServiceCallScheduled schedule() { return new ServiceCallScheduledImpl(this) }
+ @Override
ServiceCallJob job(String jobName) { return new ServiceCallJobImpl(jobName, this) }
@Override
diff --git a/framework/src/main/java/org/moqui/service/ServiceCallScheduled.java b/framework/src/main/java/org/moqui/service/ServiceCallScheduled.java
new file mode 100644
index 000000000..a890393bd
--- /dev/null
+++ b/framework/src/main/java/org/moqui/service/ServiceCallScheduled.java
@@ -0,0 +1,109 @@
+/*
+ * This software is in the public domain under CC0 1.0 Universal plus a
+ * Grant of Patent License.
+ *
+ * To the extent possible under law, the author(s) have dedicated all
+ * copyright and related and neighboring rights to this software to the
+ * public domain worldwide. This software is distributed without any
+ * warranty.
+ *
+ * You should have received a copy of the CC0 Public Domain Dedication
+ * along with this software (see the LICENSE.md file). If not, see
+ * .
+ */
+package org.moqui.service;
+
+import java.util.Map;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+
+
+/** Example use:
+ * ec.service.schedule().name("...").parameters([...]).initialDelay(50).atFixedRate(100).call()
+ * ec.service.schedule().name("...").parameters([...]).initialDelay(50).callFuture()
+ * ec.service.schedule().name("....").getScheduledFuture(urn).shutdown()
+ */
+@SuppressWarnings("unused")
+public interface ServiceCallScheduled extends ServiceCall {
+ /** Name of the service to run. The combined service name, like: "${path}.${verb}${noun}". To explicitly separate
+ * the verb and noun put a hash (#) between them, like: "${path}.${verb}#${noun}" (this is useful for calling the
+ * implicit entity CrUD services where verb is create, update, or delete and noun is the name of the entity).
+ */
+ ServiceCallScheduled name(String serviceName);
+
+ ServiceCallScheduled name(String verb, String noun);
+
+ ServiceCallScheduled name(String path, String verb, String noun);
+
+ /** Map of name, value pairs that make up the context (in parameters) passed to the service. */
+ ServiceCallScheduled parameters(Map context);
+
+ /** Single name, value pairs to put in the context (in parameters) passed to the service. */
+ ServiceCallScheduled parameter(String name, Object value);
+
+ /** Returns the name of the task for the scheduled service. */
+ String getTaskName();
+
+ /** Set a unique task name for the scheduled service. */
+ ServiceCallScheduled taskName(String taskName);
+
+ /** The time unit for all time parameters. Defaults to milliseconds. */
+ ServiceCallScheduled timeUnit(TimeUnit unit);
+
+ /** The service execution becomes enabled after the given initial delay. Defaults to 0. */
+ ServiceCallScheduled initialDelay(long delay) throws IllegalArgumentException;
+
+ /** The maximum time to wait before timeout for each service call. Override the transaction-timeout attribute in the service definition. */
+ ServiceCallScheduled transactionTimeout(int transactionTimeout);
+
+ /** Creates a periodic service that becomes enabled first after the given initial delay, and subsequently with the given period; that is executions will commence after initialDelay then initialDelay+period, then initialDelay + 2 * period, and so on. */
+ ServiceCallScheduled atFixedRate(long period) throws IllegalArgumentException;
+
+ /** Creates a periodic service that becomes enabled first after the given initial delay, and subsequently with the given delay between the termination of one execution and the commencement of the next. */
+ ServiceCallScheduled withFixedDelay(long delay) throws IllegalArgumentException;
+
+ /** The duration of the execution of the periodic task. After this duration the cyclical task will be canceled. */
+ ServiceCallScheduled duration(long duration);
+
+ /**
+ * Submits the service for the scheduled execution, ignoring the result.
+ * This effectively calls the service through a java.lang.Runnable implementation.
+ */
+ ServiceCallScheduled call() throws ServiceException;
+
+ /**
+ * Submits the service for the scheduled execution and get a java.util.concurrent.ScheduledFuture object back so you can wait for the service to
+ * complete and get the result.
+ *
+ * This effectively calls the service through a java.util.concurrent.Callable implementation.
+ */
+ ScheduledFuture> callFuture() throws ServiceException;
+
+ /** Attempts to cancel the scheduled execution of this service. */
+ boolean cancel(boolean mayInterruptIfRunning);
+
+ /** Attempts to cancel the scheduled execution of this service after the specified delay.
+ * Returns the SheduledFuture of the cancel task.
+ */
+ ScheduledFuture cancel(boolean mayInterruptIfRunning, long cancelDelay, TimeUnit unit);
+
+ /** Attempts to cancel the scheduled execution of this service after the specified delay.
+ * Returns the SheduledFuture of the cancel task.
+ */
+ ScheduledFuture cancel(boolean mayInterruptIfRunning, long cancelDelay);
+
+ /** Returns true if this scheduled service was cancelled before it completed normally. */
+ boolean isCancelled();
+
+ /** Returns true if this scheduled service completed. Completion may be due to normal termination, an exception, or cancellation -- in all of these cases, this method will return true. */
+ boolean isDone();
+
+ /** Returns true if this scheduled service has not completed. */
+ boolean isRunning();
+
+ /** Get a Runnable object to do this service call through an ExecutorService or other runner of your choice. */
+ Runnable getRunnable();
+ /** Get a Callable object to do this service call through an ExecutorService of your choice. */
+ Callable> getCallable();
+}
diff --git a/framework/src/main/java/org/moqui/service/ServiceFacade.java b/framework/src/main/java/org/moqui/service/ServiceFacade.java
index f368c3b14..2f675a8d8 100644
--- a/framework/src/main/java/org/moqui/service/ServiceFacade.java
+++ b/framework/src/main/java/org/moqui/service/ServiceFacade.java
@@ -1,12 +1,12 @@
/*
* This software is in the public domain under CC0 1.0 Universal plus a
* Grant of Patent License.
- *
+ *
* To the extent possible under law, the author(s) have dedicated all
* copyright and related and neighboring rights to this software to the
* public domain worldwide. This software is distributed without any
* warranty.
- *
+ *
* You should have received a copy of the CC0 Public Domain Dedication
* along with this software (see the LICENSE.md file). If not, see
* .
@@ -16,6 +16,7 @@
import org.moqui.util.RestClient;
import java.util.Map;
+import java.util.concurrent.ScheduledFuture;
/** ServiceFacade Interface */
@SuppressWarnings("unused")
@@ -27,6 +28,12 @@ public interface ServiceFacade {
/** Get a service caller to call a service asynchronously. */
ServiceCallAsync async();
+ /** Get a service caller to schedule a service call. */
+ ServiceCallScheduled schedule();
+
+ /** Get a service caller to schedule a service, assigning a specific taskName. */
+ ServiceCallScheduled schedule(String taskName);
+
/**
* Get a service caller to call a service job.
*
diff --git a/framework/src/main/resources/MoquiDefaultConf.xml b/framework/src/main/resources/MoquiDefaultConf.xml
index d8dc14c59..6ebdbb3e1 100644
--- a/framework/src/main/resources/MoquiDefaultConf.xml
+++ b/framework/src/main/resources/MoquiDefaultConf.xml
@@ -42,6 +42,8 @@
+
+
@@ -50,6 +52,32 @@
+
+
+
+
+
+
+
@@ -68,6 +96,7 @@
+
@@ -82,6 +111,9 @@
+
+
+
/kibana/*
+
+
+ /grafana/*
+
@@ -234,6 +270,14 @@
/kibana/*
+
+
+
+
+
+ /grafana/*
+
@@ -385,7 +429,8 @@
macro-template-location="template/screen-macro/DefaultScreenMacros.plain.ftl"/>
-
@@ -427,7 +472,8 @@
+ crypt-salt="20201202" crypt-iter="10" crypt-algo="PBEWithHmacSHA256AndAES_128" crypt-pass="${entity_ds_crypt_pass}"
+ worker-pool-queue="1024" worker-pool-core="5" worker-pool-max="100" worker-pool-alive="60">
@@ -490,6 +536,8 @@
databaseName="${entity_ds_c1_database}" user="${entity_ds_c1_user}" password="${entity_ds_c1_password}"
pinGlobalTxToPhysicalConnection="true" autoReconnectForPools="true" useUnicode="true" encoding="UTF-8"/>
+
+
@@ -859,6 +907,26 @@
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/framework/template/XmlActions.groovy.ftl b/framework/template/XmlActions.groovy.ftl
index 3d2a1555b..b676544d6 100644
--- a/framework/template/XmlActions.groovy.ftl
+++ b/framework/template/XmlActions.groovy.ftl
@@ -17,6 +17,7 @@ import static org.moqui.util.StringUtilities.*
import java.sql.Timestamp
import java.sql.Time
import java.time.*
+import java.util.concurrent.*
// these are in the context by default: ExecutionContext ec, Map context, Map result
<#visit xmlActionsRoot/>
@@ -72,6 +73,40 @@ return;
}
+<#macro "service-schedule">
+ <#assign timeUnit = "MILLISECONDS">
+ <#if .node["@time-unit"]?has_content><#assign timeUnit = .node["@time-unit"]?upper_case>
+ if (true) {
+ ec.service.schedule("${.node["@task-name"]}")<#rt>
+ <#t>.name("${.node.@name}")
+ <#t>.timeUnit(java.util.concurrent.TimeUnit.${timeUnit})
+ <#t><#if .node["@initial-delay"]?has_content>.initialDelay(${.node["@initial-delay"]!0})
+ <#t><#if .node["@polling-interval"]?has_content && .node["@type"]?has_content && .node["@type"] == "at-fixed-rate">.atFixedRate(${.node["@polling-interval"]})
+ <#t><#if .node["@polling-interval"]?has_content && .node["@type"]?has_content && .node["@type"] == "with-fixed-delay">.withFixedDelay(${.node["@polling-interval"]})
+ <#t><#if .node["@duration"]?has_content>.duration(${.node["@duration"]})
+ <#t><#if .node["@transaction-timeout"]?has_content>.transactionTimeout(${.node["@transaction-timeout"]})
+ <#if .node["@in-map"]?if_exists == "true">.parameters(context)<#elseif .node["@in-map"]?has_content && .node["@in-map"] != "false">.parameters(${.node["@in-map"]})<#list .node["field-map"] as fieldMap>.parameter("${fieldMap["@field-name"]}",<#if fieldMap["@from"]?has_content>${fieldMap["@from"]}<#elseif fieldMap["@value"]?has_content>"""${fieldMap["@value"]}"""<#else>${fieldMap["@field-name"]}).call()
+ <#if (.node["@ignore-error"]?if_exists == "true")>
+ if (ec.message.hasError()) {
+ ec.logger.warn("Ignoring error running service ${.node.@name}: " + ec.message.getErrorsString())
+ ec.message.clearErrors()
+ }
+ <#else>
+ if (ec.message.hasError()) return
+ <#t>
+ }
+
+
+<#macro "cancel-scheduled-service">
+ <#assign timeUnit = "MILLISECONDS">
+ <#if .node["@time-unit"]?has_content><#assign timeUnit = .node["@time-unit"]?upper_case>
+ if (true) {
+ ec.service.schedule("${.node["@task-name"]}")<#rt><#if .node["@cancel-delay"]?has_content>
+ <#t>.cancel(<#if .node["@may-interrupt-if-running"]?if_exists == "true">true<#else>false, ${.node["@cancel-delay"]}, java.util.concurrent.TimeUnit.${timeUnit})<#else>
+ <#t>.cancel(<#if .node["@may-interrupt-if-running"]?if_exists == "true">true<#else>false)
+ }
+
+
<#macro "script"><#if .node["@location"]?has_content>ec.resource.script("${.node["@location"]}", null)
// begin inline script
${.node}
diff --git a/framework/xsd/moqui-conf-3.xsd b/framework/xsd/moqui-conf-3.xsd
index d060a6176..253271703 100644
--- a/framework/xsd/moqui-conf-3.xsd
+++ b/framework/xsd/moqui-conf-3.xsd
@@ -77,7 +77,11 @@ along with this software (see the LICENSE.md file). If not, see
The maximum size of the worker thread pool.
The amount of time, in seconds, to keep idle worker threads alive (beyond core pool size).
-
+
+ The core (minimum) size of the scheduled thread pool.
+
+ The maximum size of the scheduled thread pool.
+
The ToolFactory to use to get a SimpleTopic for distributed NotificationMessage
@@ -543,9 +547,16 @@ along with this software (see the LICENSE.md file). If not, see
-
+
The name of the ToolFactory to use for the distributed async service ExecutorService implementation.
+
+ The name of the ToolFactory to use for the distributed scheduled service ExecutorService implementation.
+
+
+ The core (minimum) size of the scheduled thread pool.
+
+ The maximum size of the scheduled thread pool.
How often to check for and run scheduled service jobs in seconds. Set to 0 (zero) to disable.
@@ -640,6 +651,14 @@ along with this software (see the LICENSE.md file). If not, see
+
+ The maximum size of the statement worker queue.
+
+ The core (minimum) size of the statement worker thread pool.
+
+ The maximum size of the statement worker thread pool.
+
+ The amount of time, in seconds, to keep idle statement worker threads alive (beyond core pool size).
diff --git a/framework/xsd/xml-actions-3.xsd b/framework/xsd/xml-actions-3.xsd
index 01587c0f6..41f01f282 100644
--- a/framework/xsd/xml-actions-3.xsd
+++ b/framework/xsd/xml-actions-3.xsd
@@ -185,6 +185,92 @@ along with this software (see the LICENSE.md file). If not, see
-->
+
+ Schedule a service execution.
+
+
+
+
+
+
+ The combined service name, like: "${path}.${verb}${noun}". To explicitly separate the verb and noun
+ put a hash (#) between them, like: "${path}.${verb}#${noun}".
+
+
+
+
+ Creates an in parameters with variables matching the names of service in-parameters elements, doing
+ type conversions as needed.
+
+ If false (default) does nothing. If true constructs an in-map from the context.
+ Otherwise is the name of a Map in the context uses it as the source Map for the service context.
+
+
+
+ Set a unique task name for the scheduled service.
+
+
+
+
+ The scheduling type used to run the task. Possible values:
+ - one-shot
+ Schedules and executes a one-shot service that becomes enabled after the given delay.
+ - at-fixed-rate
+ Schedules a periodic service that becomes enabled first after the given initial delay, and subsequently with the given period; that is executions will commence after initialDelay then initialDelay+period, then initialDelay + 2 * period, and so on.
+ - with-fixed-delay
+ Schedules a periodic service that becomes enabled first after the given initial delay, and subsequently with the given delay between the termination of one execution and the commencement of the next.
+
+
+
+
+
+
+
+
+
+
+
+ The time unit, in java.util.concurrent.TimeUnit format, for all time parameters. Defaults to milliseconds.
+
+
+ The task becomes enabled after the given initial delay. Defaults to 0.
+
+
+ Configure a cyclical service. The service is repeated, with the given polling interval,
+ at fixed rate or with a fixed delay between the termination of one execution and the commencement of the next.
+
+
+ The duration of the execution of the periodic task. After this interval the cyclical task will be canceled.
+
+
+
+ Defines the timeout for the transaction, in seconds.
+ This value is only used if this service begins a transaction (either require-new, or
+ use-or-begin and there is no other transaction already in place).
+
+
+
+
+
+
+ Attempts to cancel the scheduled execution of the service given the task name.
+ Do nothing if this scheduled service was cancelled before it completed normally.
+
+
+
+ Set a unique task name to get the scheduled service.
+
+
+ If true the service should be interrupted; otherwise, in-progress services are allowed to complete.
+
+
+ Cancel the execution of the periodic task after the specified delay. If omitted cancels immediately. Must be greater than 0 when specified.
+
+
+ The time unit, in java.util.concurrent.TimeUnit format, for all time parameters. Defaults to milliseconds.
+
+
+
@@ -419,7 +505,7 @@ along with this software (see the LICENSE.md file). If not, see
The fields and special fields with suffixes supported are the same as the *-find fields in the XML
Forms. This means that you can use this to process the data from the various inputs generated by XML
- Forms. The suffixes include things like *_op for operators and *_ic for ignore case.
+ Forms. The suffixes include things like *_op for operators and *_ic for ignore case.
For historical reference, this does basically what the Apache OFBiz prepareFind service does.