From 48e154b400d64286ed81c115babdca530c20a511 Mon Sep 17 00:00:00 2001 From: Moqui Industrial <729502+moqui-industrial@users.noreply.github.com> Date: Thu, 14 May 2026 11:16:04 +0200 Subject: [PATCH 1/2] Add and XML-actions New XML actions for high-frequency device polling and event-driven scheduled tasks: Backed by a dedicated CustomScheduledExecutor (separate from the job runner) so short periods (50 ms, 100 ms) do not starve long-running batch jobs. ServiceCallScheduled fluent API mirrors ServiceCallSync/Async. scheduledFutureMap (ConcurrentHashMap) tracks active futures by taskName for cancel-by-name support. ServiceFacadeImpl.schedule() factory methods wire everything together and shut the executor down cleanly on destroy(). Usage example in example/service/ExampleServices.xml. --- .../org/moqui/impl/actions/XmlAction.java | 27 +- .../runner/XmlActionsScriptRunner.groovy | 55 +- .../service/ServiceCallScheduledImpl.groovy | 473 ++++++++++++++++++ .../impl/service/ServiceFacadeImpl.groovy | 162 +++++- .../moqui/service/ServiceCallScheduled.java | 120 +++++ .../java/org/moqui/service/ServiceFacade.java | 7 + framework/template/XmlActions.groovy.ftl | 36 ++ framework/xsd/service-definition-3.xsd | 1 + framework/xsd/xml-actions-3.xsd | 89 ++++ 9 files changed, 923 insertions(+), 47 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/actions/XmlAction.java b/framework/src/main/groovy/org/moqui/impl/actions/XmlAction.java index b837c6fa6..c4462ea77 100644 --- a/framework/src/main/groovy/org/moqui/impl/actions/XmlAction.java +++ b/framework/src/main/groovy/org/moqui/impl/actions/XmlAction.java @@ -28,6 +28,7 @@ import java.io.StringWriter; import java.io.Writer; import java.util.HashMap; +import java.util.concurrent.locks.ReentrantLock; import java.util.Map; public class XmlAction { @@ -39,6 +40,7 @@ public class XmlAction { protected final String location; /** The Groovy class compiled from the script transformed from the XML actions text using the FTL template. */ private Class groovyClassInternal = null; + private final ReentrantLock groovyClassLock = new ReentrantLock(); public XmlAction(ExecutionContextFactoryImpl ecfi, MNode xmlNode, String location) { this.ecfi = ecfi; @@ -93,18 +95,21 @@ public Class getGroovyClass() { if (groovyClassInternal != null) return groovyClassInternal; return makeGroovyClass(); } - protected synchronized Class makeGroovyClass() { - if (groovyClassInternal != null) return groovyClassInternal; - String curGroovy = getGroovyString(); - // if (logger.isTraceEnabled()) logger.trace("Xml Action [${location}] groovyString: ${curGroovy}") + protected Class makeGroovyClass() { + groovyClassLock.lock(); try { - groovyClassInternal = ecfi.compileGroovy(curGroovy, StringUtilities.cleanStringForJavaName(location)); - } catch (Throwable t) { - groovyClassInternal = null; - logger.error("Error parsing groovy String at [" + location + "]:\n" + writeGroovyWithLines() + "\n"); - throw t; - } - return groovyClassInternal; + if (groovyClassInternal != null) return groovyClassInternal; + String curGroovy = getGroovyString(); + // if (logger.isTraceEnabled()) logger.trace("Xml Action [${location}] groovyString: ${curGroovy}") + try { + groovyClassInternal = ecfi.compileGroovy(curGroovy, StringUtilities.cleanStringForJavaName(location)); + } catch (Throwable t) { + groovyClassInternal = null; + logger.error("Error parsing groovy String at [" + location + "]:\n" + writeGroovyWithLines() + "\n"); + throw t; + } + return groovyClassInternal; + } finally { groovyClassLock.unlock(); } } public String getGroovyString() { diff --git a/framework/src/main/groovy/org/moqui/impl/context/runner/XmlActionsScriptRunner.groovy b/framework/src/main/groovy/org/moqui/impl/context/runner/XmlActionsScriptRunner.groovy index 872a62dfb..03d7fa77b 100644 --- a/framework/src/main/groovy/org/moqui/impl/context/runner/XmlActionsScriptRunner.groovy +++ b/framework/src/main/groovy/org/moqui/impl/context/runner/XmlActionsScriptRunner.groovy @@ -25,6 +25,7 @@ import org.slf4j.Logger import org.slf4j.LoggerFactory import javax.cache.Cache +import java.util.concurrent.locks.ReentrantLock @CompileStatic class XmlActionsScriptRunner implements ScriptRunner { @@ -32,7 +33,9 @@ class XmlActionsScriptRunner implements ScriptRunner { protected ExecutionContextFactoryImpl ecfi protected Cache scriptXmlActionLocationCache - protected Template xmlActionsTemplate = null + protected volatile Template xmlActionsTemplate = null + private final ReentrantLock loadXmlActionLock = new ReentrantLock() + private final ReentrantLock makeTemplateLock = new ReentrantLock() XmlActionsScriptRunner() { } @@ -54,34 +57,40 @@ class XmlActionsScriptRunner implements ScriptRunner { if (xa == null) xa = loadXmlAction(location) return xa } - protected synchronized XmlAction loadXmlAction(String location) { - XmlAction xa = (XmlAction) scriptXmlActionLocationCache.get(location) - if (xa == null) { - xa = new XmlAction(ecfi, ecfi.resourceFacade.getLocationText(location, false), location) - scriptXmlActionLocationCache.put(location, xa) - } - return xa + protected XmlAction loadXmlAction(String location) { + loadXmlActionLock.lock() + try { + XmlAction xa = (XmlAction) scriptXmlActionLocationCache.get(location) + if (xa == null) { + xa = new XmlAction(ecfi, ecfi.resourceFacade.getLocationText(location, false), location) + scriptXmlActionLocationCache.put(location, xa) + } + return xa + } finally { loadXmlActionLock.unlock() } } Template getXmlActionsTemplate() { if (xmlActionsTemplate == null) makeXmlActionsTemplate() return xmlActionsTemplate } - protected synchronized void makeXmlActionsTemplate() { - if (xmlActionsTemplate != null) return - - String templateLocation = ecfi.confXmlRoot.first("resource-facade").attribute("xml-actions-template-location") - Template newTemplate = null - Reader templateReader = null + protected void makeXmlActionsTemplate() { + makeTemplateLock.lock() try { - templateReader = new InputStreamReader(ecfi.resourceFacade.getLocationStream(templateLocation)) - newTemplate = new Template(templateLocation, templateReader, - ecfi.resourceFacade.ftlTemplateRenderer.getFtlConfiguration()) - } catch (Exception e) { - logger.error("Error while initializing XMLActions template at [${templateLocation}]", e) - } finally { - if (templateReader != null) templateReader.close() - } - xmlActionsTemplate = newTemplate + if (xmlActionsTemplate != null) return + + String templateLocation = ecfi.confXmlRoot.first("resource-facade").attribute("xml-actions-template-location") + Template newTemplate = null + Reader templateReader = null + try { + templateReader = new InputStreamReader(ecfi.resourceFacade.getLocationStream(templateLocation)) + newTemplate = new Template(templateLocation, templateReader, + ecfi.resourceFacade.ftlTemplateRenderer.getFtlConfiguration()) + } catch (Exception e) { + logger.error("Error while initializing XMLActions template at [${templateLocation}]", e) + } finally { + if (templateReader != null) templateReader.close() + } + xmlActionsTemplate = newTemplate + } finally { makeTemplateLock.unlock() } } } 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..0f4c3d099 --- /dev/null +++ b/framework/src/main/groovy/org/moqui/impl/service/ServiceCallScheduledImpl.groovy @@ -0,0 +1,473 @@ +/* + * 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 distribute = false + 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 distribute(boolean dist) { this.distribute = dist; 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 + + boolean useDistributed = (distribute && sfi.distributedScheduledExecutorService != null) + + if (isOneShot) { + if (useDistributed) { + scheduledFuture = sfi.distributedScheduledExecutorService.schedule( + sfi.distributedScheduledExecutorService.decorateTask(taskName, runnable), initialDelay, unit) + } else { + scheduledFuture = sfi.scheduledExecutor.schedule(runnable, initialDelay, unit) + } + } else if (isPeriodic) { + if (period != null && period >= 0) { + if (useDistributed) { + scheduledFuture = sfi.distributedScheduledExecutorService.scheduleAtFixedRate( + sfi.distributedScheduledExecutorService.decorateTask(taskName, runnable), initialDelay, period, unit) + if (duration > 0) cancel(false, duration, unit) + } else { + scheduledFuture = sfi.scheduledExecutor.scheduleAtFixedRate(runnable, initialDelay, period, unit) + if (duration > 0) cancel(false, duration, unit) + } + } else if (delay != null && delay >= 0) { + if (useDistributed) { + try { + scheduledFuture = sfi.distributedScheduledExecutorService.scheduleWithFixedDelay( + sfi.distributedScheduledExecutorService.decorateTask(taskName, runnable), initialDelay, delay, unit) + if (duration > 0) cancel(false, duration, unit) + // done on distributed scheduler + if (scheduledFuture != null) { + if (logger.traceEnabled) logger.trace("Scheduled distributed fixed-delay service [${taskName}]") + // for distributed scheduled services we rely on remote registry. However, the task is also saved locally. + sfi.putScheduledFuture(taskName, scheduledFuture) + return this + } + } catch (UnsupportedOperationException | AbstractMethodError e) { + logger.warn("Method scheduleWithFixedDelay is not supported by distributed scheduler ExecutorService. The service call [${taskName}] will be scheduled local only.") + } + } + 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}].") + + // for distributed scheduled services we rely on remote registry. However, the task is also saved locally. + 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) + + boolean useDistributed = (distribute && sfi.distributedScheduledExecutorService != null) + + ScheduledServiceCallable callable + if (transactionTimeout != null) + callable = new ScheduledServiceCallable(eci, serviceName, parameters, taskName, transactionTimeout) + else + callable = new ScheduledServiceCallable(eci, serviceName, parameters, taskName) + + scheduledServiceInfo = callable + + if (useDistributed) { + scheduledFuture = sfi.distributedScheduledExecutorService.schedule( + sfi.distributedScheduledExecutorService.decorateTask(taskName, callable), initialDelay, unit) + if (duration > 0) cancel(false, duration, unit) + } else { + 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) { + // try to get the scheduledFuture instance from the distributed scheduled executor service + if (sfi.distributedScheduledExecutorService != null) { + sf = sfi.distributedScheduledExecutorService.getScheduledFuture(taskName) + } + // fallback to local registry + 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 + + // Always schedule cancellation locally. The resolved future may still be a distributed proxy. + 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) { + if ("com.hazelcast.scheduledexecutor.StaleTaskException".equals(t.getClass().getName())) { + // The distributed task may already be gone on the owner node by the time the local cancel timer fires. + sfi.removeScheduledFuture(tnFinal) + if (logger.infoEnabled) logger.info("Scheduled service [${tnFinal}] was already gone when delayed cancellation ran") + } else { + 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..914ff1ce3 100644 --- a/framework/src/main/groovy/org/moqui/impl/service/ServiceFacadeImpl.groovy +++ b/framework/src/main/groovy/org/moqui/impl/service/ServiceFacadeImpl.groovy @@ -19,6 +19,7 @@ import org.moqui.impl.context.ArtifactExecutionInfoImpl import org.moqui.impl.context.ArtifactExecutionInfoImpl.ArtifactTypeStats import org.moqui.impl.context.ContextJavaUtil import org.moqui.impl.context.ContextJavaUtil.CustomScheduledExecutor +import org.moqui.impl.context.ContextJavaUtil.CustomDistributedScheduledExecutor import org.moqui.resource.ResourceReference import org.moqui.context.ToolFactory import org.moqui.impl.context.ExecutionContextFactoryImpl @@ -59,11 +60,19 @@ class ServiceFacadeImpl implements ServiceFacade { protected final Map serviceRunners = new HashMap<>() private ScheduledJobRunner jobRunner = null - public final ThreadPoolExecutor jobWorkerPool - private LoadRunner loadRunner = null + public final ExecutorService jobWorkerPool + private volatile LoadRunner loadRunner = null + private final ReentrantLock loadRunnerLock = new ReentrantLock() + private long jobRunnerRate = 0L /** Distributed ExecutorService for async services, etc */ protected ExecutorService distributedExecutorService = null + /** An executor for scheduled services */ + protected CustomScheduledExecutor scheduledExecutor = null + /** Distributed ExecutorService for scheduled services */ + protected CustomDistributedScheduledExecutor distributedScheduledExecutorService = 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,9 +93,10 @@ class ServiceFacadeImpl implements ServiceFacade { restApi = new RestApi(ecfi) jobWorkerPool = makeWorkerPool() + scheduledExecutor = makeScheduledExecutor() } - private ThreadPoolExecutor makeWorkerPool() { + private ExecutorService makeWorkerPool() { MNode serviceFacadeNode = ecfi.confXmlRoot.first("service-facade") int jobQueueMax = (serviceFacadeNode.attribute("job-queue-max") ?: "0") as int @@ -99,11 +109,19 @@ class ServiceFacadeImpl implements ServiceFacade { } long aliveTime = (serviceFacadeNode.attribute("worker-pool-alive") ?: "120") as long - logger.info("Initializing Service Job ThreadPoolExecutor: queue limit ${jobQueueMax}, pool-core ${coreSize}, pool-max ${maxSize}, pool-alive ${aliveTime}s") - // make the actual queue at least maxSize to allow for stuffing the queue to get it to add threads to the pool + logger.info("Initializing Service Job worker pool with Virtual Threads (MoquiJob): queue limit ${jobQueueMax}, pool-core ${coreSize}, pool-max ${maxSize}, pool-alive ${aliveTime}s") BlockingQueue workQueue = new LinkedBlockingQueue<>(jobQueueMax < maxSize ? maxSize : jobQueueMax) - return new ContextJavaUtil.WorkerThreadPoolExecutor(ecfi, coreSize, maxSize, aliveTime, TimeUnit.SECONDS, - workQueue, new ContextJavaUtil.JobThreadFactory()) + return new ContextJavaUtil.VirtualThreadExecutorService(ecfi, "MoquiJob", coreSize, maxSize, aliveTime, TimeUnit.SECONDS, workQueue) + } + + 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() { @@ -115,7 +133,7 @@ class ServiceFacadeImpl implements ServiceFacade { MNode serviceFacadeNode = ecfi.confXmlRoot.first("service-facade") // get distributed ExecutorService - String distEsFactoryName = serviceFacadeNode.attribute("distributed-factory") + String distEsFactoryName = serviceFacadeNode.attribute("distributed-executor-factory") if (distEsFactoryName) { logger.info("Getting Async Distributed Service ExecutorService (using ToolFactory ${distEsFactoryName})") ToolFactory esToolFactory = ecfi.getToolFactory(distEsFactoryName) @@ -126,12 +144,28 @@ class ServiceFacadeImpl implements ServiceFacade { distributedExecutorService = esToolFactory.getInstance() } } else { - logger.info("No distributed-factory specified, distributed async service calls will be run local only") + logger.info("No distributed-executor-factory specified, distributed async service calls will be run local only") distributedExecutorService = null } + // get distributed scheduled ExecutorService + String distSchedEsFactoryName = serviceFacadeNode.attribute("distributed-scheduled-executor-factory") + if (distSchedEsFactoryName) { + logger.info("Getting Distributed Scheduled Service ExecutorService (using ToolFactory ${distSchedEsFactoryName})") + ToolFactory sesToolFactory = ecfi.getToolFactory(distSchedEsFactoryName) + if (sesToolFactory == null) { + logger.warn("Could not find ExecutorService ToolFactory with name ${distSchedEsFactoryName}, distributed scheduled service calls will be run local only") + distributedScheduledExecutorService = null + } else { + distributedScheduledExecutorService = sesToolFactory.getInstance() + } + } else { + logger.info("No distributed-scheduled-executor-factory specified, distributed scheduled service calls will be run local only") + distributedScheduledExecutorService = null + } + // setup service job runner - long jobRunnerRate = (serviceFacadeNode.attribute("scheduled-job-check-time") ?: "60") as long + jobRunnerRate = (serviceFacadeNode.attribute("scheduled-job-check-time") ?: "60") as long if (jobRunnerRate > 0L) { // wait before first run to make sure all is loaded and we're past an initial activity burst long initialDelay = 120L @@ -150,6 +184,11 @@ class ServiceFacadeImpl implements ServiceFacade { this.distributedExecutorService = executorService } + void setDistributedScheduledExecutorService(CustomDistributedScheduledExecutor executorService) { + logger.info("Setting DistributedScheduledExecutorService to ${executorService.class.name}, was ${this.distributedScheduledExecutorService?.class?.name}") + this.distributedScheduledExecutorService = executorService + } + void warmCache() { logger.info("Warming cache for all service definitions") long startTime = System.currentTimeMillis() @@ -164,12 +203,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 +620,89 @@ 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) { + // try to get the scheduledFuture instance from the distributed scheduled executor service + if (distributedScheduledExecutorService != null) { + ScheduledFuture scheduledFuture = distributedScheduledExecutorService.getScheduledFuture(taskName) + if (scheduledFuture != null) { + // update the local map + scheduledFutureMap.put(taskName, scheduledFuture) + return scheduledFuture + } + } + // if was not found in the distributed scheduled executor service, try to get the local scheduledFuture instance + 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.") + // try to get the scheduledFuture instance from the distributed scheduled executor service + if (distributedScheduledExecutorService) { + ScheduledFuture scheduledFuture = distributedScheduledExecutorService.getScheduledFuture(taskName) + // if it was found call the dispose method + if (scheduledFuture != null) distributedScheduledExecutorService.dispose(scheduledFuture) + } + // remove the local entry + 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 @@ -613,9 +743,15 @@ class ServiceFacadeImpl implements ServiceFacade { // Service LoadRunner Classes // ========================== - synchronized LoadRunner getLoadRunner() { - if (loadRunner == null) loadRunner = new LoadRunner(ecfi) - return loadRunner + LoadRunner getLoadRunner() { + if (loadRunner != null) return loadRunner + loadRunnerLock.lock() + try { + if (loadRunner == null) loadRunner = new LoadRunner(ecfi) + return loadRunner + } finally { + loadRunnerLock.unlock() + } } static class LoadRunnerServiceRunnable implements Runnable, Externalizable { 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..0fab09fb8 --- /dev/null +++ b/framework/src/main/java/org/moqui/service/ServiceCallScheduled.java @@ -0,0 +1,120 @@ +/* + * 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.TimeUnit; +import java.util.concurrent.Callable; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeoutException; + + +/** Example use: + * ec.service.schedule().name("...").parameters([...]).distribute(true).initialDelay(50).atFixedRate(100).call() + * ec.service.schedule().name("...").parameters([...]).distribute(true).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); + + /** If true the scheduled service call will be run distributed and may run on a different member of the cluster. Parameter + * entries MUST be java.io.Serializable (or java.io.Externalizable). + * + * If false it will be run local only (default). + * + * @return Reference to this for convenience. + */ + ServiceCallScheduled distribute(boolean dist); + + /** 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..3612452fe 100644 --- a/framework/src/main/java/org/moqui/service/ServiceFacade.java +++ b/framework/src/main/java/org/moqui/service/ServiceFacade.java @@ -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/template/XmlActions.groovy.ftl b/framework/template/XmlActions.groovy.ftl index 3d2a1555b..d56011015 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,41 @@ 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["@distribute"]?if_exists == "true">.distribute(true) + <#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/service-definition-3.xsd b/framework/xsd/service-definition-3.xsd index 44818fc94..7e0b86e8c 100644 --- a/framework/xsd/service-definition-3.xsd +++ b/framework/xsd/service-definition-3.xsd @@ -73,6 +73,7 @@ along with this software (see the LICENSE.md file). If not, see + diff --git a/framework/xsd/xml-actions-3.xsd b/framework/xsd/xml-actions-3.xsd index 01587c0f6..036b86c3c 100644 --- a/framework/xsd/xml-actions-3.xsd +++ b/framework/xsd/xml-actions-3.xsd @@ -185,6 +185,95 @@ 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. + + + + + + + + + + + + If false this will be run local only. By default runs on any server in a cluster listening for async or scheduled distributed services (if a distributed executor is configured). + + + 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. + + + From 1e34e5708a8dbdb23217666adbdd0ecdb19e44d4 Mon Sep 17 00:00:00 2001 From: Moqui Industrial <729502+moqui-industrial@users.noreply.github.com> Date: Thu, 14 May 2026 11:16:15 +0200 Subject: [PATCH 2/2] Add distributed-scheduled-executor-factory to service-facade conf Introduces CustomDistributedScheduledExecutor interface (ContextJavaUtil) and wires it through ServiceFacadeImpl so a Hazelcast (or other) implementation can replace the local ScheduledExecutorService for cluster-wide scheduled service execution. BREAKING: distributed-factory attribute renamed to distributed-executor-factory. Update any MoquiConf.xml that uses distributed-factory="..." accordingly. The distributed-scheduled-executor-factory attribute is new and optional; omitting it keeps the current local-only behaviour. --- .../src/main/resources/MoquiDefaultConf.xml | 72 ++++++++++++++++++- framework/xsd/moqui-conf-3.xsd | 21 +++++- 2 files changed, 90 insertions(+), 3 deletions(-) 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/xsd/moqui-conf-3.xsd b/framework/xsd/moqui-conf-3.xsd index d060a6176..c2116c6f7 100644 --- a/framework/xsd/moqui-conf-3.xsd +++ b/framework/xsd/moqui-conf-3.xsd @@ -77,6 +77,10 @@ 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).