You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ode.apache.org by ms...@apache.org on 2007/09/11 01:54:34 UTC
svn commit: r574409 - in /ode/trunk:
bpel-dao/src/main/java/org/apache/ode/bpel/dao/
bpel-runtime/src/main/java/org/apache/ode/bpel/engine/
bpel-runtime/src/main/java/org/apache/ode/bpel/memdao/
bpel-runtime/src/test/java/org/apache/ode/bpel/runtime/ d...
Author: mszefler
Date: Mon Sep 10 16:54:33 2007
New Revision: 574409
URL: http://svn.apache.org/viewvc?rev=574409&view=rev
Log:
Some fixes to ActivityRecoveryTest mocks.
Modified:
ode/trunk/bpel-dao/src/main/java/org/apache/ode/bpel/dao/MessageExchangeDAO.java
ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/engine/BpelProcess.java
ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/engine/BpelRuntimeContextImpl.java
ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/engine/BpelServerImpl.java
ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/engine/PartnerLinkMyRoleImpl.java
ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/engine/PartnerLinkPartnerRoleImpl.java
ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/memdao/MessageExchangeDAOImpl.java
ode/trunk/bpel-runtime/src/test/java/org/apache/ode/bpel/runtime/MockBpelServer.java
ode/trunk/dao-hibernate/src/main/java/org/apache/ode/daohib/bpel/MessageExchangeDaoImpl.java
ode/trunk/dao-jpa/src/main/java/org/apache/ode/dao/jpa/BPELDAOConnectionImpl.java
ode/trunk/dao-jpa/src/main/java/org/apache/ode/dao/jpa/MessageExchangeDAOImpl.java
Modified: ode/trunk/bpel-dao/src/main/java/org/apache/ode/bpel/dao/MessageExchangeDAO.java
URL: http://svn.apache.org/viewvc/ode/trunk/bpel-dao/src/main/java/org/apache/ode/bpel/dao/MessageExchangeDAO.java?rev=574409&r1=574408&r2=574409&view=diff
==============================================================================
--- ode/trunk/bpel-dao/src/main/java/org/apache/ode/bpel/dao/MessageExchangeDAO.java (original)
+++ ode/trunk/bpel-dao/src/main/java/org/apache/ode/bpel/dao/MessageExchangeDAO.java Mon Sep 10 16:54:33 2007
@@ -26,6 +26,7 @@
import org.apache.ode.bpel.iapi.InvocationStyle;
import org.apache.ode.bpel.iapi.MessageExchange.AckType;
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.w3c.dom.Element;
@@ -167,7 +168,7 @@
*/
void setPartnersKey(String correlationId);
- void setPattern(String string);
+ void setPattern(MessageExchangePattern pattern);
void setOperation(String opname);
@@ -175,7 +176,7 @@
Element getEPR();
- String getPattern();
+ MessageExchangePattern getPattern();
/**
* Get the response channel.
Modified: ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/engine/BpelProcess.java
URL: http://svn.apache.org/viewvc/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/engine/BpelProcess.java?rev=574409&r1=574408&r2=574409&view=diff
==============================================================================
--- ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/engine/BpelProcess.java (original)
+++ ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/engine/BpelProcess.java Mon Sep 10 16:54:33 2007
@@ -59,6 +59,7 @@
import org.apache.ode.bpel.iapi.ProcessConf;
import org.apache.ode.bpel.iapi.MessageExchange.AckType;
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.Scheduler.JobInfo;
@@ -225,6 +226,7 @@
return;
}
+ mexdao.setPattern((op.getOutput() == null) ? MessageExchangePattern.REQUEST_ONLY : MessageExchangePattern.REQUEST_RESPONSE);
if (!processInterceptors(mexdao, InterceptorInvoker.__onProcessInvoked)) {
__log.debug("Aborting processing of mex " + mexdao.getMessageExchangeId() + " due to interceptors.");
onMyRoleMexAck(mexdao, oldstatus);
Modified: ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/engine/BpelRuntimeContextImpl.java
URL: http://svn.apache.org/viewvc/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/engine/BpelRuntimeContextImpl.java?rev=574409&r1=574408&r2=574409&view=diff
==============================================================================
--- ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/engine/BpelRuntimeContextImpl.java (original)
+++ ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/engine/BpelRuntimeContextImpl.java Mon Sep 10 16:54:33 2007
@@ -120,8 +120,12 @@
BpelRuntimeContextImpl(BpelInstanceWorker instanceWorker, ProcessInstanceDAO dao) {
this(instanceWorker, dao, new ExecutionQueueImpl(null));
+ if (ProcessState.isFinished(dao.getState()))
+ throw new BpelEngineException("Invalid process state (process is finished)!!!");
+
// The following allows us to skip deserialization of the soup if our execution state in memory is the same
// as that in the database.
+
Object cachedState = instanceWorker.getCachedState(dao.getExecutionStateCounter());
if (cachedState != null) {
if (__log.isDebugEnabled())
@@ -131,7 +135,7 @@
_vpu.setContext(_soup);
} else {
if (__log.isDebugEnabled())
- __log.debug("CACHE MISS: Loading state to resume instance " + dao.getInstanceId() + " from database ");
+ __log.debug("CACHE MISS: state #" + dao.getExecutionStateCounter() + " is stale; loading state to resume instance " + dao.getInstanceId() + " from database ");
byte[] daoState = dao.getExecutionState();
ByteArrayInputStream iis = new ByteArrayInputStream(daoState);
try {
@@ -693,7 +697,7 @@
mexDao.setProcess(_dao.getProcess());
mexDao.setInstance(_dao);
mexDao.setPattern((operation.getOutput() != null ? MessageExchangePattern.REQUEST_RESPONSE
- : MessageExchangePattern.REQUEST_ONLY).toString());
+ : MessageExchangePattern.REQUEST_ONLY));
mexDao.setChannel(channel == null ? null : channel.export());
MessageDAO message = mexDao.createMessage(operation.getInput().getMessage().getQName());
@@ -967,9 +971,8 @@
MessageExchangeDAO mexDao = _dao.getConnection().getMessageExchange(mexId);
if (mexDao != null) {
Status status = mexDao.getStatus();
- MessageExchangePattern pattern = MessageExchange.MessageExchangePattern.valueOf(mexDao.getPattern());
InvocationStyle istyle = mexDao.getInvocationStyle();
- if (pattern == MessageExchangePattern.REQUEST_ONLY) {
+ if (mexDao.getPattern() == MessageExchangePattern.REQUEST_ONLY) {
mexDao.setAckType(AckType.ONEWAY);
mexDao.setStatus(Status.COMPLETED);
continue;
Modified: ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/engine/BpelServerImpl.java
URL: http://svn.apache.org/viewvc/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/engine/BpelServerImpl.java?rev=574409&r1=574408&r2=574409&view=diff
==============================================================================
--- ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/engine/BpelServerImpl.java (original)
+++ ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/engine/BpelServerImpl.java Mon Sep 10 16:54:33 2007
@@ -144,6 +144,11 @@
public BpelServerImpl() {
}
+ protected void waitForQuiessence() {
+ _mngmtLock.writeLock().lock();
+ _mngmtLock.writeLock().unlock();
+ }
+
public void start() {
_mngmtLock.writeLock().lock();
try {
Modified: ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/engine/PartnerLinkMyRoleImpl.java
URL: http://svn.apache.org/viewvc/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/engine/PartnerLinkMyRoleImpl.java?rev=574409&r1=574408&r2=574409&view=diff
==============================================================================
--- ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/engine/PartnerLinkMyRoleImpl.java (original)
+++ ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/engine/PartnerLinkMyRoleImpl.java Mon Sep 10 16:54:33 2007
@@ -248,7 +248,7 @@
// send process instance event
NewProcessInstanceEvent evt = new NewProcessInstanceEvent(new QName(_process.getOProcess().targetNamespace, _process
- .getOProcess().getName()), _process.getProcessDAO().getProcessId(), newInstance.getInstanceId());
+ .getOProcess().getName()), processDAO.getProcessId(), newInstance.getInstanceId());
evt.setPortType(mex.getPortType());
evt.setOperation(operation.getName());
evt.setMexId(mex.getMessageExchangeId());
Modified: ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/engine/PartnerLinkPartnerRoleImpl.java
URL: http://svn.apache.org/viewvc/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/engine/PartnerLinkPartnerRoleImpl.java?rev=574409&r1=574408&r2=574409&view=diff
==============================================================================
--- ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/engine/PartnerLinkPartnerRoleImpl.java (original)
+++ ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/engine/PartnerLinkPartnerRoleImpl.java Mon Sep 10 16:54:33 2007
@@ -37,7 +37,6 @@
import org.apache.ode.bpel.iapi.MessageExchange.MessageExchangePattern;
import org.apache.ode.bpel.iapi.MessageExchange.Status;
import org.apache.ode.bpel.o.OPartnerLink;
-import org.omg.PortableServer._ServantActivatorStub;
import org.w3c.dom.Element;
/**
@@ -114,7 +113,7 @@
Operation operation = _plinkDef.getPartnerRoleOperation(mexDao.getOperation());
Set<InvocationStyle> supportedStyles = _contexts.mexContext.getSupportedInvocationStyle(_channel, partnerEpr);
- boolean oneway = MessageExchangePattern.valueOf(mexDao.getPattern()) == MessageExchangePattern.REQUEST_ONLY;
+ boolean oneway = mexDao.getPattern() == MessageExchangePattern.REQUEST_ONLY;
try {
if (_process.isInMemory()) {
@@ -346,6 +345,26 @@
final Throwable ferr = err;
+ // For one-way invokes, theres nothing to follow up on.
+ if (_unreliableMex._operation.getOutput() == null) {
+ _unreliableMex.setState(State.DEAD);
+ // todo Perhaps we should log a failure if ferr != null
+ return;
+ }
+
+ // Handle system failure in a transaction
+ if (ferr != null) {
+ _process.enqueueInstanceTransaction(_unreliableMex.getIID(), new Runnable() {
+ public void run() {
+ MessageExchangeDAO mexdao = _process.loadMexDao(_unreliableMex.getMessageExchangeId());
+ MexDaoUtil.setFailed(mexdao, FailureType.OTHER, ferr.toString());
+ _unreliableMex.setState(State.DEAD);
+ }
+ } );
+ return;
+ }
+
+
// We proceed handling the response in a transaction. Note that if for some reason the following transaction
// fails, the unreliable invoke will be in an "unknown" state, and will require manual intervention to either
// retry or force fail.
@@ -353,10 +372,7 @@
public void run() {
MessageExchangeDAO mexdao = _process.loadMexDao(_unreliableMex.getMessageExchangeId());
- if (ferr != null) {
- MexDaoUtil.setFailed(mexdao, FailureType.OTHER, ferr.toString());
- _unreliableMex.setState(State.DEAD);
- } else if (_unreliableMex.getStatus() == Status.ACK) {
+ if (_unreliableMex.getStatus() == Status.ACK) {
_unreliableMex.save(mexdao);
_unreliableMex.setState(State.DEAD);
} else if (_unreliableMex.getStatus() == Status.REQ && !_unreliableMex._asyncReply) {
Modified: ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/memdao/MessageExchangeDAOImpl.java
URL: http://svn.apache.org/viewvc/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/memdao/MessageExchangeDAOImpl.java?rev=574409&r1=574408&r2=574409&view=diff
==============================================================================
--- ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/memdao/MessageExchangeDAOImpl.java (original)
+++ ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/memdao/MessageExchangeDAOImpl.java Mon Sep 10 16:54:33 2007
@@ -35,6 +35,7 @@
import org.apache.ode.bpel.iapi.InvocationStyle;
import org.apache.ode.bpel.iapi.MessageExchange.AckType;
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.w3c.dom.Element;
@@ -49,7 +50,7 @@
Status status;
int partnerLinkModelId;
String correlationId;
- String pattern;
+ MessageExchangePattern pattern;
Element ePR;
String channel;
QName fault;
@@ -145,8 +146,8 @@
}
- public void setPattern(String string) {
- this.pattern = string;
+ public void setPattern(MessageExchangePattern pattern) {
+ this.pattern = pattern;
}
@@ -165,7 +166,7 @@
}
- public String getPattern() {
+ public MessageExchangePattern getPattern() {
return pattern;
}
Modified: ode/trunk/bpel-runtime/src/test/java/org/apache/ode/bpel/runtime/MockBpelServer.java
URL: http://svn.apache.org/viewvc/ode/trunk/bpel-runtime/src/test/java/org/apache/ode/bpel/runtime/MockBpelServer.java?rev=574409&r1=574408&r2=574409&view=diff
==============================================================================
--- ode/trunk/bpel-runtime/src/test/java/org/apache/ode/bpel/runtime/MockBpelServer.java (original)
+++ ode/trunk/bpel-runtime/src/test/java/org/apache/ode/bpel/runtime/MockBpelServer.java Mon Sep 10 16:54:33 2007
@@ -20,9 +20,7 @@
import java.io.File;
import java.util.Collection;
-import java.util.Date;
import java.util.HashMap;
-import java.util.Map;
import java.util.Properties;
import java.util.Set;
@@ -35,7 +33,6 @@
import org.apache.ode.bpel.dao.BpelDAOConnectionFactoryJDBC;
import org.apache.ode.bpel.engine.BpelServerImpl;
import org.apache.ode.bpel.iapi.BindingContext;
-import org.apache.ode.bpel.iapi.BpelEngineException;
import org.apache.ode.bpel.iapi.ContextException;
import org.apache.ode.bpel.iapi.Endpoint;
import org.apache.ode.bpel.iapi.EndpointReference;
@@ -60,7 +57,7 @@
class MockBpelServer {
- BpelServerImpl _server;
+ DebugBpelServerImpl _server;
ProcessStoreImpl _store;
@@ -70,7 +67,7 @@
DataSource _dataSource;
- SchedulerWrapper _scheduler;
+ MockScheduler _scheduler;
BpelDAOConnectionFactory _daoCF;
@@ -86,7 +83,8 @@
public MockBpelServer() {
try {
- _server = new BpelServerImpl();
+ _server = new DebugBpelServerImpl() ;
+
createTransactionManager();
createDataSource();
createDAOConnection();
@@ -170,7 +168,8 @@
throw new RuntimeException("No transaction manager");
if (_dataSource == null)
throw new RuntimeException("No data source");
- _scheduler = new SchedulerWrapper(_server, _txManager, _dataSource);
+ _scheduler = new MockScheduler(_txManager);
+ _scheduler.setJobProcessor(_server);
return _scheduler;
}
@@ -287,53 +286,15 @@
return _bindContext;
}
- private class SchedulerWrapper implements Scheduler {
-
- MockScheduler _scheduler;
-
- long _nextSchedule;
-
- SchedulerWrapper(BpelServerImpl server, TransactionManager txManager, DataSource dataSource) {
- _scheduler = new MockScheduler(_txManager);
- _scheduler.setJobProcessor(server);
- }
-
- public String schedulePersistedJob(Map<String, Object> jobDetail, Date when) throws ContextException {
- String jobId = _scheduler.schedulePersistedJob(jobDetail, when);
- _nextSchedule = when == null ? System.currentTimeMillis() : when.getTime();
- return jobId;
- }
-
- public void cancelJob(String jobId) throws ContextException {
- _scheduler.cancelJob(jobId);
- }
-
- public void start() {
- _scheduler.start();
- }
-
- public void stop() {
- _scheduler.stop();
- }
-
- public void shutdown() {
- _scheduler.shutdown();
- }
-
- public void setJobProcessor(JobProcessor processor) throws ContextException {
- _scheduler.setJobProcessor(processor);
-
- }
-
- public void jobCompleted(String jobId) {
- _scheduler.jobCompleted(jobId);
-
- }
+
+ static class DebugBpelServerImpl extends BpelServerImpl {
+ public void waitForQuiessence() {
+ super.waitForQuiessence();
+ }
}
+
public void waitForBlocking() {
- // TODO Auto-generated method stub
-
+ _server.waitForQuiessence();
}
-
}
Modified: ode/trunk/dao-hibernate/src/main/java/org/apache/ode/daohib/bpel/MessageExchangeDaoImpl.java
URL: http://svn.apache.org/viewvc/ode/trunk/dao-hibernate/src/main/java/org/apache/ode/daohib/bpel/MessageExchangeDaoImpl.java?rev=574409&r1=574408&r2=574409&view=diff
==============================================================================
--- ode/trunk/dao-hibernate/src/main/java/org/apache/ode/daohib/bpel/MessageExchangeDaoImpl.java (original)
+++ ode/trunk/dao-hibernate/src/main/java/org/apache/ode/daohib/bpel/MessageExchangeDaoImpl.java Mon Sep 10 16:54:33 2007
@@ -33,6 +33,7 @@
import org.apache.ode.bpel.iapi.InvocationStyle;
import org.apache.ode.bpel.iapi.MessageExchange.AckType;
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.daohib.SessionManager;
import org.apache.ode.daohib.bpel.hobj.HLargeData;
@@ -133,8 +134,8 @@
update();
}
- public void setPattern(String pattern) {
- _hself.setPattern(pattern);
+ public void setPattern(MessageExchangePattern pattern) {
+ _hself.setPattern(pattern == null ? null : pattern.toString());
update();
}
@@ -192,8 +193,8 @@
}
}
- public String getPattern() {
- return _hself.getPattern();
+ public MessageExchangePattern getPattern() {
+ return _hself.getPattern() == null ? null : MessageExchangePattern.valueOf(_hself.getPattern());
}
public String getChannel() {
Modified: ode/trunk/dao-jpa/src/main/java/org/apache/ode/dao/jpa/BPELDAOConnectionImpl.java
URL: http://svn.apache.org/viewvc/ode/trunk/dao-jpa/src/main/java/org/apache/ode/dao/jpa/BPELDAOConnectionImpl.java?rev=574409&r1=574408&r2=574409&view=diff
==============================================================================
--- ode/trunk/dao-jpa/src/main/java/org/apache/ode/dao/jpa/BPELDAOConnectionImpl.java (original)
+++ ode/trunk/dao-jpa/src/main/java/org/apache/ode/dao/jpa/BPELDAOConnectionImpl.java Mon Sep 10 16:54:33 2007
@@ -71,6 +71,7 @@
public MessageExchangeDAO createMessageExchange(String mexId, char dir) {
MessageExchangeDAOImpl ret = new MessageExchangeDAOImpl(mexId, dir);
+
_em.persist(ret);
return ret;
}
Modified: ode/trunk/dao-jpa/src/main/java/org/apache/ode/dao/jpa/MessageExchangeDAOImpl.java
URL: http://svn.apache.org/viewvc/ode/trunk/dao-jpa/src/main/java/org/apache/ode/dao/jpa/MessageExchangeDAOImpl.java?rev=574409&r1=574408&r2=574409&view=diff
==============================================================================
--- ode/trunk/dao-jpa/src/main/java/org/apache/ode/dao/jpa/MessageExchangeDAOImpl.java (original)
+++ ode/trunk/dao-jpa/src/main/java/org/apache/ode/dao/jpa/MessageExchangeDAOImpl.java Mon Sep 10 16:54:33 2007
@@ -49,6 +49,7 @@
import org.apache.ode.bpel.iapi.InvocationStyle;
import org.apache.ode.bpel.iapi.MessageExchange.AckType;
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.utils.DOMUtils;
import org.w3c.dom.Element;
@@ -67,9 +68,9 @@
private String _correlationId;
@Basic @Column(name="CORRELATION_STATUS")
private String _correlationStatus;
- @Basic @Column(name="CREATE_TIME")
+ @Basic @Column(name="CREATE_TIME" , nullable = false)
private Date _createTime;
- @Basic @Column(name="DIRECTION")
+ @Basic @Column(name="DIRECTION" ,nullable=false)
private char _direction;
@Lob @Column(name="EPR")
private String _epr;
@@ -79,11 +80,11 @@
private String _fault;
@Basic @Column(name="FAULT_EXPLANATION")
private String _faultExplanation;
- @Basic @Column(name="OPERATION")
+ @Basic @Column(name="OPERATION", nullable=false)
private String _operation;
- @Basic @Column(name="PARTNER_LINK_MODEL_ID")
+ @Basic @Column(name="PARTNER_LINK_MODEL_ID", nullable=false)
private int _partnerLinkModelId;
- @Basic @Column(name="PATTERN")
+ @Basic @Column(name="PATTERN", nullable=false)
private String _pattern;
@Basic @Column(name="PORT_TYPE")
private String _portType;
@@ -132,6 +133,7 @@
public MessageExchangeDAOImpl(String mexId, char direction){
_direction = direction;
_id = mexId;
+ _createTime = new Date();
}
public MessageDAO createMessage(QName type) {
@@ -203,8 +205,8 @@
return _partnerLinkModelId;
}
- public String getPattern() {
- return _pattern;
+ public MessageExchangePattern getPattern() {
+ return _pattern == null ? null : MessageExchangePattern.valueOf(_pattern);
}
public QName getPortType() {
@@ -291,8 +293,8 @@
_partnerLinkModelId = modelId;
}
- public void setPattern(String pattern) {
- _pattern = pattern;
+ public void setPattern(MessageExchangePattern pattern) {
+ _pattern = pattern == null ? null : pattern.toString();
}
public void setPortType(QName porttype) {