You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ode.apache.org by mr...@apache.org on 2007/02/24 20:34:49 UTC
svn commit: r511331 - in /incubator/ode/trunk:
axis2/src/main/java/org/apache/ode/axis2/
bpel-api/src/main/java/org/apache/ode/bpel/iapi/
bpel-dao/src/main/java/org/apache/ode/bpel/dao/
bpel-runtime/src/main/java/org/apache/ode/bpel/engine/ bpel-runtim...
Author: mriou
Date: Sat Feb 24 11:34:48 2007
New Revision: 511331
URL: http://svn.apache.org/viewvc?view=rev&rev=511331
Log:
Added a release() method on MessageExchange to allow the mex to clean the resources they're holding (mostly dao stuff). The in-mem DAOs are now being cleaned up properly once the instance has finished its execution and the mex have been released so that the GC can do its job. Otherwise all in-mem DAOs are just one big memory leak.
Modified:
incubator/ode/trunk/axis2/src/main/java/org/apache/ode/axis2/ODEService.java
incubator/ode/trunk/axis2/src/main/java/org/apache/ode/axis2/P2PMexContextImpl.java
incubator/ode/trunk/bpel-api/src/main/java/org/apache/ode/bpel/iapi/MessageExchange.java
incubator/ode/trunk/bpel-dao/src/main/java/org/apache/ode/bpel/dao/MessageExchangeDAO.java
incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/engine/BpelRuntimeContextImpl.java
incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/engine/MessageExchangeImpl.java
incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/memdao/BpelDAOConnectionImpl.java
incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/memdao/CorrelationSetDaoImpl.java
incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/memdao/CorrelatorDaoImpl.java
incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/memdao/DaoBaseImpl.java
incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/memdao/MessageDAOImpl.java
incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/memdao/MessageExchangeDAOImpl.java
incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/memdao/ProcessDaoImpl.java
incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/memdao/ProcessInstanceDaoImpl.java
incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/memdao/XmlDataDaoImpl.java
incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/runtime/BpelRuntimeContext.java
incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/runtime/INVOKE.java
incubator/ode/trunk/dao-hibernate/src/main/java/org/apache/ode/daohib/bpel/MessageExchangeDaoImpl.java
incubator/ode/trunk/dao-jpa/src/main/java/org/apache/ode/dao/jpa/MessageExchangeDAOImpl.java
incubator/ode/trunk/jbi/src/main/java/org/apache/ode/jbi/OdeService.java
Modified: incubator/ode/trunk/axis2/src/main/java/org/apache/ode/axis2/ODEService.java
URL: http://svn.apache.org/viewvc/incubator/ode/trunk/axis2/src/main/java/org/apache/ode/axis2/ODEService.java?view=diff&rev=511331&r1=511330&r2=511331
==============================================================================
--- incubator/ode/trunk/axis2/src/main/java/org/apache/ode/axis2/ODEService.java (original)
+++ incubator/ode/trunk/axis2/src/main/java/org/apache/ode/axis2/ODEService.java Sat Feb 24 11:34:48 2007
@@ -192,6 +192,8 @@
__log.error("Error processing response for MEX " + odeMex, e);
throw new OdeFault("An exception occured when invoking ODE.", e);
} finally {
+ if (odeMex!= null) odeMex.release();
+ else __log.warn("Couldn't release a message exchange, it's null.");
if (commit)
try {
if (__log.isDebugEnabled()) __log.debug("Comitting transaction.");
@@ -205,7 +207,6 @@
} catch (Exception ex) {
throw new OdeFault("Rollback failed!", ex);
}
-
}
}
}
Modified: incubator/ode/trunk/axis2/src/main/java/org/apache/ode/axis2/P2PMexContextImpl.java
URL: http://svn.apache.org/viewvc/incubator/ode/trunk/axis2/src/main/java/org/apache/ode/axis2/P2PMexContextImpl.java?view=diff&rev=511331&r1=511330&r2=511331
==============================================================================
--- incubator/ode/trunk/axis2/src/main/java/org/apache/ode/axis2/P2PMexContextImpl.java (original)
+++ incubator/ode/trunk/axis2/src/main/java/org/apache/ode/axis2/P2PMexContextImpl.java Sat Feb 24 11:34:48 2007
@@ -47,25 +47,20 @@
if (myService != null) {
// Defer invoke until tx is comitted.
_scheduler.registerSynchronizer(new Scheduler.Synchronizer() {
-
public void afterCompletion(boolean success) {
- if (!success)
- return;
-
+ if (!success) return;
try {
_scheduler.execIsolatedTransaction(new Callable<Void>() {
-
public Void call() throws Exception {
- buildAndInvokeMyRoleMex(pmex);
+ MyRoleMessageExchange mymex = buildAndInvokeMyRoleMex(pmex);
+ mymex.release();
return null;
}
});
-
} catch (Exception ex) {
__log.error("Unexpected error", ex);
throw new RuntimeException(ex);
}
-
}
public void beforeCompletion() {
@@ -118,7 +113,6 @@
odeMex.getOperationName());
copyHeader(pmex, odeMex);
-
odeMex.invoke(pmex.getRequest());
Modified: incubator/ode/trunk/bpel-api/src/main/java/org/apache/ode/bpel/iapi/MessageExchange.java
URL: http://svn.apache.org/viewvc/incubator/ode/trunk/bpel-api/src/main/java/org/apache/ode/bpel/iapi/MessageExchange.java?view=diff&rev=511331&r1=511330&r2=511331
==============================================================================
--- incubator/ode/trunk/bpel-api/src/main/java/org/apache/ode/bpel/iapi/MessageExchange.java (original)
+++ incubator/ode/trunk/bpel-api/src/main/java/org/apache/ode/bpel/iapi/MessageExchange.java Sat Feb 24 11:34:48 2007
@@ -223,6 +223,12 @@
*/
public Set<String> getPropertyNames();
+ /**
+ * Should be called by the external partner when it's done with the
+ * message exchange. Ncessary for a better resource management and
+ * proper mex cleanup.
+ */
+ public void release();
public static final String PROPERTY_SEP_MYROLE_SESSIONID = "org.apache.ode.bpel.myRoleSessionId";
public static final String PROPERTY_SEP_PARTNERROLE_SESSIONID = "org.apache.ode.bpel.partnerRoleSessionId";
Modified: incubator/ode/trunk/bpel-dao/src/main/java/org/apache/ode/bpel/dao/MessageExchangeDAO.java
URL: http://svn.apache.org/viewvc/incubator/ode/trunk/bpel-dao/src/main/java/org/apache/ode/bpel/dao/MessageExchangeDAO.java?view=diff&rev=511331&r1=511330&r2=511331
==============================================================================
--- incubator/ode/trunk/bpel-dao/src/main/java/org/apache/ode/bpel/dao/MessageExchangeDAO.java (original)
+++ incubator/ode/trunk/bpel-dao/src/main/java/org/apache/ode/bpel/dao/MessageExchangeDAO.java Sat Feb 24 11:34:48 2007
@@ -85,9 +85,7 @@
/**
* Set state of last message sent/received.
- *
- * @param string
- * state to be set
+ * @param string state to be set
*/
void setStatus(String status);
@@ -233,5 +231,7 @@
void setPartnerLink(PartnerLinkDAO plinkDAO);
PartnerLinkDAO getPartnerLink();
+
+ void release();
}
Modified: incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/engine/BpelRuntimeContextImpl.java
URL: http://svn.apache.org/viewvc/incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/engine/BpelRuntimeContextImpl.java?view=diff&rev=511331&r1=511330&r2=511331
==============================================================================
--- incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/engine/BpelRuntimeContextImpl.java (original)
+++ incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/engine/BpelRuntimeContextImpl.java Sat Feb 24 11:34:48 2007
@@ -1058,28 +1058,32 @@
throw new BpelEngineException(msg);
}
+ MessageDAO response;
MessageExchange.Status status = MessageExchange.Status.valueOf(dao.getStatus());
switch (status) {
- case FAULT:
- case RESPONSE:
- MessageDAO response = dao.getResponse();
- if (response == null) {
- // this also should not happen
- String msg = "Engine requested response for message exchange that did not have one: " + mexId;
+ case FAULT:
+ case RESPONSE:
+ response = dao.getResponse();
+ if (response == null) {
+ // this also should not happen
+ String msg = "Engine requested response for message exchange that did not have one: " + mexId;
+ __log.fatal(msg);
+ throw new BpelEngineException(msg);
+ }
+ break;
+ default:
+ // We should not be in any other state when requesting this.
+ String msg = "Engine requested response while the message exchange " + mexId + " was in the state "
+ + status;
__log.fatal(msg);
throw new BpelEngineException(msg);
- }
-
- return response;
-
- default:
- // We should not be in any other state when requesting this.
- String msg = "Engine requested response while the message exchange " + mexId + " was in the state "
- + status;
- __log.fatal(msg);
- throw new BpelEngineException(msg);
}
+ return response;
+ }
+ public void releasePartnerMex(String mexId) {
+ MessageExchangeDAO dao = _dao.getConnection().getMessageExchange(mexId);
+ dao.release();
}
public Node getPartData(Element message, Part part) {
Modified: incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/engine/MessageExchangeImpl.java
URL: http://svn.apache.org/viewvc/incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/engine/MessageExchangeImpl.java?view=diff&rev=511331&r1=511330&r2=511331
==============================================================================
--- incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/engine/MessageExchangeImpl.java (original)
+++ incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/engine/MessageExchangeImpl.java Sat Feb 24 11:34:48 2007
@@ -50,13 +50,10 @@
protected EndpointReference _epr;
- protected final MessageExchangeDAO _dao;
+ protected MessageExchangeDAO _dao;
/**
* Constructor: requires the minimal information for a message exchange.
- * @param pattern
- * @param opname
- * @param epr
*/
MessageExchangeImpl(BpelEngineImpl engine,
MessageExchangeDAO dao,
@@ -232,6 +229,13 @@
public Set<String> getPropertyNames() {
return getDAO().getPropertyNames();
}
+
+ public void release() {
+ __log.debug("Releasing mex " + getMessageExchangeId());
+ _dao.release();
+ _dao = null;
+ }
+
public String toString() {
return "MEX["+getDAO().getMessageExchangeId() +"]";
}
Modified: incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/memdao/BpelDAOConnectionImpl.java
URL: http://svn.apache.org/viewvc/incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/memdao/BpelDAOConnectionImpl.java?view=diff&rev=511331&r1=511330&r2=511331
==============================================================================
--- incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/memdao/BpelDAOConnectionImpl.java (original)
+++ incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/memdao/BpelDAOConnectionImpl.java Sat Feb 24 11:34:48 2007
@@ -14,6 +14,8 @@
import org.apache.ode.utils.ISO8601DateParser;
import org.apache.ode.utils.stl.CollectionsX;
import org.apache.ode.utils.stl.UnaryFunction;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
import javax.xml.namespace.QName;
import java.util.*;
@@ -24,6 +26,8 @@
* A very simple, in-memory implementation of the {@link BpelDAOConnection} interface.
*/
class BpelDAOConnectionImpl implements BpelDAOConnection {
+ private static final Log __log = LogFactory.getLog(BpelDAOConnectionImpl.class);
+
private Map<QName, ProcessDaoImpl> _store;
private List<BpelEvent> _events = new LinkedList<BpelEvent>();
private static Map<String,MessageExchangeDAO> _mexStore = Collections.synchronizedMap(new HashMap<String,MessageExchangeDAO>());
@@ -263,5 +267,13 @@
public Collection<ProcessInstanceDAO> instanceQuery(String expression) {
//TODO
throw new UnsupportedOperationException();
+ }
+
+ static void removeMessageExchange(String mexId) {
+ // Cleaning up mex
+ __log.debug("Removing mex " + mexId + " from memory store.");
+ MessageExchangeDAO mex = _mexStore.remove(mexId);
+ if (mex == null)
+ __log.warn("Couldn't find mex " + mexId + " for cleanup.");
}
}
Modified: incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/memdao/CorrelationSetDaoImpl.java
URL: http://svn.apache.org/viewvc/incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/memdao/CorrelationSetDaoImpl.java?view=diff&rev=511331&r1=511330&r2=511331
==============================================================================
--- incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/memdao/CorrelationSetDaoImpl.java (original)
+++ incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/memdao/CorrelationSetDaoImpl.java Sat Feb 24 11:34:48 2007
@@ -17,8 +17,7 @@
/**
* A very simple, in-memory implementation of the {@link CorrelationSetDAO} interface.
*/
-class CorrelationSetDaoImpl
- implements CorrelationSetDAO {
+class CorrelationSetDaoImpl extends DaoBaseImpl implements CorrelationSetDAO {
private Long _csetId;
private ScopeDAO _scope;
private String _name;
Modified: incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/memdao/CorrelatorDaoImpl.java
URL: http://svn.apache.org/viewvc/incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/memdao/CorrelatorDaoImpl.java?view=diff&rev=511331&r1=511330&r2=511331
==============================================================================
--- incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/memdao/CorrelatorDaoImpl.java (original)
+++ incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/memdao/CorrelatorDaoImpl.java Sat Feb 24 11:34:48 2007
@@ -25,112 +25,112 @@
* A very simple, in-memory implementation of the {@link CorrelatorDAO} interface.
*/
class CorrelatorDaoImpl extends DaoBaseImpl implements CorrelatorDAO {
- private static final Log __log = LogFactory.getLog(CorrelatorDaoImpl.class);
+ private static final Log __log = LogFactory.getLog(CorrelatorDaoImpl.class);
- private String _correlatorId;
- private List<MsgQueueEntry> _messages;
- private List<MessageRouteDaoImpl> _routes;
-
- CorrelatorDaoImpl(String correlatorId) {
- _messages = new ArrayList<MsgQueueEntry>();
- _routes = new ArrayList<MessageRouteDaoImpl>();
- _correlatorId = correlatorId;
- }
-
- public MessageExchangeDAO dequeueMessage(CorrelationKey key) {
- if (__log.isDebugEnabled()) {
- __log.debug("dequeueEarliest: MATCHING correlationKey=" + key);
- }
- for (Iterator i = _messages.iterator(); i.hasNext();) {
- MsgQueueEntry mqe = (MsgQueueEntry)i.next();
- Set<CorrelationKey> keyset = (Set<CorrelationKey>)ArrayUtils.makeCollection(HashSet.class, mqe.keys);
- if ((key == null) || keyset.contains(key)) {
- i.remove();
- return mqe.message;
- }
- }
- if (__log.isDebugEnabled()) {
- __log.debug("dequeueEarliest: MATCH NOT FOUND!");
- }
- return null;
- }
-
- public MessageRouteDAO findRoute(CorrelationKey key) {
- if (__log.isDebugEnabled()) {
- __log.debug("findRoute: key=" + key);
- }
- for (MessageRouteDaoImpl we : _routes) {
- if ((we._ckey == null && key == null) || (we._ckey != null && key != null && we._ckey.equals(key))) {
- return we;
- }
- }
- return null;
- }
-
- public String getCorrelatorId() {
- return _correlatorId;
- }
-
- public void removeRoutes(String routeGroupId, ProcessInstanceDAO target) {
- ((ProcessInstanceDaoImpl)target).removeRoutes(routeGroupId);
- }
-
- public void enqueueMessage(MessageExchangeDAO mex, CorrelationKey[] keys) {
- if (__log.isDebugEnabled()) {
- __log.debug("enqueueProcessInvocation: data=" + mex + " keys="
- + ArrayUtils.makeCollection(ArrayList.class, keys));
- }
-
- MsgQueueEntry mqe = new MsgQueueEntry(mex, keys);
- _messages.add(mqe);
- }
-
- public void addRoute(String routeId,ProcessInstanceDAO target, int idx, CorrelationKey key) {
- if (__log.isDebugEnabled()) {
- __log.debug("addRoute: target=" + target + " correlationKey=" + key);
- }
-
- MessageRouteDaoImpl mr = new MessageRouteDaoImpl((ProcessInstanceDaoImpl)target, routeId, key, idx);
- _routes.add(mr);
- }
-
-
- public boolean checkRoute(CorrelationKey ckey) {
- return true;
- }
-
- void _removeRoutes(String routeGroupId, ProcessInstanceDaoImpl target) {
- for (Iterator<MessageRouteDaoImpl> i = _routes.iterator(); i.hasNext();) {
- MessageRouteDaoImpl we = i.next();
- if ((we._groupId.equals(routeGroupId) || routeGroupId == null) && we._instance == target) {
- i.remove();
- }
- }
-
- }
- /**
- * @see java.lang.Object#toString()
- */
- public String toString() {
- StringBuffer buf = new StringBuffer("{CorrelatorDaoImpl corrId=");
- buf.append(_correlatorId);
- buf.append(" waiters=");
- buf.append(_routes);
- buf.append(" messages=");
- buf.append(_messages);
- buf.append('}');
-
- return buf.toString();
- }
-
- private class MsgQueueEntry {
- public final MessageExchangeDAO message;
- public final CorrelationKey[] keys;
-
- private MsgQueueEntry(MessageExchangeDAO mex,
- CorrelationKey[] keys) {
- this.message = mex;
- this.keys = keys;
+ private String _correlatorId;
+ private List<MsgQueueEntry> _messages;
+ private List<MessageRouteDaoImpl> _routes;
+
+ CorrelatorDaoImpl(String correlatorId) {
+ _messages = new ArrayList<MsgQueueEntry>();
+ _routes = new ArrayList<MessageRouteDaoImpl>();
+ _correlatorId = correlatorId;
+ }
+
+ public MessageExchangeDAO dequeueMessage(CorrelationKey key) {
+ if (__log.isDebugEnabled()) {
+ __log.debug("dequeueEarliest: MATCHING correlationKey=" + key);
+ }
+ for (Iterator i = _messages.iterator(); i.hasNext();) {
+ MsgQueueEntry mqe = (MsgQueueEntry)i.next();
+ Set<CorrelationKey> keyset = (Set<CorrelationKey>)ArrayUtils.makeCollection(HashSet.class, mqe.keys);
+ if ((key == null) || keyset.contains(key)) {
+ i.remove();
+ return mqe.message;
+ }
+ }
+ if (__log.isDebugEnabled()) {
+ __log.debug("dequeueEarliest: MATCH NOT FOUND!");
+ }
+ return null;
+ }
+
+ public MessageRouteDAO findRoute(CorrelationKey key) {
+ if (__log.isDebugEnabled()) {
+ __log.debug("findRoute: key=" + key);
+ }
+ for (MessageRouteDaoImpl we : _routes) {
+ if ((we._ckey == null && key == null) || (we._ckey != null && key != null && we._ckey.equals(key))) {
+ return we;
+ }
+ }
+ return null;
+ }
+
+ public String getCorrelatorId() {
+ return _correlatorId;
+ }
+
+ public void removeRoutes(String routeGroupId, ProcessInstanceDAO target) {
+ ((ProcessInstanceDaoImpl)target).removeRoutes(routeGroupId);
+ }
+
+ public void enqueueMessage(MessageExchangeDAO mex, CorrelationKey[] keys) {
+ if (__log.isDebugEnabled()) {
+ __log.debug("enqueueProcessInvocation: data=" + mex + " keys="
+ + ArrayUtils.makeCollection(ArrayList.class, keys));
+ }
+
+ MsgQueueEntry mqe = new MsgQueueEntry(mex, keys);
+ _messages.add(mqe);
+ }
+
+ public void addRoute(String routeId,ProcessInstanceDAO target, int idx, CorrelationKey key) {
+ if (__log.isDebugEnabled()) {
+ __log.debug("addRoute: target=" + target + " correlationKey=" + key);
+ }
+
+ MessageRouteDaoImpl mr = new MessageRouteDaoImpl((ProcessInstanceDaoImpl)target, routeId, key, idx);
+ _routes.add(mr);
+ }
+
+
+ public boolean checkRoute(CorrelationKey ckey) {
+ return true;
+ }
+
+ void _removeRoutes(String routeGroupId, ProcessInstanceDaoImpl target) {
+ for (Iterator<MessageRouteDaoImpl> i = _routes.iterator(); i.hasNext();) {
+ MessageRouteDaoImpl we = i.next();
+ if ((we._groupId.equals(routeGroupId) || routeGroupId == null) && we._instance == target) {
+ i.remove();
+ }
+ }
+ }
+
+ /**
+ * @see java.lang.Object#toString()
+ */
+ public String toString() {
+ StringBuffer buf = new StringBuffer("{CorrelatorDaoImpl corrId=");
+ buf.append(_correlatorId);
+ buf.append(" waiters=");
+ buf.append(_routes);
+ buf.append(" messages=");
+ buf.append(_messages);
+ buf.append('}');
+
+ return buf.toString();
+ }
+
+ private class MsgQueueEntry {
+ public final MessageExchangeDAO message;
+ public final CorrelationKey[] keys;
+
+ private MsgQueueEntry(MessageExchangeDAO mex,
+ CorrelationKey[] keys) {
+ this.message = mex;
+ this.keys = keys;
+ }
}
- }
}
Modified: incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/memdao/DaoBaseImpl.java
URL: http://svn.apache.org/viewvc/incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/memdao/DaoBaseImpl.java?view=diff&rev=511331&r1=511330&r2=511331
==============================================================================
--- incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/memdao/DaoBaseImpl.java (original)
+++ incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/memdao/DaoBaseImpl.java Sat Feb 24 11:34:48 2007
@@ -6,6 +6,8 @@
package org.apache.ode.bpel.memdao;
import org.apache.ode.utils.uuid.UUIDGen;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
import java.util.Date;
@@ -14,13 +16,18 @@
* Base-class for in-memory data-access objects.
*/
class DaoBaseImpl {
- private static final UUIDGen __uuidGen = new UUIDGen();
+ private static final Log __logger = LogFactory.getLog(DaoBaseImpl.class);
+ private static final UUIDGen __uuidGen = new UUIDGen();
- Date _createTime = new Date();
- String _uuid = __uuidGen.nextUUID();
-
- public Date getCreateTime() {
- return _createTime;
- }
+ Date _createTime = new Date();
+ String _uuid = __uuidGen.nextUUID();
+ public Date getCreateTime() {
+ return _createTime;
+ }
+
+ protected void finalize() throws Throwable {
+ super.finalize();
+ __logger.debug("Finalizing " + this);
+ }
}
Modified: incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/memdao/MessageDAOImpl.java
URL: http://svn.apache.org/viewvc/incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/memdao/MessageDAOImpl.java?view=diff&rev=511331&r1=511330&r2=511331
==============================================================================
--- incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/memdao/MessageDAOImpl.java (original)
+++ incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/memdao/MessageDAOImpl.java Sat Feb 24 11:34:48 2007
@@ -7,7 +7,7 @@
import org.apache.ode.utils.DOMUtils;
import org.w3c.dom.Element;
-public class MessageDAOImpl implements MessageDAO {
+public class MessageDAOImpl extends DaoBaseImpl implements MessageDAO {
private QName type;
private Element data;
Modified: incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/memdao/MessageExchangeDAOImpl.java
URL: http://svn.apache.org/viewvc/incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/memdao/MessageExchangeDAOImpl.java?view=diff&rev=511331&r1=511330&r2=511331
==============================================================================
--- incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/memdao/MessageExchangeDAOImpl.java (original)
+++ incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/memdao/MessageExchangeDAOImpl.java Sat Feb 24 11:34:48 2007
@@ -14,7 +14,7 @@
import java.util.Properties;
import java.util.Set;
-public class MessageExchangeDAOImpl implements MessageExchangeDAO {
+public class MessageExchangeDAOImpl extends DaoBaseImpl implements MessageExchangeDAO {
private String messageExchangeId;
private MessageDAO response;
@@ -40,9 +40,9 @@
private Properties properties = new Properties();
private PartnerLinkDAOImpl _plink;
- public MessageExchangeDAOImpl(char direction, String mesageEchangeId){
+ public MessageExchangeDAOImpl(char direction, String messageEchangeId){
this.direction = direction;
- this.messageExchangeId = mesageEchangeId;
+ this.messageExchangeId = messageEchangeId;
}
public String getMessageExchangeId() {
@@ -243,4 +243,17 @@
return retVal;
}
+ public void release() {
+ instance = null;
+ process = null;
+ _plink = null;
+ request = null;
+ response = null;
+ BpelDAOConnectionImpl.removeMessageExchange(getMessageExchangeId());
+ }
+
+
+ public String toString() {
+ return "mem.mex(direction=" + direction + " id=" + messageExchangeId + ")";
+ }
}
Modified: incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/memdao/ProcessDaoImpl.java
URL: http://svn.apache.org/viewvc/incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/memdao/ProcessDaoImpl.java?view=diff&rev=511331&r1=511330&r2=511331
==============================================================================
--- incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/memdao/ProcessDaoImpl.java (original)
+++ incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/memdao/ProcessDaoImpl.java Sat Feb 24 11:34:48 2007
@@ -98,6 +98,11 @@
}
public void instanceCompleted(ProcessInstanceDAO instance) {
+ // Cleaning up
+ __log.debug("Removing completed process instance " + instance.getInstanceId() + " from in-memory store.");
+ ProcessInstanceDAO removed = _instances.remove(instance.getInstanceId());
+ if (removed == null)
+ __log.warn("Couldn't find process instance " + instance.getInstanceId() + " for cleanup.");
}
public void delete() {
Modified: incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/memdao/ProcessInstanceDaoImpl.java
URL: http://svn.apache.org/viewvc/incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/memdao/ProcessInstanceDaoImpl.java?view=diff&rev=511331&r1=511330&r2=511331
==============================================================================
--- incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/memdao/ProcessInstanceDaoImpl.java (original)
+++ incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/memdao/ProcessInstanceDaoImpl.java Sat Feb 24 11:34:48 2007
@@ -380,4 +380,7 @@
return _conn;
}
+ public String toString() {
+ return "mem.instance(type=" + _processDao.getType() + " iid=" + _instanceId + ")";
+ }
}
Modified: incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/memdao/XmlDataDaoImpl.java
URL: http://svn.apache.org/viewvc/incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/memdao/XmlDataDaoImpl.java?view=diff&rev=511331&r1=511330&r2=511331
==============================================================================
--- incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/memdao/XmlDataDaoImpl.java (original)
+++ incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/memdao/XmlDataDaoImpl.java Sat Feb 24 11:34:48 2007
@@ -19,7 +19,7 @@
/**
* A very simple, in-memory implementation of the {@link XmlDataDAO} interface.
*/
-class XmlDataDaoImpl implements XmlDataDAO {
+class XmlDataDaoImpl extends DaoBaseImpl implements XmlDataDAO {
private Node _data;
private Properties _properties = new Properties();
@@ -56,13 +56,16 @@
* @see XmlDataDAO#set(org.w3c.dom.Node)
*/
public void set(Node val) {
- if (!(val instanceof Element)) _data = val;
+ if (val == null || !(val instanceof Element)) {
+ _data = val;
+ return;
+ }
// For some reason we're getting some weird DOM trees from ServiceMix. Until we
// spot where it exactly comes from, this fixes it.
// The weirdness lies in elements being in xmlns="" when printed with a DOMWriter
// but having a ns when the elements are queried directly.
try {
- _data = DOMUtils.stringToDOM(DOMUtils.domToString(_data));
+ _data = DOMUtils.stringToDOM(DOMUtils.domToString(val));
} catch (SAXException e) {
// Should never happen but life is full of surprises
throw new RuntimeException("Couldn't reread!", e);
Modified: incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/runtime/BpelRuntimeContext.java
URL: http://svn.apache.org/viewvc/incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/runtime/BpelRuntimeContext.java?view=diff&rev=511331&r1=511330&r2=511331
==============================================================================
--- incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/runtime/BpelRuntimeContext.java (original)
+++ incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/runtime/BpelRuntimeContext.java Sat Feb 24 11:34:48 2007
@@ -52,205 +52,207 @@
*/
public interface BpelRuntimeContext {
- Long getPid();
-
- /**
- * Checks for variable initialization, i.e. has had a 'write'
- *
- * @param variable variable
- *
- * @return <code>true</code> if initialized
- */
- boolean isVariableInitialized(VariableInstance variable);
-
- /**
- * Create a scope instance object.
- * @param parentScopeId _id of parent scope (null if root scope)
- * @param scopeType the type of scope, i.e. the name of the scope
- *
- * @return scope instance identifier
- */
- Long createScopeInstance(Long parentScopeId, OScope scopeType);
-
- /**
- * Initializes endpoint references for partner links inside a scope.
- * @param parentScopeId
- * @param partnerLinks
- */
- void initializePartnerLinks(Long parentScopeId, Collection<OPartnerLink> partnerLinks);
-
- /**
- *
- * @param var variable to read
- * @return
- */
- Node fetchVariableData(VariableInstance var, boolean forWriting)
- throws FaultException;
-
- Node fetchVariableData(VariableInstance var, OMessageVarType.Part partname, boolean forWriting)
- throws FaultException;
-
- /**
- * Fetches the my-role endpoint reference data.
- * @param pLink
- * @param isMyEPR
- * @return
- * @throws FaultException
- */
- Element fetchMyRoleEndpointReferenceData(PartnerLinkInstance pLink);
-
- Element fetchPartnerRoleEndpointReferenceData(PartnerLinkInstance pLink) throws FaultException;
-
- /**
- * Determine if the partner role of an endpoint has been initialized (either explicitly throug assginment or via the
- * deployment descriptor)
- * @param pLink partner link
- * @return
- */
- boolean isPartnerRoleEndpointInitialized(PartnerLinkInstance pLink);
-
- /**
- * Fetches our session id associated with the partner link instance. This will always return a
- * non-null value.
- * @param pLink partner link
- */
- String fetchMySessionId(PartnerLinkInstance pLink);
-
- /**
- * Fetches the partner's session id associated with the partner link instance.
- * @param pLink partner link
- */
- String fetchPartnersSessionId(PartnerLinkInstance pLink);
-
- /**
- * Initialize the partner's session id for this partner link instance.
- * @param pLink partner link
- * @param session session identifier
- */
- void initializePartnersSessionId(PartnerLinkInstance pLink, String session);
-
- /**
- * Evaluate a property alias query expression against a variable, returning the normalized
- * {@link String} representation of the property value.
- * @param var variable to read
- * @param property property to read
- * @return value of property for variable, in String form
- * @throws FaultException in case of selection or other fault
- */
- String readProperty(VariableInstance var, OProcess.OProperty property)
- throws FaultException;
-
- Node initializeVariable(VariableInstance var, Node initData);
-
- /**
- * Writes a partner EPR.
- *
- * @param variable
- * @param data
- * @throws FaultException
- */
- void writeEndpointReference(PartnerLinkInstance variable, Element data) throws FaultException;
-
- Node convertEndpointReference(Element epr, Node targetNode);
-
- void commitChanges(VariableInstance var, Node changes);
-
- boolean isCorrelationInitialized(CorrelationSetInstance cset);
-
- CorrelationKey readCorrelation(CorrelationSetInstance cset);
-
- void writeCorrelation(CorrelationSetInstance cset, CorrelationKey correlation);
-
- /**
- * Should be invoked by process template, signalling process completion
- * with no faults.
- *
- */
- void completedOk();
-
- /**
- * Should be invoked by process template, signalling process completion
- * with fault.
- */
- void completedFault(FaultData faultData);
-
-
- /**
- * Non-deterministic selection on incoming message-exchanges.
- */
- void select(PickResponseChannel response, Date timeout, boolean createInstnace,
- Selector[] selectors) throws FaultException;
-
- /**
- * Cancel a timer, or pick.
- * @param timerResponseChannel
- */
- void cancel(TimerResponseChannel timerResponseChannel);
-
- /**
- * Send a reply to an open message-exchange.
- * @param msg response message
- * @param fault fault name, if this is a fault reply, otherwise <code>null</code>
- */
- void reply(PartnerLinkInstance plink, String opName, String mexId, Element msg,
- QName fault)
- throws FaultException;
-
-
- String invoke(PartnerLinkInstance partnerLinkInstance,
- Operation operation,
- Element outboundMsg,
- InvokeResponseChannel invokeResponseChannel) throws FaultException;
-
-
- /**
- * Registers a timer for future notification.
- * @param timerChannel channel for timer notification
- * @param timeToFire future time to fire timer notification
- */
- void registerTimer(TimerResponseChannel timerChannel, Date timeToFire);
-
- /**
- * Terminates the process / sets state flag to terminate
- * and ceases all processing on the VPU.
- */
- void terminate();
-
- /**
- * Sends the bpel event.
- * @param event
- */
- void sendEvent(ProcessInstanceEvent event);
-
- ExpressionLanguageRuntimeRegistry getExpLangRuntime();
-
-
- /**
- * Generate a unique (and monotonic) ID in the context of this instance.
- * @return
- */
- long genId();
-
- Element getPartnerResponse(String mexId);
-
- Element getMyRequest(String mexId);
-
- QName getPartnerFault(String mexId);
-
- String getPartnerFaultExplanation(String mexId);
-
- QName getPartnerResponseType(String mexId);
-
- Node getPartData(Element message, Part part);
-
- Element getSourceEPR(String mexId);
+ Long getPid();
- void registerActivityForRecovery(ActivityRecoveryChannel channel, long activityId, String reason,
- Date dateTime, Element details, String[] actions, int retries);
+ /**
+ * Checks for variable initialization, i.e. has had a 'write'
+ *
+ * @param variable variable
+ *
+ * @return <code>true</code> if initialized
+ */
+ boolean isVariableInitialized(VariableInstance variable);
+
+ /**
+ * Create a scope instance object.
+ * @param parentScopeId _id of parent scope (null if root scope)
+ * @param scopeType the type of scope, i.e. the name of the scope
+ *
+ * @return scope instance identifier
+ */
+ Long createScopeInstance(Long parentScopeId, OScope scopeType);
+
+ /**
+ * Initializes endpoint references for partner links inside a scope.
+ * @param parentScopeId
+ * @param partnerLinks
+ */
+ void initializePartnerLinks(Long parentScopeId, Collection<OPartnerLink> partnerLinks);
+
+ /**
+ *
+ * @param var variable to read
+ * @return
+ */
+ Node fetchVariableData(VariableInstance var, boolean forWriting)
+ throws FaultException;
+
+ Node fetchVariableData(VariableInstance var, OMessageVarType.Part partname, boolean forWriting)
+ throws FaultException;
+
+ /**
+ * Fetches the my-role endpoint reference data.
+ * @param pLink
+ * @param isMyEPR
+ * @return
+ * @throws FaultException
+ */
+ Element fetchMyRoleEndpointReferenceData(PartnerLinkInstance pLink);
+
+ Element fetchPartnerRoleEndpointReferenceData(PartnerLinkInstance pLink) throws FaultException;
+
+ /**
+ * Determine if the partner role of an endpoint has been initialized (either explicitly throug assginment or via the
+ * deployment descriptor)
+ * @param pLink partner link
+ * @return
+ */
+ boolean isPartnerRoleEndpointInitialized(PartnerLinkInstance pLink);
+
+ /**
+ * Fetches our session id associated with the partner link instance. This will always return a
+ * non-null value.
+ * @param pLink partner link
+ */
+ String fetchMySessionId(PartnerLinkInstance pLink);
+
+ /**
+ * Fetches the partner's session id associated with the partner link instance.
+ * @param pLink partner link
+ */
+ String fetchPartnersSessionId(PartnerLinkInstance pLink);
+
+ /**
+ * Initialize the partner's session id for this partner link instance.
+ * @param pLink partner link
+ * @param session session identifier
+ */
+ void initializePartnersSessionId(PartnerLinkInstance pLink, String session);
+
+ /**
+ * Evaluate a property alias query expression against a variable, returning the normalized
+ * {@link String} representation of the property value.
+ * @param var variable to read
+ * @param property property to read
+ * @return value of property for variable, in String form
+ * @throws FaultException in case of selection or other fault
+ */
+ String readProperty(VariableInstance var, OProcess.OProperty property)
+ throws FaultException;
+
+ Node initializeVariable(VariableInstance var, Node initData);
+
+ /**
+ * Writes a partner EPR.
+ *
+ * @param variable
+ * @param data
+ * @throws FaultException
+ */
+ void writeEndpointReference(PartnerLinkInstance variable, Element data) throws FaultException;
+
+ Node convertEndpointReference(Element epr, Node targetNode);
+
+ void commitChanges(VariableInstance var, Node changes);
+
+ boolean isCorrelationInitialized(CorrelationSetInstance cset);
+
+ CorrelationKey readCorrelation(CorrelationSetInstance cset);
+
+ void writeCorrelation(CorrelationSetInstance cset, CorrelationKey correlation);
+
+ /**
+ * Should be invoked by process template, signalling process completion
+ * with no faults.
+ *
+ */
+ void completedOk();
+
+ /**
+ * Should be invoked by process template, signalling process completion
+ * with fault.
+ */
+ void completedFault(FaultData faultData);
+
+
+ /**
+ * Non-deterministic selection on incoming message-exchanges.
+ */
+ void select(PickResponseChannel response, Date timeout, boolean createInstnace,
+ Selector[] selectors) throws FaultException;
+
+ /**
+ * Cancel a timer, or pick.
+ * @param timerResponseChannel
+ */
+ void cancel(TimerResponseChannel timerResponseChannel);
+
+ /**
+ * Send a reply to an open message-exchange.
+ * @param msg response message
+ * @param fault fault name, if this is a fault reply, otherwise <code>null</code>
+ */
+ void reply(PartnerLinkInstance plink, String opName, String mexId, Element msg,
+ QName fault)
+ throws FaultException;
+
+
+ String invoke(PartnerLinkInstance partnerLinkInstance,
+ Operation operation,
+ Element outboundMsg,
+ InvokeResponseChannel invokeResponseChannel) throws FaultException;
+
+
+ /**
+ * Registers a timer for future notification.
+ * @param timerChannel channel for timer notification
+ * @param timeToFire future time to fire timer notification
+ */
+ void registerTimer(TimerResponseChannel timerChannel, Date timeToFire);
+
+ /**
+ * Terminates the process / sets state flag to terminate
+ * and ceases all processing on the VPU.
+ */
+ void terminate();
+
+ /**
+ * Sends the bpel event.
+ * @param event
+ */
+ void sendEvent(ProcessInstanceEvent event);
+
+ ExpressionLanguageRuntimeRegistry getExpLangRuntime();
+
+
+ /**
+ * Generate a unique (and monotonic) ID in the context of this instance.
+ * @return
+ */
+ long genId();
+
+ Element getPartnerResponse(String mexId);
+
+ Element getMyRequest(String mexId);
+
+ QName getPartnerFault(String mexId);
+
+ String getPartnerFaultExplanation(String mexId);
+
+ QName getPartnerResponseType(String mexId);
+
+ Node getPartData(Element message, Part part);
+
+ Element getSourceEPR(String mexId);
+
+ void registerActivityForRecovery(ActivityRecoveryChannel channel, long activityId, String reason,
+ Date dateTime, Element details, String[] actions, int retries);
- void unregisterActivityForRecovery(ActivityRecoveryChannel channel);
+ void unregisterActivityForRecovery(ActivityRecoveryChannel channel);
- void recoverActivity(String channel, long activityId, String action, FaultData fault);
+ void recoverActivity(String channel, long activityId, String action, FaultData fault);
- String getSourceSessionId(String mexId);
+ String getSourceSessionId(String mexId);
+
+ void releasePartnerMex(String mexId);
}
Modified: incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/runtime/INVOKE.java
URL: http://svn.apache.org/viewvc/incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/runtime/INVOKE.java?view=diff&rev=511331&r1=511330&r2=511331
==============================================================================
--- incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/runtime/INVOKE.java (original)
+++ incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/runtime/INVOKE.java Sat Feb 24 11:34:48 2007
@@ -143,6 +143,8 @@
// TODO update output variable with data from non-initiate
// correlation sets
+
+ getBpelRuntimeContext().releasePartnerMex(mexId);
_self.parent.completed(fault, CompensationHandler.emptySet());
}
Modified: incubator/ode/trunk/dao-hibernate/src/main/java/org/apache/ode/daohib/bpel/MessageExchangeDaoImpl.java
URL: http://svn.apache.org/viewvc/incubator/ode/trunk/dao-hibernate/src/main/java/org/apache/ode/daohib/bpel/MessageExchangeDaoImpl.java?view=diff&rev=511331&r1=511330&r2=511331
==============================================================================
--- incubator/ode/trunk/dao-hibernate/src/main/java/org/apache/ode/daohib/bpel/MessageExchangeDaoImpl.java (original)
+++ incubator/ode/trunk/dao-hibernate/src/main/java/org/apache/ode/daohib/bpel/MessageExchangeDaoImpl.java Sat Feb 24 11:34:48 2007
@@ -300,4 +300,8 @@
return Collections.unmodifiableSet(_hself.getProperties().keySet());
}
+ public void release() {
+ // no-op for now, could be used to do some cleanup
+ }
+
}
Modified: incubator/ode/trunk/dao-jpa/src/main/java/org/apache/ode/dao/jpa/MessageExchangeDAOImpl.java
URL: http://svn.apache.org/viewvc/incubator/ode/trunk/dao-jpa/src/main/java/org/apache/ode/dao/jpa/MessageExchangeDAOImpl.java?view=diff&rev=511331&r1=511330&r2=511331
==============================================================================
--- incubator/ode/trunk/dao-jpa/src/main/java/org/apache/ode/dao/jpa/MessageExchangeDAOImpl.java (original)
+++ incubator/ode/trunk/dao-jpa/src/main/java/org/apache/ode/dao/jpa/MessageExchangeDAOImpl.java Sat Feb 24 11:34:48 2007
@@ -297,4 +297,8 @@
public Collection<CorrelationKey> getCorrelationKeys() {
return _correlationKeys;
}
+
+ public void release() {
+ // no-op for now, could be used to do some cleanup
+ }
}
Modified: incubator/ode/trunk/jbi/src/main/java/org/apache/ode/jbi/OdeService.java
URL: http://svn.apache.org/viewvc/incubator/ode/trunk/jbi/src/main/java/org/apache/ode/jbi/OdeService.java?view=diff&rev=511331&r1=511330&r2=511331
==============================================================================
--- incubator/ode/trunk/jbi/src/main/java/org/apache/ode/jbi/OdeService.java (original)
+++ incubator/ode/trunk/jbi/src/main/java/org/apache/ode/jbi/OdeService.java Sat Feb 24 11:34:48 2007
@@ -247,6 +247,9 @@
_jbiMexTracker.consume(jbiMex.getExchangeId());
} finally {
+ if (odeMex!= null) odeMex.release();
+ else __log.warn("Couldn't release a message exchange, it's null.");
+
if (success) {
__log.debug("Commiting ODE MEX " + odeMex);
_ode.getTransactionManager().commit();