Enhance service semaphore to support parameter-based locking

applications/datamodel/DATAMODEL_CHANGES.adoc

@@ -419,6 +419,7 @@ No changes
 |SalaryStep |lastModifiedByUserLogin |Added |No |NA
 |SalesOpportunity |nextStepDate |Added |No |NA
 |ServiceSemaphore |lockedByInstanceId |Added |No |NA
+|ServiceSemaphore |parameterValue |Added |Yes |NA
 |ShoppingListItem |modifiedPrice |Added |No |NA
 |SkillType |parentTypeId |Added |No |NA
 |SkillType |hasTable |Added |No |NA

framework/service/dtd/services.xsd

@@ -131,6 +131,7 @@ under the License.
         </xs:attribute>
         <xs:attribute name="semaphore-wait-seconds" type="xs:int" default="300"/>
         <xs:attribute name="semaphore-sleep" type="xs:int" default="500"/>
+        <xs:attribute name="semaphore-parameter-name" type="xs:string"/>
     </xs:attributeGroup>
     <xs:element name="notification">
         <xs:complexType>

framework/service/entitydef/entitymodel.xml

@@ -212,10 +212,12 @@ under the License.
     <entity entity-name="ServiceSemaphore" package-name="org.apache.ofbiz.service.semaphore" title="Semaphore Lock"
             sequence-bank-size="100">
         <field name="serviceName" type="name"></field>
+        <field name="parameterValue" type="name"/>
         <field name="lockedByInstanceId" type="id"></field>
         <field name="lockThread" type="name"></field>
         <field name="lockTime" type="date-time"></field>
         <prim-key field="serviceName"/>
+        <prim-key field="parameterValue"/>
     </entity>
 
 </entitymodel>

framework/service/src/main/java/org/apache/ofbiz/service/ModelService.java

@@ -185,6 +185,9 @@ public class ModelService extends AbstractMap<String, Object> implements Seriali
     /** Semaphore sleep time (in milliseconds) */
     private int semaphoreSleep;
 
+    /** Semaphore parameter name */
+    private String semaphoreParameterName;
+
     /** Require a new transaction for this service */
     private boolean hideResultInLog;
 
@@ -398,6 +401,14 @@ public void setSemaphoreSleep(int semaphoreSleep) {
         this.semaphoreSleep = semaphoreSleep;
     }
 
+    /**
+     * Sets semaphore parameter name.
+     * @param semaphoreParameterName the semaphore sleep
+     */
+    public void setSemaphoreParameterName(String semaphoreParameterName) {
+        this.semaphoreParameterName = semaphoreParameterName;
+    }
+
     /**
      * Sets hide result in log.
      * @param hideResultInLog the hide result in log
@@ -574,6 +585,14 @@ public int getSemaphoreSleep() {
         return semaphoreSleep;
     }
 
+    /**
+     * Gets semaphore parameter name.
+     * @return the semaphore parameter name
+     */
+    public String getSemaphoreParameterName() {
+        return semaphoreParameterName;
+    }
+
     /**
      * Gets impl services.
      * @return the impl services
@@ -791,6 +810,7 @@ public ModelService(ModelService model) {
         this.debug = model.debug;
         this.semaphoreWait = model.semaphoreWait;
         this.semaphoreSleep = model.semaphoreSleep;
+        this.semaphoreParameterName = model.semaphoreParameterName;
         this.contextInfo = model.contextInfo;
         this.definitionLocation = model.definitionLocation;
         this.description = model.description;

framework/service/src/main/java/org/apache/ofbiz/service/ModelServiceReader.java

@@ -214,6 +214,11 @@ private ModelService createModelService(Element serviceElement, String resourceL
         }
         service.setSemaphoreSleep(semaphoreSleep);
 
+        String semaphoreParamName = UtilXml.checkEmpty(serviceElement.getAttribute("semaphore-parameter-name"));
+        if (UtilValidate.isNotEmpty(semaphoreParamName)) {
+            service.setSemaphoreParameterName(semaphoreParamName);
+        }
+
         // set the max retry field
         String maxRetryStr = UtilXml.checkEmpty(serviceElement.getAttribute("max-retry"));
         int maxRetry = 0;

framework/service/src/main/java/org/apache/ofbiz/service/ServiceDispatcher.java

@@ -284,9 +284,16 @@ public Map<String, Object> runSync(String localName, ModelService modelService,
         boolean isError = false;
         boolean beganTrans = false;
         try {
+            String semaphoreParamValue = "_NA_";
+            String semaphoreParamName = modelService.getSemaphoreParameterName();
+            if (UtilValidate.isNotEmpty(semaphoreParamName)) {
+                Object paramObj = params.get(semaphoreParamName);
+                semaphoreParamValue = (paramObj != null) ? paramObj.toString() : "_NULL_";
+            }
+
             // check for semaphore and acquire a lock
             if ("wait".equals(modelService.getSemaphore()) || "fail".equals(modelService.getSemaphore())) {
-                lock = new ServiceSemaphore(delegator, modelService);
+                lock = new ServiceSemaphore(delegator, modelService, semaphoreParamValue);
                 lock.acquire();
             }
 

framework/service/src/main/java/org/apache/ofbiz/service/semaphore/ServiceSemaphore.java

@@ -50,17 +50,19 @@ public final class ServiceSemaphore {
     private Delegator delegator;
     private GenericValue lock;
     private ModelService model;
+    private String parameterValue;
 
     private int wait = 0;
     private int mode;
     private Timestamp lockTime = null;
 
-    public ServiceSemaphore(Delegator delegator, ModelService model) {
+    public ServiceSemaphore(Delegator delegator, ModelService model, String parameterValue) {
         this.delegator = delegator;
         this.mode = "wait".equals(model.getSemaphore()) ? SEMAPHORE_MODE_WAIT
                 : ("fail".equals(model.getSemaphore()) ? SEMAPHORE_MODE_FAIL : SEMAPHORE_MODE_NONE);
         this.model = model;
         this.lock = null;
+        this.parameterValue = parameterValue;
     }
 
     /**
@@ -147,8 +149,8 @@ private boolean checkLockNeedToWait() throws SemaphoreFailException {
 
         try {
             if (EntityQuery.use(delegator).from("ServiceSemaphore")
-                    .where("serviceName", model.getName()).queryCount() == 0) {
-                semaphore = delegator.makeValue("ServiceSemaphore", "serviceName", model.getName(),
+                    .where("serviceName", model.getName(), "parameterValue", parameterValue).queryCount() == 0) {
+                semaphore = delegator.makeValue("ServiceSemaphore", "serviceName", model.getName(), "parameterValue", parameterValue,
                         "lockedByInstanceId", JobManager.INSTANCE_ID, "lockThread", threadName, "lockTime", lockTime);
 
                 // use the special method below so we can reuse the unique tx functions
@@ -196,7 +198,7 @@ private synchronized boolean dbWrite(GenericValue value, boolean delete) {
                 } else {
                     // Last check before inserting data in this transaction to avoid error log
                     isError = EntityQuery.use(delegator).from("ServiceSemaphore")
-                            .where("serviceName", model.getName()).queryCount() != 0;
+                            .where("serviceName", model.getName(), "parameterValue", parameterValue).queryCount() != 0;
                     if (!isError) {
                         lock = value.create();
                     }