Author: jeff.yuchang Date: 2012-06-13 08:19:06 -0400 (Wed, 13 Jun 2012) New Revision: 1572 Modified: branches/ODE/RiftSaw-ODE-trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/engine/BpelRuntimeContextImpl.java branches/ODE/RiftSaw-ODE-trunk/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/ODE/RiftSaw-ODE-trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/engine/BpelRuntimeContextImpl.java =================================================================== --- branches/ODE/RiftSaw-ODE-trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/engine/BpelRuntimeContextImpl.java 2012-06-13 12:16:22 UTC (rev 1571) +++ branches/ODE/RiftSaw-ODE-trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/engine/BpelRuntimeContextImpl.java 2012-06-13 12:19:06 UTC (rev 1572) @@ -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/ODE/RiftSaw-ODE-trunk/dao-jpa/src/main/java/org/apache/ode/dao/jpa/bpel/CorrelatorDAOImpl.java =================================================================== --- branches/ODE/RiftSaw-ODE-trunk/dao-jpa/src/main/java/org/apache/ode/dao/jpa/bpel/CorrelatorDAOImpl.java 2012-06-13 12:16:22 UTC (rev 1571) +++ branches/ODE/RiftSaw-ODE-trunk/dao-jpa/src/main/java/org/apache/ode/dao/jpa/bpel/CorrelatorDAOImpl.java 2012-06-13 12:19:06 UTC (rev 1572) @@ -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; }