7:16 a.m.
Author: jeff.yuchang
Date: 2012-06-13 08:16:22 -0400 (Wed, 13 Jun 2012)
New Revision: 1571
Modified:
branches/RiftSaw-ODE-2.3.x/bpel-runtime/src/main/java/org/apache/ode/bpel/engine/BpelRuntimeContextImpl.java
branches/RiftSaw-ODE-2.3.x/dao-jpa/src/main/java/org/apache/ode/dao/jpa/bpel/CorrelatorDAOImpl.java
Log:
*RIFTSAW-495: now only dequeue the MEX that are not 'MATCHED', and updated the
status from 'QUEUED' to 'MATCHED' once its been matched.
Modified:
branches/RiftSaw-ODE-2.3.x/bpel-runtime/src/main/java/org/apache/ode/bpel/engine/BpelRuntimeContextImpl.java
===================================================================
---
branches/RiftSaw-ODE-2.3.x/bpel-runtime/src/main/java/org/apache/ode/bpel/engine/BpelRuntimeContextImpl.java 2012-06-13
10:51:38 UTC (rev 1570)
+++
branches/RiftSaw-ODE-2.3.x/bpel-runtime/src/main/java/org/apache/ode/bpel/engine/BpelRuntimeContextImpl.java 2012-06-13
12:16:22 UTC (rev 1571)
@@ -61,6 +61,7 @@
import org.apache.ode.bpel.iapi.MessageExchange.FailureType;
import org.apache.ode.bpel.iapi.MessageExchange.MessageExchangePattern;
import org.apache.ode.bpel.iapi.MessageExchange.Status;
+import org.apache.ode.bpel.iapi.MyRoleMessageExchange.CorrelationStatus;
import org.apache.ode.bpel.iapi.ProcessConf.CLEANUP_CATEGORY;
import org.apache.ode.bpel.iapi.ProcessConf.PartnerRoleConfig;
import org.apache.ode.bpel.iapi.Scheduler.JobDetails;
@@ -1490,6 +1491,11 @@
__log.warn("A message arrived before a receive is ready for a
request/response pattern. This may be processed to success. However, you should consider
revising your process since a TCP port and a container thread will be held for a longer
time and the process will not scale under heavy load.");
}
+ //marked the MEX as 'MATCHED' from 'QUEUED'
+ if (CorrelationStatus.QUEUED.toString().equals(mexdao.getCorrelationStatus()))
{
+ mexdao.setCorrelationStatus(CorrelationStatus.MATCHED.toString());
+ }
+
for (MessageRouteDAO mroute : mroutes) {
// We have a match, so we can get rid of the routing entries.
correlator.removeRoutes(mroute.getGroupId(), _dao);
@@ -1503,7 +1509,7 @@
BpelProcess.__log.debug("SELECT: " + mroute.getGroupId() +
": matched to MESSAGE " + mexdao
+ " on CKEYSET " + ckeySet);
}
-
+
MyRoleMessageExchangeImpl mex = new MyRoleMessageExchangeImpl(_bpelProcess,
_bpelProcess._engine, mexdao);
inputMsgMatch(mroute.getGroupId(), mroute.getIndex(), mex);
Modified:
branches/RiftSaw-ODE-2.3.x/dao-jpa/src/main/java/org/apache/ode/dao/jpa/bpel/CorrelatorDAOImpl.java
===================================================================
---
branches/RiftSaw-ODE-2.3.x/dao-jpa/src/main/java/org/apache/ode/dao/jpa/bpel/CorrelatorDAOImpl.java 2012-06-13
10:51:38 UTC (rev 1570)
+++
branches/RiftSaw-ODE-2.3.x/dao-jpa/src/main/java/org/apache/ode/dao/jpa/bpel/CorrelatorDAOImpl.java 2012-06-13
12:16:22 UTC (rev 1571)
@@ -22,6 +22,7 @@
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.ode.bpel.common.CorrelationKeySet;
+import org.apache.ode.bpel.iapi.MyRoleMessageExchange.CorrelationStatus;
import org.apache.ode.dao.bpel.*;
import javax.persistence.*;
@@ -75,7 +76,8 @@
// TODO: this thing does not seem to be scalable: loading up based on a
correlator???
for (Iterator<MessageExchangeDAOImpl> itr = _exchanges.iterator();
itr.hasNext();) {
MessageExchangeDAOImpl mex = itr.next();
- if (mex.getCorrelationKeySet().isRoutableTo(correlationKeySet, false)) {
+ if (!CorrelationStatus.MATCHED.toString().equals(mex.getCorrelationStatus())
+ && mex.getCorrelationKeySet().isRoutableTo(correlationKeySet,
false)) {
itr.remove();
return mex;
}