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 2009/05/12 17:23:36 UTC
svn commit: r773939 [5/6] - in /ode/trunk: ./
axis2/src/main/java/org/apache/ode/axis2/
axis2/src/main/java/org/apache/ode/axis2/soapbinding/
bpel-api/src/main/java/org/apache/ode/bpel/evt/
bpel-api/src/main/java/org/apache/ode/bpel/iapi/ bpel-api/src/...
Modified: ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/ASSIGN.java
URL: http://svn.apache.org/viewvc/ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/ASSIGN.java?rev=773939&r1=773938&r2=773939&view=diff
==============================================================================
--- ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/ASSIGN.java (original)
+++ ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/ASSIGN.java Tue May 12 15:23:30 2009
@@ -198,7 +198,7 @@
} else if (ocopy.to.getVariable().type instanceof OPropertyVarType) {
// For poperty assignment, the property, the variable that points to it and the correlation set
// all have the same name
- CorrelationSetInstance csetInstance = _scopeFrame.resolve(ocopy.to.getVariable().name);
+ CorrelationSetInstance csetInstance = _scopeFrame.resolveCorrelation(ocopy.to.getVariable().name);
CorrelationKey ckey = new CorrelationKey(csetInstance.declaration.getId(), new String[] { rvalue.getTextContent() });
if (__log.isDebugEnabled()) __log.debug("Writing correlation " + csetInstance.getName()
+ " using value " + rvalue.getTextContent());
@@ -290,8 +290,9 @@
QName partName = new QName(null, part.name);
Node qualLVal = DOMUtils.findChildByName((Element) data, partName);
if (part.type instanceof OElementVarType) {
- QName elName = ((OElementVarType) part.type).elementType;
- qualLVal = DOMUtils.findChildByName((Element) qualLVal, elName);
+// QName elName = ((OElementVarType) part.type).elementType;
+// qualLVal = DOMUtils.findChildByName((Element) qualLVal, elName);
+ qualLVal = DOMUtils.getFirstChildElement((Element) qualLVal);
} else if (part.type == null) {
// Special case of header parts never referenced in the WSDL def
if (qualLVal != null && qualLVal.getNodeType() == Node.ELEMENT_NODE
Modified: ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/BoundVariableEvaluationContext.java
URL: http://svn.apache.org/viewvc/ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/BoundVariableEvaluationContext.java?rev=773939&r1=773938&r2=773939&view=diff
==============================================================================
--- ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/BoundVariableEvaluationContext.java (original)
+++ ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/BoundVariableEvaluationContext.java Tue May 12 15:23:30 2009
@@ -58,7 +58,7 @@
}
public Long getProcessId() {
- throw new InvalidProcessException("Process id not available in this context.");
+ return null;
}
public boolean narrowTypes() {
Modified: ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/EH_EVENT.java
URL: http://svn.apache.org/viewvc/ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/EH_EVENT.java?rev=773939&r1=773938&r2=773939&view=diff
==============================================================================
--- ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/EH_EVENT.java (original)
+++ ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/EH_EVENT.java Tue May 12 15:23:30 2009
@@ -20,6 +20,7 @@
import java.util.HashSet;
import java.util.Set;
+import java.util.Map;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -36,10 +37,13 @@
import org.apache.ode.bpel.rtrep.v2.channels.TerminationChannelListener;
import org.apache.ode.bpel.rapi.InvalidProcessException;
import org.apache.ode.bpel.evt.VariableModificationEvent;
+import org.apache.ode.bpel.evar.ExternalVariableModuleException;
import org.apache.ode.jacob.ChannelListener;
import org.apache.ode.jacob.SynchChannel;
+import org.apache.ode.utils.DOMUtils;
import org.w3c.dom.Element;
import org.w3c.dom.Node;
+import org.w3c.dom.Document;
/**
* Message event handler.
@@ -68,10 +72,8 @@
/** Has a termination of this handler been requested. */
private boolean _terminated;
-
private boolean _childrenTerminated;
-
EH_EVENT(ParentScopeChannel psc,TerminationChannel tc, EventHandlerControlChannel ehc, OEventHandler.OEvent o, ScopeFrame scopeFrame) {
_scopeFrame = scopeFrame;
_oevent = o;
@@ -80,9 +82,8 @@
_ehc = ehc;
}
-
public void run() {
- instance(new SELECT(_scopeFrame));
+ instance(new SELECT(_scopeFrame, 0));
}
/**
@@ -96,6 +97,7 @@
_childrenTerminated = true;
}
}
+
/**
* Template that does the actual selection interaction with the runtime system, and
* then waits on the pick response channel.
@@ -104,8 +106,11 @@
private static final long serialVersionUID = 1L;
- public SELECT(ScopeFrame scopeFrame) {
+ private int _counter;
+
+ public SELECT(ScopeFrame scopeFrame, int counter) {
_scopeFrame = scopeFrame;
+ _counter = counter;
}
/**
@@ -115,30 +120,36 @@
Selector selector;
try {
PickResponseChannel pickResponseChannel = newChannel(PickResponseChannel.class);
- CorrelationKey key;
- PartnerLinkInstance pLinkInstance = _scopeFrame.resolve(_oevent.partnerLink);
- if (_oevent.matchCorrelation == null) {
- // Adding a route for opaque correlation. In this case correlation is done on "out-of-band" session id.
- String sessionId = getBpelRuntime().fetchMySessionId(pLinkInstance);
- key = new CorrelationKey(-1, new String[] {sessionId});
+ if (_oevent.isRestful()) {
+ getBpelRuntime().checkResourceRoute(_scopeFrame.resolve(_oevent.resource),
+ _oevent.messageExchangeId+_counter, pickResponseChannel, 0);
} else {
- if (!getBpelRuntime().isCorrelationInitialized(_scopeFrame.resolve(_oevent.matchCorrelation))) {
- throw new FaultException(_oevent.getOwner().constants.qnCorrelationViolation,"Correlation not initialized.");
+ CorrelationKey key;
+ PartnerLinkInstance pLinkInstance = _scopeFrame.resolve(_oevent.partnerLink);
+ if (_oevent.matchCorrelation == null) {
+ // Adding a route for opaque correlation. In this case correlation is done on "out-of-band" session id.
+ String sessionId = getBpelRuntime().fetchMySessionId(pLinkInstance);
+ key = new CorrelationKey(-1, new String[] {sessionId});
+ } else {
+ if (!getBpelRuntime().isCorrelationInitialized(_scopeFrame.resolve(_oevent.matchCorrelation))) {
+ throw new FaultException(_oevent.getOwner().constants.qnCorrelationViolation,"Correlation not initialized.");
+ }
+ key = getBpelRuntime().readCorrelation(_scopeFrame.resolve(_oevent.matchCorrelation));
+ assert key != null;
}
- key = getBpelRuntime().readCorrelation(_scopeFrame.resolve(_oevent.matchCorrelation));
- assert key != null;
- }
- selector = new Selector(0,pLinkInstance,_oevent.operation.getName(), _oevent.operation.getOutput() == null, _oevent.messageExchangeId, key);
- getBpelRuntime().select(pickResponseChannel, null, false, new Selector[] { selector} );
- instance(new WAITING(pickResponseChannel, _scopeFrame));
+ selector = new Selector(0,pLinkInstance,_oevent.operation.getName(),
+ _oevent.operation.getOutput() == null, _oevent.messageExchangeId+_counter, key);
+ getBpelRuntime().select(pickResponseChannel, null, false, new Selector[] { selector} );
+ }
+ instance(new WAITING(pickResponseChannel, _scopeFrame, _counter));
} catch(FaultException e){
__log.error(e);
if (_fault == null) {
_fault = createFault(e.getQName(), _oevent);
}
terminateActive();
- instance(new WAITING(null, _scopeFrame));
+ instance(new WAITING(null, _scopeFrame, _counter));
}
}
}
@@ -149,10 +160,12 @@
private class WAITING extends BpelJacobRunnable {
private static final long serialVersionUID = 1L;
private PickResponseChannel _pickResponseChannel;
+ private int _counter;
- private WAITING(PickResponseChannel pickResponseChannel, ScopeFrame scopeFrame) {
+ private WAITING(PickResponseChannel pickResponseChannel, ScopeFrame scopeFrame, int counter) {
_pickResponseChannel = pickResponseChannel;
_scopeFrame = scopeFrame;
+ _counter = counter;
}
public void run() {
@@ -163,7 +176,6 @@
if (!_terminated) {
mlset.add(new TerminationChannelListener(_tc) {
private static final long serialVersionUID = 7666910462948788042L;
-
public void terminate() {
terminateActive();
_terminated = true;
@@ -172,13 +184,11 @@
instance(WAITING.this);
}
});
-
}
if (!_stopped) {
mlset.add(new EventHandlerControlChannelListener(_ehc) {
private static final long serialVersionUID = -1050788954724647970L;
-
public void stop() {
_stopped = true;
if (_pickResponseChannel != null)
@@ -186,18 +196,15 @@
instance(WAITING.this);
}
});
-
}
for (final ActivityInfo ai : _active) {
mlset.add(new ParentScopeChannelListener(ai.parent) {
private static final long serialVersionUID = 5341207762415360982L;
-
public void compensate(OScope scope, SynchChannel ret) {
_psc.compensate(scope, ret);
instance(WAITING.this);
}
-
public void completed(FaultData faultData, Set<CompensationHandler> compensations) {
_active.remove(ai);
_comps.addAll(compensations);
@@ -208,7 +215,6 @@
} else
instance(WAITING.this);
}
-
public void cancelled() { completed(null, CompensationHandler.emptySet()); }
public void failure(String reason, Element data) { completed(null, CompensationHandler.emptySet()); }
});
@@ -218,14 +224,19 @@
mlset.add(new PickResponseChannelListener(_pickResponseChannel) {
private static final long serialVersionUID = -4929999153478677288L;
-
public void onRequestRcvd(int selectorIdx, String mexId) {
// The receipt of the message causes a new scope to be created:
ScopeFrame ehScopeFrame = new ScopeFrame(_oevent,
getBpelRuntime().createScopeInstance(_scopeFrame.scopeInstanceId, _oevent),
- _scopeFrame,
- _comps,
- _fault);
+ _scopeFrame, _comps, _fault);
+ ehScopeFrame.eventScope = true;
+ if (_oevent.isRestful()) {
+ getBpelRuntime().associateEvent(_scopeFrame.resolve(_oevent.resource),
+ _oevent.messageExchangeId+_counter, _oevent.messageExchangeId+ehScopeFrame.scopeInstanceId);
+ } else {
+ getBpelRuntime().associateEvent(_scopeFrame.resolve(_oevent.partnerLink), _oevent.operation.getName(),
+ _oevent.messageExchangeId+_counter, _oevent.messageExchangeId+ehScopeFrame.scopeInstanceId);
+ }
if (_oevent.variable != null) {
Element msgEl = getBpelRuntime().getMyRequest(mexId);
@@ -248,13 +259,35 @@
}
}
+ if (_oevent.isRestful()) {
+ // Retrieving the map of properties associated with a RESTful mex. Those are the query
+ // parameters and are used to set implicit variables associated with these parameters
+ // in the process.
+ Map<String,String> props = getBpelRuntime().getProperties(mexId);
+ for (Map.Entry<String, String> entry : props.entrySet()) {
+ VariableInstance vi = _scopeFrame.resolveVariable(entry.getKey());
+ if (vi != null) {
+ // Always expected to be a string
+ Document doc = DOMUtils.newDocument();
+ Node textNode = doc.createTextNode(entry.getValue());
+ try {
+ initializeVariable(vi, textNode);
+ } catch (ExternalVariableModuleException e) {
+ __log.error("Exception while initializing external variable", e);
+ _psc.failure(e.toString(), null);
+ return;
+ }
+ }
+ }
+
+ }
try {
for (OScope.CorrelationSet cset : _oevent.initCorrelations) {
initializeCorrelation(ehScopeFrame.resolve(cset), ehScopeFrame.resolve(_oevent.variable));
}
- if (_oevent.partnerLink.hasPartnerRole()) {
+ if (_oevent.partnerLink != null && _oevent.partnerLink.hasPartnerRole()) {
// Trying to initialize partner epr based on a message-provided epr/session.
if (!getBpelRuntime().isPartnerRoleEndpointInitialized(ehScopeFrame
.resolve(_oevent.partnerLink)) || !_oevent.partnerLink.initializePartnerRole) {
@@ -271,28 +304,21 @@
partnersSessionId);
}
-
-
-
} catch (FaultException e) {
__log.error(e);
if (_fault == null) {
_fault = createFault(e.getQName(), _oevent);
terminateActive();
}
- instance(new WAITING(null, _scopeFrame));
+ instance(new WAITING(null, _scopeFrame, _counter));
return;
}
-
-
// load 'onMessage' activity; we'll do this even if a stop/terminate has been
// requested becasue we cannot undo the receipt of the message at this point.
- ActivityInfo child = new ActivityInfo(genMonotonic(),
- _oevent.activity,
+ ActivityInfo child = new ActivityInfo(genMonotonic(), _oevent.activity,
newChannel(TerminationChannel.class), newChannel(ParentScopeChannel.class));
-
_active.add(child);
LinkFrame lf = new LinkFrame(null);
@@ -301,23 +327,20 @@
// If we previously terminated the other activiites, then we do the same
// here; this is easier then undoing the receive.
- if (_childrenTerminated)
- replication(child.self).terminate();
-
+ if (_childrenTerminated) replication(child.self).terminate();
if (_terminated || _stopped || _fault != null)
- instance(new WAITING(null, _scopeFrame));
+ instance(new WAITING(null, _scopeFrame, _counter));
else
- instance(new SELECT(_scopeFrame));
+ instance(new SELECT(_scopeFrame, _counter+1));
}
-
public void onTimeout() {
- instance(new WAITING(null, _scopeFrame));
+ instance(new WAITING(null, _scopeFrame, _counter));
}
public void onCancel() {
- instance(new WAITING(null, _scopeFrame));
+ instance(new WAITING(null, _scopeFrame, _counter));
}
});
Modified: ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/ExprEvaluationContextImpl.java
URL: http://svn.apache.org/viewvc/ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/ExprEvaluationContextImpl.java?rev=773939&r1=773938&r2=773939&view=diff
==============================================================================
--- ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/ExprEvaluationContextImpl.java (original)
+++ ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/ExprEvaluationContextImpl.java Tue May 12 15:23:30 2009
@@ -74,7 +74,7 @@
if (variable.type instanceof OConstantVarType) {
ret = ((OConstantVarType) variable.type).getValue();
} else if (variable.type instanceof OPropertyVarType) {
- CorrelationSetInstance set = _scopeInstance.resolve(variable.name);
+ CorrelationSetInstance set = _scopeInstance.resolveCorrelation(variable.name);
CorrelationKey key = _native.readCorrelation(set);
if (key == null) return null;
String keyValue = key.getValues()[0];
@@ -118,7 +118,7 @@
}
public Long getProcessId() {
- return _native.getPid();
+ return _native.getInstanceId();
}
public boolean narrowTypes() {
Modified: ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/ExtensionContextImpl.java
URL: http://svn.apache.org/viewvc/ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/ExtensionContextImpl.java?rev=773939&r1=773938&r2=773939&view=diff
==============================================================================
--- ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/ExtensionContextImpl.java (original)
+++ ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/ExtensionContextImpl.java Tue May 12 15:23:30 2009
@@ -57,7 +57,7 @@
}
public Long getProcessId() {
- return _context.getPid();
+ return _context.getInstanceId();
}
public Map<String, OScope.Variable> getVisibleVariables() throws FaultException {
Modified: ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/INVOKE.java
URL: http://svn.apache.org/viewvc/ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/INVOKE.java?rev=773939&r1=773938&r2=773939&view=diff
==============================================================================
--- ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/INVOKE.java (original)
+++ ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/INVOKE.java Tue May 12 15:23:30 2009
@@ -26,18 +26,13 @@
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.ode.bpel.common.FaultException;
-import org.apache.ode.bpel.evt.ActivityFailureEvent;
-import org.apache.ode.bpel.evt.ActivityRecoveryEvent;
import org.apache.ode.bpel.evt.VariableModificationEvent;
-import org.apache.ode.bpel.rtrep.v2.channels.ActivityRecoveryChannel;
-import org.apache.ode.bpel.rtrep.v2.channels.ActivityRecoveryChannelListener;
import org.apache.ode.bpel.rtrep.v2.channels.FaultData;
import org.apache.ode.bpel.rtrep.v2.channels.InvokeResponseChannel;
import org.apache.ode.bpel.rtrep.v2.channels.InvokeResponseChannelListener;
import org.apache.ode.bpel.rtrep.v2.channels.TerminationChannelListener;
-import org.apache.ode.bpel.rtrep.v2.channels.TimerResponseChannel;
-import org.apache.ode.bpel.rtrep.v2.channels.TimerResponseChannelListener;
import org.apache.ode.bpel.evar.ExternalVariableModuleException;
+import org.apache.ode.bpel.iapi.Resource;
import org.w3c.dom.Element;
import org.w3c.dom.Node;
@@ -49,19 +44,10 @@
private static final Log __log = LogFactory.getLog(INVOKE.class);
private OInvoke _oinvoke;
- // Records number of invocations on the activity.
- private int _invoked;
- // Date/time of last failure.
- private Date _lastFailure;
- // Reason for last failure.
- private String _failureReason;
- // Data associated with failure.
- private Element _failureData;
public INVOKE(ActivityInfo self, ScopeFrame scopeFrame, LinkFrame linkFrame) {
super(self, scopeFrame, linkFrame);
_oinvoke = (OInvoke) _self.o;
- _invoked = 0;
}
public final void run() {
@@ -78,8 +64,33 @@
_self.parent.failure(e.toString(), null);
return;
}
- ++_invoked;
+ if (_oinvoke.isRestful()) {
+ restInvoke(outboundMsg);
+ } else {
+ wsdlInvoke(outboundMsg);
+ }
+ }
+
+ private void restInvoke(Element outboundMsg) {
+ VariableInstance outputVar = null;
+ if (_oinvoke.outputVar != null) outputVar = _scopeFrame.resolve(_oinvoke.outputVar);
+ InvokeResponseChannel invokeResponseChannel = newChannel(InvokeResponseChannel.class);
+ try {
+ String path = getBpelRuntime().getExpLangRuntime()
+ .evaluateAsString(_oinvoke.resource.getSubpath(), getEvaluationContext());
+ String mexId = getBpelRuntime().invoke(invokeResponseChannel.export(),
+ new Resource(path, "application/xml", _oinvoke.resource.getMethod()), outboundMsg);
+ setupListeners(mexId, invokeResponseChannel, outputVar);
+ } catch (FaultException fault) {
+ __log.error(fault);
+ FaultData faultData = createFault(fault.getQName(), _oinvoke, fault.getMessage());
+ _self.parent.completed(faultData, CompensationHandler.emptySet());
+ }
+
+ }
+
+ private void wsdlInvoke(Element outboundMsg) {
// if there is no output variable, then this is a one-way invoke
boolean isTwoWay = _oinvoke.outputVar != null;
@@ -87,7 +98,7 @@
if (!isTwoWay) {
FaultData faultData = null;
getBpelRuntime().invoke(null, _scopeFrame.resolve(_oinvoke.partnerLink),
- _oinvoke.operation, outboundMsg, null);
+ _oinvoke.operation, outboundMsg, null);
_self.parent.completed(faultData, CompensationHandler.emptySet());
} else /* two-way */{
@@ -95,129 +106,141 @@
final InvokeResponseChannel invokeResponseChannel = newChannel(InvokeResponseChannel.class);
final String mexId = getBpelRuntime().invoke(invokeResponseChannel.export(),
- _scopeFrame.resolve(_oinvoke.partnerLink), _oinvoke.operation,
- outboundMsg, invokeResponseChannel);
-
- object(false, new InvokeResponseChannelListener(invokeResponseChannel) {
- private static final long serialVersionUID = 4496880438819196765L;
+ _scopeFrame.resolve(_oinvoke.partnerLink), _oinvoke.operation,
+ outboundMsg, invokeResponseChannel);
- public void onResponse() {
- // we don't have to write variable data -> this already
- // happened in the nativeAPI impl
- FaultData fault = null;
-
- Element response;
- try {
- response = getBpelRuntime().getPartnerResponse(mexId);
- } catch (Exception e) {
- __log.error(e);
- // TODO: Better error handling
- throw new RuntimeException(e);
- }
+ setupListeners(mexId, invokeResponseChannel, outputVar);
+ }
+ } catch (FaultException fault) {
+ __log.error(fault);
+ FaultData faultData = createFault(fault.getQName(), _oinvoke, fault.getMessage());
+ _self.parent.completed(faultData, CompensationHandler.emptySet());
+ }
+ }
- try {
- initializeVariable(outputVar, response);
- } catch (ExternalVariableModuleException e) {
- __log.error("Exception while initializing external variable", e);
- _self.parent.failure(e.toString(), null);
- return;
- }
-
- // Generating event
- VariableModificationEvent se = new VariableModificationEvent(outputVar.declaration.name);
- se.setNewValue(response);
- if (_oinvoke.debugInfo != null)
- se.setLineNo(_oinvoke.debugInfo.startLine);
- sendEvent(se);
-
- try {
- for (OScope.CorrelationSet anInitCorrelationsOutput : _oinvoke.initCorrelationsOutput) {
- initializeCorrelation(_scopeFrame.resolve(anInitCorrelationsOutput), outputVar);
- }
- if (_oinvoke.partnerLink.hasPartnerRole()) {
- // Trying to initialize partner epr based on a message-provided epr/session.
- if (!getBpelRuntime().isPartnerRoleEndpointInitialized(_scopeFrame
- .resolve(_oinvoke.partnerLink)) || !_oinvoke.partnerLink.initializePartnerRole) {
-
- Node fromEpr = getBpelRuntime().getSourceEPR(mexId);
- if (fromEpr != null) {
- getBpelRuntime().writeEndpointReference(
- _scopeFrame.resolve(_oinvoke.partnerLink), (Element) fromEpr);
- }
- }
-
- String partnersSessionId = getBpelRuntime().getSourceSessionId(mexId);
- if (partnersSessionId != null)
- getBpelRuntime().initializePartnersSessionId(_scopeFrame.resolve(_oinvoke.partnerLink),
- partnersSessionId);
-
+ private void setupListeners(final String mexId, final InvokeResponseChannel invokeResponseChannel, final VariableInstance outputVar) {
+ object(false, new InvokeResponseChannelListener(invokeResponseChannel) {
+ private static final long serialVersionUID = 4496880438819196765L;
+
+ public void onResponse() {
+ // we don't have to write variable data -> this already
+ // happened in the nativeAPI impl
+ FaultData fault = null;
+
+ Element response = null;
+ try {
+ response = getBpelRuntime().getPartnerResponse(mexId);
+ } catch (Exception e) {
+ // In RESTful invokes, we discover an empty response after the fact (204), so we could
+ // very well have no response here.
+ __log.debug(e);
+ }
+
+ if (outputVar != null && response ==null) {
+ String msg = "The process is expected to set an output variable after an invocation but " +
+ "there's no response provided by the partner. The process is going to be failed.";
+ __log.info(msg);
+ _self.parent.failure(msg, null);
+ }
+
+ try {
+ if (outputVar != null) initializeVariable(outputVar, response);
+ } catch (ExternalVariableModuleException e) {
+ __log.error("Exception while initializing external variable", e);
+ _self.parent.failure(e.toString(), null);
+ return;
+ }
+
+ // Generating event
+ if (outputVar != null) {
+ VariableModificationEvent se = new VariableModificationEvent(outputVar.declaration.name);
+ se.setNewValue(response);
+ if (_oinvoke.debugInfo != null)
+ se.setLineNo(_oinvoke.debugInfo.startLine);
+ sendEvent(se);
+ }
+
+ try {
+ for (OScope.CorrelationSet anInitCorrelationsOutput : _oinvoke.initCorrelationsOutput) {
+ initializeCorrelation(_scopeFrame.resolve(anInitCorrelationsOutput), outputVar);
+ }
+ if (_oinvoke.partnerLink != null && _oinvoke.partnerLink.hasPartnerRole()) {
+ // Trying to initialize partner epr based on a message-provided epr/session.
+ if (!getBpelRuntime().isPartnerRoleEndpointInitialized(_scopeFrame
+ .resolve(_oinvoke.partnerLink)) || !_oinvoke.partnerLink.initializePartnerRole) {
+
+ Node fromEpr = getBpelRuntime().getSourceEPR(mexId);
+ if (fromEpr != null) {
+ getBpelRuntime().writeEndpointReference(
+ _scopeFrame.resolve(_oinvoke.partnerLink), (Element) fromEpr);
}
- } catch (FaultException e) {
- fault = createFault(e.getQName(), _oinvoke);
}
- // TODO update output variable with data from non-initiate correlation sets
-
- _self.parent.completed(fault, CompensationHandler.emptySet());
- getBpelRuntime().releasePartnerMex(mexId, fault == null);
+ String partnersSessionId = getBpelRuntime().getSourceSessionId(mexId);
+ if (partnersSessionId != null)
+ getBpelRuntime().initializePartnersSessionId(
+ _scopeFrame.resolve(_oinvoke.partnerLink), partnersSessionId);
}
+ } catch (FaultException e) {
+ fault = createFault(e.getQName(), _oinvoke);
+ }
+
+ // TODO update output variable with data from non-initiate correlation sets
+
+ _self.parent.completed(fault, CompensationHandler.emptySet());
+ getBpelRuntime().releasePartnerMex(mexId, fault == null);
+ }
+
+ public void onFault() {
+ QName faultName = getBpelRuntime().getPartnerFault(mexId);
+ Element msg = getBpelRuntime().getPartnerResponse(mexId);
+ QName msgType = getBpelRuntime().getPartnerResponseType(mexId);
+ FaultData fault = createFault(faultName, msg,
+ _oinvoke.getOwner().messageTypes.get(msgType), _self.o);
+ _self.parent.completed(fault, CompensationHandler.emptySet());
+ getBpelRuntime().releasePartnerMex(mexId, false);
+ }
+
+ public void onFailure() {
+ // This indicates a communication failure. We don't throw a fault,
+ // because there is no fault, instead we'll re-incarnate the invoke
+ // and either retry or indicate failure condition.
+ // admin to resume the process.
+ _self.parent.failure(getBpelRuntime().getPartnerFaultExplanation(mexId), null);
+ getBpelRuntime().releasePartnerMex(mexId, false);
+ }
- public void onFault() {
- QName faultName = getBpelRuntime().getPartnerFault(mexId);
- Element msg = getBpelRuntime().getPartnerResponse(mexId);
- QName msgType = getBpelRuntime().getPartnerResponseType(mexId);
- FaultData fault = createFault(faultName, msg,
- _oinvoke.getOwner().messageTypes.get(msgType), _self.o);
- _self.parent.completed(fault, CompensationHandler.emptySet());
- getBpelRuntime().releasePartnerMex(mexId, false);
- }
+ }.or(new TerminationChannelListener(_self.self) {
+ private static final long serialVersionUID = 4219496341785922396L;
+ public void terminate() {
+ _self.parent.completed(null, CompensationHandler.emptySet());
+ object(new InvokeResponseChannelListener(invokeResponseChannel) {
+ private static final long serialVersionUID = 688746737897792929L;
public void onFailure() {
- // This indicates a communication failure. We don't throw a fault,
- // because there is no fault, instead we'll re-incarnate the invoke
- // and either retry or indicate failure condition.
- // admin to resume the process.
- _self.parent.failure(getBpelRuntime().getPartnerFaultExplanation(mexId), null);
- getBpelRuntime().releasePartnerMex(mexId, false);
+ __log.debug("Failure on invoke ignored, the invoke has already been terminated: " + _oinvoke.toString());
}
-
- }.or(new TerminationChannelListener(_self.self) {
- private static final long serialVersionUID = 4219496341785922396L;
-
- public void terminate() {
- _self.parent.completed(null, CompensationHandler.emptySet());
- object(new InvokeResponseChannelListener(invokeResponseChannel) {
- private static final long serialVersionUID = 688746737897792929L;
- public void onFailure() {
- __log.debug("Failure on invoke ignored, the invoke has already been terminated: " + _oinvoke.toString());
- }
- public void onFault() {
- __log.debug("Fault on invoke ignored, the invoke has already been terminated: " + _oinvoke.toString());
- }
- public void onResponse() {
- __log.debug("Response on invoke ignored, the invoke has already been terminated: " + _oinvoke.toString());
- }
- });
- }
- }));
-
+ public void onFault() {
+ __log.debug("Fault on invoke ignored, the invoke has already been terminated: " + _oinvoke.toString());
+ }
+ public void onResponse() {
+ __log.debug("Response on invoke ignored, the invoke has already been terminated: " + _oinvoke.toString());
+ }
+ });
}
- } catch (FaultException fault) {
- __log.error(fault);
- FaultData faultData = createFault(fault.getQName(), _oinvoke, fault.getMessage());
- _self.parent.completed(faultData, CompensationHandler.emptySet());
- }
+ }));
}
private Element setupOutbound(OInvoke oinvoke, Collection<OScope.CorrelationSet> outboundInitiations)
- throws FaultException, ExternalVariableModuleException {
+ throws FaultException, ExternalVariableModuleException {
if (outboundInitiations.size() > 0) {
for (OScope.CorrelationSet c : outboundInitiations) {
initializeCorrelation(_scopeFrame.resolve(c), _scopeFrame.resolve(oinvoke.inputVar));
}
}
- if (oinvoke.operation.getInput().getMessage().getParts().size() > 0) {
+ if ((oinvoke.operation != null && oinvoke.operation.getInput().getMessage().getParts().size() > 0)
+ || (oinvoke.isRestful() && oinvoke.inputVar != null)) {
sendVariableReadEvent(_scopeFrame.resolve(oinvoke.inputVar));
Node outboundMsg = fetchVariableData(_scopeFrame.resolve(oinvoke.inputVar), false);
// TODO outbound message should be updated with non-initiate correlation sets
@@ -226,96 +249,4 @@
} else return null;
}
- private void retryOrFailure(String reason, Element data) {
- _lastFailure = new Date();
- _failureReason = reason;
- _failureData = data;
-
- OFailureHandling failureHandling = _oinvoke.getFailureHandling();
- if (failureHandling != null && failureHandling.faultOnFailure) {
- // No attempt to retry or enter activity recovery state, simply fault.
- if (__log.isDebugEnabled())
- __log.debug("ActivityRecovery: Invoke activity " + _self.aId + " faulting on failure");
- FaultData faultData = createFault(OFailureHandling.FAILURE_FAULT_NAME, _oinvoke, reason);
- _self.parent.completed(faultData, CompensationHandler.emptySet());
- return;
- }
- // If maximum number of retries, enter activity recovery state.
- if (failureHandling == null || _invoked > failureHandling.retryFor) {
- requireRecovery();
- return;
- }
-
- if (__log.isDebugEnabled())
- __log.debug("ActivityRecovery: Retrying invoke activity " + _self.aId);
- Date future = new Date(new Date().getTime() + failureHandling.retryDelay * 1000);
- final TimerResponseChannel timerChannel = newChannel(TimerResponseChannel.class);
- getBpelRuntime().registerTimer(timerChannel, future);
-
- object(false, new TimerResponseChannelListener(timerChannel) {
- private static final long serialVersionUID = -261911108068231376L;
- public void onTimeout() {
- instance(INVOKE.this);
- }
- public void onCancel() {
- INVOKE.this.requireRecovery();
- }
- }.or(new TerminationChannelListener(_self.self) {
- private static final long serialVersionUID = -4416795170896911290L;
-
- public void terminate() {
- _self.parent.completed(null, CompensationHandler.emptySet());
- object(new TimerResponseChannelListener(timerChannel) {
- private static final long serialVersionUID = 4822348066868313717L;
- public void onTimeout() { }
- public void onCancel() { }
- });
- }
- }));
- }
-
- private void requireRecovery() {
- if (__log.isDebugEnabled())
- __log.debug("ActivityRecovery: Invoke activity " + _self.aId + " requires recovery");
- sendEvent(new ActivityFailureEvent(_failureReason));
- final ActivityRecoveryChannel recoveryChannel = newChannel(ActivityRecoveryChannel.class);
- getBpelRuntime().registerActivityForRecovery(recoveryChannel, _self.aId, _failureReason, _lastFailure, _failureData,
- new String[] { "retry", "cancel", "fault" }, _invoked - 1);
- object(false, new ActivityRecoveryChannelListener(recoveryChannel) {
- private static final long serialVersionUID = 8397883882810521685L;
- public void retry() {
- if (__log.isDebugEnabled())
- __log.debug("ActivityRecovery: Retrying invoke activity " + _self.aId + " (user initiated)");
- sendEvent(new ActivityRecoveryEvent("retry"));
- getBpelRuntime().unregisterActivityForRecovery(recoveryChannel);
- instance(INVOKE.this);
- }
- public void cancel() {
- if (__log.isDebugEnabled())
- __log.debug("ActivityRecovery: Cancelling invoke activity " + _self.aId + " (user initiated)");
- sendEvent(new ActivityRecoveryEvent("cancel"));
- getBpelRuntime().unregisterActivityForRecovery(recoveryChannel);
- _self.parent.cancelled();
- }
- public void fault(FaultData faultData) {
- if (__log.isDebugEnabled())
- __log.debug("ActivityRecovery: Faulting invoke activity " + _self.aId + " (user initiated)");
- sendEvent(new ActivityRecoveryEvent("fault"));
- getBpelRuntime().unregisterActivityForRecovery(recoveryChannel);
- if (faultData == null)
- faultData = createFault(OFailureHandling.FAILURE_FAULT_NAME, _self.o, _failureReason);
- _self.parent.completed(faultData, CompensationHandler.emptySet());
- }
- }.or(new TerminationChannelListener(_self.self) {
- private static final long serialVersionUID = 2148587381204858397L;
-
- public void terminate() {
- if (__log.isDebugEnabled())
- __log.debug("ActivityRecovery: Cancelling invoke activity " + _self.aId + " (terminated by scope)");
- getBpelRuntime().unregisterActivityForRecovery(recoveryChannel);
- _self.parent.completed(null, CompensationHandler.emptySet());
- }
- }));
- }
-
}
Added: ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/OCollect.java
URL: http://svn.apache.org/viewvc/ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/OCollect.java?rev=773939&view=auto
==============================================================================
--- ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/OCollect.java (added)
+++ ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/OCollect.java Tue May 12 15:23:30 2009
@@ -0,0 +1,30 @@
+package org.apache.ode.bpel.rtrep.v2;
+
+/**
+ * RESTful read activity. Provides an idempotent GET method on a given resource.
+ */
+public class OCollect extends OActivity {
+
+ private OResource resource;
+ private OScope.Variable variable;
+
+ public OCollect(OProcess owner, OActivity parent) {
+ super(owner, parent);
+ }
+
+ public OResource getResource() {
+ return resource;
+ }
+
+ public void setResource(OResource resource) {
+ this.resource = resource;
+ }
+
+ public OScope.Variable getVariable() {
+ return variable;
+ }
+
+ public void setVariable(OScope.Variable variable) {
+ this.variable = variable;
+ }
+}
Added: ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/OComm.java
URL: http://svn.apache.org/viewvc/ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/OComm.java?rev=773939&view=auto
==============================================================================
--- ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/OComm.java (added)
+++ ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/OComm.java Tue May 12 15:23:30 2009
@@ -0,0 +1,12 @@
+package org.apache.ode.bpel.rtrep.v2;
+
+import javax.wsdl.Operation;
+
+public interface OComm {
+
+ OPartnerLink getPartnerLink();
+ Operation getOperation();
+ OResource getResource();
+
+ boolean isRestful();
+}
Modified: ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/OConstants.java
URL: http://svn.apache.org/viewvc/ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/OConstants.java?rev=773939&r1=773938&r2=773939&view=diff
==============================================================================
--- ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/OConstants.java (original)
+++ ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/OConstants.java Tue May 12 15:23:30 2009
@@ -18,13 +18,15 @@
*/
package org.apache.ode.bpel.rtrep.v2;
+import org.apache.ode.bpel.rapi.ConstantsModel;
+
import javax.xml.namespace.QName;
/**
* Compiled BPEL constants. Mostly the qualified names of the standard
* faults.
*/
-public class OConstants extends OBase {
+public class OConstants extends OBase implements ConstantsModel {
private static final long serialVersionUID = 1L;
@@ -55,4 +57,76 @@
super(owner);
}
+ public QName getConflictingReceive() {
+ return qnConflictingReceive;
+ }
+
+ public QName getCorrelationViolation() {
+ return qnCorrelationViolation;
+ }
+
+ public QName getDuplicateInstance() {
+ return qnDuplicateInstance;
+ }
+
+ public QName getForEachCounterError() {
+ return qnForEachCounterError;
+ }
+
+ public QName getForcedTermination() {
+ return qnForcedTermination;
+ }
+
+ public QName getInvalidBranchCondition() {
+ return qnInvalidBranchCondition;
+ }
+
+ public QName getInvalidExpressionValue() {
+ return qnInvalidExpressionValue;
+ }
+
+ public QName getJoinFailure() {
+ return qnJoinFailure;
+ }
+
+ public QName getMismatchedAssignmentFailure() {
+ return qnMismatchedAssignmentFailure;
+ }
+
+ public QName getMissingReply() {
+ return qnMissingReply;
+ }
+
+ public QName getMissingRequest() {
+ return qnMissingRequest;
+ }
+
+ public QName getRetiredProcess() {
+ return qnRetiredProcess;
+ }
+
+ public QName getSelectionFailure() {
+ return qnSelectionFailure;
+ }
+
+ public QName getSubLanguageExecutionFault() {
+ return qnSubLanguageExecutionFault;
+ }
+
+ public QName getUninitializedPartnerRole() {
+ return qnUninitializedPartnerRole;
+ }
+
+ public QName getUninitializedVariable() {
+ return qnUninitializedVariable;
+ }
+
+ public QName getUnknownFault() {
+ return qnUnknownFault;
+ }
+
+ public QName getXsltInvalidSource() {
+ return qnXsltInvalidSource;
+ }
+
}
Modified: ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/OEventHandler.java
URL: http://svn.apache.org/viewvc/ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/OEventHandler.java?rev=773939&r1=773938&r2=773939&view=diff
==============================================================================
--- ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/OEventHandler.java (original)
+++ ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/OEventHandler.java Tue May 12 15:23:30 2009
@@ -49,7 +49,7 @@
}
}
- public static class OEvent extends OScope {
+ public static class OEvent extends OScope implements OComm {
static final long serialVersionUID = -1L ;
/** Correlations to initialize. */
@@ -58,6 +58,7 @@
/** Correlation set to match on. */
public OScope.CorrelationSet matchCorrelation;
+ public OResource resource;
public OPartnerLink partnerLink;
public Operation operation;
public OScope.Variable variable;
@@ -65,7 +66,6 @@
/** OASIS addition for disambiguating receives (optional). */
public String messageExchangeId = "";
-
public String getCorrelatorId() {
return partnerLink.getId() + "." + operation.getName();
}
@@ -73,5 +73,22 @@
public OEvent(OProcess owner, OActivity parent) {
super(owner, parent);
}
+
+ public Operation getOperation() {
+ return operation;
+ }
+
+ public OPartnerLink getPartnerLink() {
+ return partnerLink;
+ }
+
+ public OResource getResource() {
+ return resource;
+ }
+
+ public boolean isRestful() {
+ return partnerLink == null && resource != null;
+ }
+
}
}
Modified: ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/OInvoke.java
URL: http://svn.apache.org/viewvc/ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/OInvoke.java?rev=773939&r1=773938&r2=773939&view=diff
==============================================================================
--- ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/OInvoke.java (original)
+++ ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/OInvoke.java Tue May 12 15:23:30 2009
@@ -26,13 +26,14 @@
/**
* Compiled rerpresentation of the BPEL <code><invoke></code> activity.
*/
-public class OInvoke extends OActivity {
+public class OInvoke extends OActivity implements OComm {
static final long serialVersionUID = -1L ;
public OPartnerLink partnerLink;
public OScope.Variable inputVar;
public OScope.Variable outputVar;
public Operation operation;
+ public OResource resource;
/** Correlation sets initialized on the input message. */
public final List<OScope.CorrelationSet> initCorrelationsInput = new ArrayList<OScope.CorrelationSet>();
@@ -49,4 +50,21 @@
public OInvoke(OProcess owner, OActivity parent) {
super(owner, parent);
}
+
+ public OPartnerLink getPartnerLink() {
+ return partnerLink;
+ }
+
+ public Operation getOperation() {
+ return operation;
+ }
+
+ public OResource getResource() {
+ return resource;
+ }
+
+ public boolean isRestful() {
+ return resource != null;
+ }
+
}
Modified: ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/OPickReceive.java
URL: http://svn.apache.org/viewvc/ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/OPickReceive.java?rev=773939&r1=773938&r2=773939&view=diff
==============================================================================
--- ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/OPickReceive.java (original)
+++ ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/OPickReceive.java Tue May 12 15:23:30 2009
@@ -52,7 +52,7 @@
}
}
- public static class OnMessage extends OBase {
+ public static class OnMessage extends OBase implements OComm {
static final long serialVersionUID = -1L ;
@@ -62,6 +62,7 @@
/** Correlation set to match on. */
public OScope.CorrelationSet matchCorrelation;
+ public OResource resource;
public OPartnerLink partnerLink;
public Operation operation;
public OScope.Variable variable;
@@ -77,5 +78,21 @@
public String getCorrelatorId() {
return partnerLink.getId() + "." + operation.getName();
}
+
+ public boolean isRestful() {
+ return partnerLink == null && resource != null;
+ }
+
+ public OPartnerLink getPartnerLink() {
+ return partnerLink;
+ }
+
+ public Operation getOperation() {
+ return operation;
+ }
+
+ public OResource getResource() {
+ return resource;
+ }
}
}
Modified: ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/OProcess.java
URL: http://svn.apache.org/viewvc/ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/OProcess.java?rev=773939&r1=773938&r2=773939&view=diff
==============================================================================
--- ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/OProcess.java (original)
+++ ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/OProcess.java Tue May 12 15:23:30 2009
@@ -18,12 +18,7 @@
*/
package org.apache.ode.bpel.rtrep.v2;
-import org.apache.ode.bpel.rapi.ConstantsModel;
-import org.apache.ode.bpel.rapi.ProcessModel;
-import org.apache.ode.bpel.rapi.PartnerLinkModel;
-import org.apache.ode.bpel.rapi.PropertyAliasModel;
-import org.apache.ode.bpel.rapi.ActivityModel;
-import org.apache.ode.bpel.rapi.ScopeModel;
+import org.apache.ode.bpel.rapi.*;
import javax.wsdl.Operation;
import javax.xml.namespace.QName;
@@ -63,9 +58,16 @@
/** Process-level atomic scope flag */
public boolean atomicScope;
+ /** First receive met during the compilation of the process. Serves as a hint when the instantiating
+ * receive isn't explicitly specified (optional in simpel). */
+ public OActivity firstReceive;
+
/** All partner links in the process. */
public final Set<PartnerLinkModel> allPartnerLinks = new HashSet<PartnerLinkModel>();
+ /** All web resources in the process. */
+ public final Set<ResourceModel> providedResources = new HashSet<ResourceModel>();
+
public final List<OProperty> properties = new ArrayList<OProperty>();
/** All declared extensions in the process. **/
@@ -119,7 +121,6 @@
throw new UnsupportedOperationException();
}
-
public Set<PartnerLinkModel> getAllPartnerLinks() {
return Collections.unmodifiableSet(allPartnerLinks);
}
@@ -131,6 +132,10 @@
return null;
}
+ public Set<ResourceModel> getProvidedResources() {
+ return Collections.unmodifiableSet(providedResources);
+ }
+
public PartnerLinkModel getPartnerLink(int partnerLinkModelId) {
return (PartnerLinkModel) getChild(partnerLinkModelId);
}
@@ -240,85 +245,16 @@
public int getModelVersion() {
return 2;
}
-
- public ConstantsModel getConstantsModel() {
- return new ConstantsModel() {
- public QName getConflictingReceive() {
- return constants.qnConflictingReceive;
- }
-
- public QName getCorrelationViolation() {
- return constants.qnCorrelationViolation;
- }
-
- public QName getDuplicateInstance() {
- return constants.qnDuplicateInstance;
- }
-
- public QName getForEachCounterError() {
- return constants.qnForEachCounterError;
- }
-
- public QName getForcedTermination() {
- return constants.qnForcedTermination;
- }
-
- public QName getInvalidBranchCondition() {
- return constants.qnInvalidBranchCondition;
- }
-
- public QName getInvalidExpressionValue() {
- return constants.qnInvalidExpressionValue;
- }
-
- public QName getJoinFailure() {
- return constants.qnJoinFailure;
- }
-
- public QName getMismatchedAssignmentFailure() {
- return constants.qnMismatchedAssignmentFailure;
- }
-
- public QName getMissingReply() {
- return constants.qnMissingReply;
- }
-
- public QName getMissingRequest() {
- return constants.qnMissingRequest;
- }
-
- public QName getRetiredProcess() {
- return constants.qnRetiredProcess;
- }
-
- public QName getSelectionFailure() {
- return constants.qnSelectionFailure;
- }
-
- public QName getSubLanguageExecutionFault() {
- return constants.qnSubLanguageExecutionFault;
- }
-
- public QName getUninitializedPartnerRole() {
- return constants.qnUninitializedPartnerRole;
- }
-
- public QName getUninitializedVariable() {
- return constants.qnUninitializedVariable;
- }
-
- public QName getUnknownFault() {
- return constants.qnUnknownFault;
- }
-
- public QName getXsltInvalidSource() {
- return constants.qnXsltInvalidSource;
- }
- };
+ public ConstantsModel getConstantsModel() {
+ return constants;
}
public ScopeModel getProcessScope() {
return processScope;
}
+
+ public byte[] getGlobalState() {
+ return globalState;
+ }
}
Modified: ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/OReply.java
URL: http://svn.apache.org/viewvc/ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/OReply.java?rev=773939&r1=773938&r2=773939&view=diff
==============================================================================
--- ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/OReply.java (original)
+++ ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/OReply.java Tue May 12 15:23:30 2009
@@ -37,6 +37,8 @@
/** The type of the fault (if isFaultReply). */
public QName fault;
+ public OResource resource;
+
public OPartnerLink partnerLink;
public Operation operation;
public OScope.Variable variable;
Added: ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/OResource.java
URL: http://svn.apache.org/viewvc/ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/OResource.java?rev=773939&view=auto
==============================================================================
--- ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/OResource.java (added)
+++ ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/OResource.java Tue May 12 15:23:30 2009
@@ -0,0 +1,77 @@
+package org.apache.ode.bpel.rtrep.v2;
+
+import org.apache.ode.bpel.rapi.ResourceModel;
+
+/**
+ * RESTful resource representation at the process level
+ */
+public class OResource extends OBase implements ResourceModel {
+
+ private String name;
+ private OExpression subpath;
+ private OResource reference;
+ private String method;
+ private boolean instantiateResource;
+ private boolean inbound;
+ private OScope declaringScope;
+
+ public OResource(OProcess owner) {
+ super(owner);
+ }
+
+ public String getName() {
+ return name;
+ }
+
+ public void setName(String name) {
+ this.name = name;
+ }
+
+ public OExpression getSubpath() {
+ return subpath;
+ }
+
+ public void setSubpath(OExpression subpath) {
+ this.subpath = subpath;
+ }
+
+ public OResource getReference() {
+ return reference;
+ }
+
+ public void setReference(OResource reference) {
+ this.reference = reference;
+ }
+
+ public String getMethod() {
+ return method;
+ }
+
+ public void setMethod(String method) {
+ this.method = method;
+ }
+
+ public boolean isInstantiateResource() {
+ return instantiateResource;
+ }
+
+ public void setInstantiateResource(boolean instantiateResource) {
+ this.instantiateResource = instantiateResource;
+ }
+
+ public boolean isInbound() {
+ return inbound;
+ }
+
+ public void setInbound(boolean inbound) {
+ this.inbound = inbound;
+ }
+
+ public OScope getDeclaringScope() {
+ return declaringScope;
+ }
+
+ public void setDeclaringScope(OScope declaringScope) {
+ this.declaringScope = declaringScope;
+ }
+}
Modified: ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/OScope.java
URL: http://svn.apache.org/viewvc/ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/OScope.java?rev=773939&r1=773938&r2=773939&view=diff
==============================================================================
--- ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/OScope.java (original)
+++ ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/OScope.java Tue May 12 15:23:30 2009
@@ -68,6 +68,9 @@
/** Variables declared within the scope. */
public final Map<String,Variable> variables = new LinkedHashMap<String,Variable>();
+ /** Resources declared within the scope. */
+ public final HashMap<String,OResource> resource = new HashMap<String,OResource>();
+
/** OCorrelation sets declared within the scope. */
public final Map<String,CorrelationSet> correlationSets = new HashMap<String, CorrelationSet>();
@@ -234,7 +237,7 @@
}
public String getDescription() {
- StringBuffer buf = new StringBuffer(declaringScope.name);
+ StringBuffer buf = new StringBuffer(declaringScope.name != null ? declaringScope.name : "");
buf.append('.');
buf.append(name);
return buf.toString();
Modified: ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/OVarType.java
URL: http://svn.apache.org/viewvc/ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/OVarType.java?rev=773939&r1=773938&r2=773939&view=diff
==============================================================================
--- ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/OVarType.java (original)
+++ ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/OVarType.java Tue May 12 15:23:30 2009
@@ -31,6 +31,7 @@
public static int SCHEMA_TYPE = 0;
public static int NUMBER_TYPE = 1;
public static int STRING_TYPE = 2;
+ public static int BOOLEAN_TYPE = 3;
public int underlyingType = SCHEMA_TYPE;
Modified: ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/OdeInternalInstance.java
URL: http://svn.apache.org/viewvc/ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/OdeInternalInstance.java?rev=773939&r1=773938&r2=773939&view=diff
==============================================================================
--- ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/OdeInternalInstance.java (original)
+++ ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/OdeInternalInstance.java Tue May 12 15:23:30 2009
@@ -17,6 +17,7 @@
import javax.xml.namespace.QName;
import java.util.Collection;
import java.util.Date;
+import java.util.Map;
import java.net.URI;
@@ -37,9 +38,18 @@
void initializePartnerLinks(Long parentScopeId, Collection<OPartnerLink> partnerLinks);
+ void initializeResource(Long parentScopeId, OResource resource, String url);
+
+ void initializeInstantiatingUrl(String url);
+
+ String getInstantiatingUrl();
+
String invoke(String invokeId, PartnerLinkInstance instance, Operation operation, Element outboundMsg, Object object)
throws FaultException;
+ String invoke(String requestId, org.apache.ode.bpel.iapi.Resource resource, Element outgoingMessage)
+ throws FaultException;
+
Node getPartData(Element message, OMessageVarType.Part part);
Element getPartnerResponse(String mexId);
@@ -64,6 +74,10 @@
Element getMyRequest(String mexId);
+ Map<String,String> getProperties(String mexId);
+
+ void setInstantiatingMex(String mexId);
+
void registerTimer(TimerResponseChannel timerChannel, Date future);
void registerActivityForRecovery(ActivityRecoveryChannel recoveryChannel, long id, String reason, Date dateTime,
@@ -76,6 +90,9 @@
void select(PickResponseChannel pickResponseChannel, Date timeout, boolean createInstance, Selector[] selectors)
throws FaultException;
+ void checkResourceRoute(ResourceInstance resourceInstance, String mexRef,
+ PickResponseChannel pickResponseChannel, int selectorIdx);
+
CorrelationKey readCorrelation(CorrelationSetInstance cset);
ExpressionLanguageRuntimeRegistry getExpLangRuntime();
@@ -99,7 +116,7 @@
void sendEvent(ProcessInstanceStartedEvent evt);
- Long getPid();
+ Long getInstanceId();
URI getBaseResourceURI();
@@ -119,9 +136,6 @@
void forceRollback();
- void reply(PartnerLinkInstance plink, String opName, String bpelmex, Element element, QName fault)
- throws FaultException;
-
int getRetryDelay();
boolean isFirstTry();
@@ -140,4 +154,13 @@
* @return propertyValue - the value corresponding to the process property name.
*/
Node getProcessProperty(QName propertyName);
+ void associateEvent(PartnerLinkInstance plinkInstance, String opName, String mexRef, String scopeIid);
+
+ void associateEvent(ResourceInstance resourceInstance, String mexRef, String scopeIid);
+
+ void reply(PartnerLinkInstance plink, String opName, String bpelmex, Element element, QName fault)
+ throws FaultException;
+
+ void reply(ResourceInstance resource, String bpelmex, Element element, QName fault) throws FaultException;
+
}
Modified: ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/OutstandingRequestManager.java
URL: http://svn.apache.org/viewvc/ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/OutstandingRequestManager.java?rev=773939&r1=773938&r2=773939&view=diff
==============================================================================
--- ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/OutstandingRequestManager.java (original)
+++ ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/OutstandingRequestManager.java Tue May 12 15:23:30 2009
@@ -49,6 +49,9 @@
private final Map<RequestIdTuple, Entry> _byRid = new HashMap<RequestIdTuple, Entry>();
private final Map<String, Entry> _byChannel = new HashMap<String, Entry>();
+ private final Map<RequestResTuple, RestEntry> _byRestRid = new HashMap<RequestResTuple, RestEntry>();
+ private final Map<String, RestEntry> _byRestChannel = new HashMap<String, RestEntry>();
+
int findConflict(Selector selectors[]) {
if (__log.isTraceEnabled()) {
__log.trace(ObjectPrinter.stringifyMethodEnter("findConflict", new Object[] { "selectors", selectors}) );
@@ -75,12 +78,9 @@
* @param selectors selectors for this receive/pick
*/
void register(String pickResponseChannel, Selector selectors[]) {
- if (__log.isTraceEnabled()) {
+ if (__log.isTraceEnabled())
__log.trace(ObjectPrinter.stringifyMethodEnter("register", new Object[] {
- "pickResponseChannel", pickResponseChannel,
- "selectors", selectors
- }) );
- }
+ "pickResponseChannel", pickResponseChannel, "selectors", selectors }) );
if (_byChannel.containsKey(pickResponseChannel)) {
String errmsg = "INTERNAL ERROR: Duplicate ENTRY for RESPONSE CHANNEL " + pickResponseChannel;
@@ -106,6 +106,28 @@
_byChannel.put(pickResponseChannel, entry);
}
+ void register(String pickResponseChannel, ResourceInstance resource, String method, String mexRef) {
+ if (__log.isTraceEnabled())
+ __log.trace(ObjectPrinter.stringifyMethodEnter("register", new Object[] {
+ "pickResponseChannel", pickResponseChannel}) );
+
+ if (_byRestChannel.containsKey(pickResponseChannel)) {
+ String errmsg = "INTERNAL ERROR: Duplicate ENTRY for RESPONSE CHANNEL " + pickResponseChannel;
+ __log.fatal(errmsg);
+ throw new IllegalArgumentException(errmsg);
+ }
+
+ final RequestResTuple rid = new RequestResTuple(resource, method, mexRef);
+ if (_byRestRid.containsKey(rid)) {
+ String errmsg = "INTERNAL ERROR: Duplicate ENTRY for RID " + rid;
+ __log.fatal(errmsg);
+ throw new IllegalStateException(errmsg);
+ }
+ RestEntry entry = new RestEntry(pickResponseChannel);
+ _byRestRid.put(rid, entry);
+ _byRestChannel.put(pickResponseChannel, entry);
+ }
+
/**
* Cancel a previous registration.
* @see #register(String, Selector[])
@@ -114,13 +136,15 @@
void cancel(String pickResponseChannel) {
if (__log.isTraceEnabled())
__log.trace(ObjectPrinter.stringifyMethodEnter("cancel", new Object[] {
- "pickResponseChannel", pickResponseChannel
- }) );
+ "pickResponseChannel", pickResponseChannel }) );
Entry entry = _byChannel.remove(pickResponseChannel);
- if (entry != null) {
+ if (entry != null)
while(_byRid.values().remove(entry));
- }
+
+ RestEntry restEntry = _byRestChannel.remove(pickResponseChannel);
+ if (restEntry != null)
+ while(_byRestRid.values().remove(restEntry));
}
/**
@@ -138,18 +162,41 @@
Entry entry = _byChannel.get(pickResponseChannel);
if (entry == null) {
- String errmsg = "INTERNAL ERROR: No ENTRY for RESPONSE CHANNEL " + pickResponseChannel;
- __log.fatal(errmsg);
- throw new IllegalArgumentException(errmsg);
+ RestEntry restEntry = _byRestChannel.get(pickResponseChannel);
+ if (restEntry == null) {
+ String errmsg = "INTERNAL ERROR: No ENTRY for RESPONSE CHANNEL " + pickResponseChannel;
+ __log.fatal(errmsg);
+ throw new IllegalArgumentException(errmsg);
+ } else {
+ if (restEntry.mexRef != null) {
+ String errmsg = "INTERNAL ERROR: Duplicate ASSOCIATION for CHANEL " + pickResponseChannel;
+ __log.fatal(errmsg);
+ throw new IllegalStateException(errmsg);
+ }
+ restEntry.mexRef = mexRef;
+ }
+ } else {
+ if (entry.mexRef != null) {
+ String errmsg = "INTERNAL ERROR: Duplicate ASSOCIATION for CHANEL " + pickResponseChannel;
+ __log.fatal(errmsg);
+ throw new IllegalStateException(errmsg);
+ }
+ entry.mexRef = mexRef;
}
+ }
- if (entry.mexRef != null) {
- String errmsg = "INTERNAL ERROR: Duplicate ASSOCIATION for CHANEL " + pickResponseChannel;
- __log.fatal(errmsg);
- throw new IllegalStateException(errmsg);
- }
+ public void associateEvent(PartnerLinkInstance plinkInstance, String opName, String mexRef, String scopeIid) {
+ RequestIdTuple rid = new RequestIdTuple(plinkInstance, opName, mexRef);
+ Entry entry = _byRid.remove(rid);
+ rid.mexId = scopeIid;
+ _byRid.put(rid, entry);
+ }
- entry.mexRef = mexRef;
+ public void associateEvent(ResourceInstance resourceInstance, String method, String mexRef, String scopeIid) {
+ RequestResTuple rid = new RequestResTuple(resourceInstance, method, mexRef);
+ RestEntry entry = _byRestRid.remove(rid);
+ rid.mexId = scopeIid;
+ _byRestRid.put(rid, entry);
}
/**
@@ -181,6 +228,24 @@
return entry.mexRef;
}
+ public String release(ResourceInstance resourceInstance, String method, String mexId) {
+ if (__log.isTraceEnabled())
+ __log.trace(ObjectPrinter.stringifyMethodEnter("release", new Object[] {
+ "resource", resourceInstance, "method", method, "mexId", mexId }) );
+
+ final RequestResTuple rid = new RequestResTuple(resourceInstance, method, mexId);
+ RestEntry entry = _byRestRid.get(rid);
+ if (entry == null) {
+ if (__log.isDebugEnabled()) {
+ __log.debug("==release: RID " + rid + " not found in " + _byRestRid);
+ }
+ return null;
+ }
+ while(_byRestChannel.values().remove(entry));
+ while(_byRestRid.values().remove(entry));
+ return entry.mexRef;
+ }
+
/**
* "Release" all outstanding incoming messages exchanges. Makes the object forget about
* the previous registrations
@@ -248,6 +313,36 @@
}
}
+ private class RequestResTuple implements Serializable {
+ private static final long serialVersionUID = -1059359612839777482L;
+ /** Name of the operation. */
+ ResourceInstance resource;
+ /** Message exchange identifier. */
+ String method;
+ /** Message exchange identifier. */
+ String mexId;
+
+ /** Constructor. */
+ private RequestResTuple(ResourceInstance resource, String method, String mexId) {
+ this.resource = resource;
+ this.method = method;
+ this.mexId = mexId;
+ }
+
+ public int hashCode() {
+ return this.resource.hashCode() ^ this.method.hashCode() ^ this.mexId.hashCode();
+ }
+
+ public boolean equals(Object obj) {
+ RequestResTuple other = (RequestResTuple) obj;
+ return other.resource.equals(resource) && other.method.equals(method) && other.mexId.equals(mexId);
+ }
+
+ public String toString() {
+ return ObjectPrinter.toString(this, new Object[] {"url", resource, "method", method, "mexId", mexId});
+ }
+ }
+
private class Entry implements Serializable {
private static final long serialVersionUID = -583743124656582887L;
final String pickResponseChannel;
@@ -267,4 +362,21 @@
});
}
}
+
+ private class RestEntry implements Serializable {
+ private static final long serialVersionUID = -583733124656582887L;
+ final String pickResponseChannel;
+ String mexRef;
+
+ private RestEntry(String pickResponseChannel) {
+ this.pickResponseChannel = pickResponseChannel;
+ }
+
+ public String toString() {
+ return ObjectPrinter.toString(this, new Object[] {
+ "pickResponseChannel", pickResponseChannel,
+ "mexRef", mexRef
+ });
+ }
+ }
}
Modified: ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/PICK.java
URL: http://svn.apache.org/viewvc/ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/PICK.java?rev=773939&r1=773938&r2=773939&view=diff
==============================================================================
--- ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/PICK.java (original)
+++ ode/trunk/runtimes/src/main/java/org/apache/ode/bpel/rtrep/v2/PICK.java Tue May 12 15:23:30 2009
@@ -21,6 +21,7 @@
import java.util.Calendar;
import java.util.Collection;
import java.util.Date;
+import java.util.Map;
import javax.xml.namespace.QName;
@@ -35,11 +36,13 @@
import org.apache.ode.bpel.rtrep.v2.channels.TerminationChannelListener;
import org.apache.ode.bpel.rapi.InvalidProcessException;
import org.apache.ode.utils.DOMUtils;
+import org.apache.ode.utils.GUID;
import org.apache.ode.utils.xsd.Duration;
import org.apache.ode.bpel.evar.ExternalVariableModuleException;
import org.apache.ode.bpel.iapi.BpelEngineException;
import org.w3c.dom.Element;
import org.w3c.dom.Node;
+import org.w3c.dom.Document;
/**
* Template for the BPEL <code>pick</code> activity.
@@ -69,39 +72,7 @@
Selector[] selectors;
try {
- selectors = new Selector[_opick.onMessages.size()];
- int idx = 0;
- for (OPickReceive.OnMessage onMessage : _opick.onMessages) {
- CorrelationKey key = null; // this will be the case for the
- // createInstance activity
-
- PartnerLinkInstance pLinkInstance = _scopeFrame.resolve(onMessage.partnerLink);
- if (onMessage.matchCorrelation == null && !_opick.createInstanceFlag) {
- // Adding a route for opaque correlation. In this case,
- // correlation is on "out-of-band" session-id
- String sessionId = getBpelRuntime().fetchMySessionId(pLinkInstance);
- key = new CorrelationKey(-1, new String[] { sessionId });
- } else if (onMessage.matchCorrelation != null) {
- if (!getBpelRuntime().isCorrelationInitialized(
- _scopeFrame.resolve(onMessage.matchCorrelation))) {
- // the following should really test if this is a "join"
- // type correlation...
- if (!_opick.createInstanceFlag)
- throw new FaultException(_opick.getOwner().constants.qnCorrelationViolation,
- "Correlation not initialized.");
- } else {
-
- key = getBpelRuntime().readCorrelation(_scopeFrame.resolve(onMessage.matchCorrelation));
-
- assert key != null;
- }
- }
-
- selectors[idx] = new Selector(idx, pLinkInstance, onMessage.operation.getName(), onMessage.operation
- .getOutput() == null, onMessage.messageExchangeId, key);
- idx++;
- }
-
+ // Pick onAlarm
timeout = null;
for (OPickReceive.OnAlarm onAlarm : _opick.onAlarms) {
Date dt = onAlarm.forExpr != null ? offsetFromNow(getBpelRuntime().getExpLangRuntime()
@@ -112,7 +83,43 @@
_alarm = onAlarm;
}
}
- getBpelRuntime().select(pickResponseChannel, timeout, _opick.createInstanceFlag, selectors);
+
+ // Pick onMessages (identical to a receive)
+ selectors = new Selector[_opick.onMessages.size()];
+ int idx = 0;
+ for (OPickReceive.OnMessage onMessage : _opick.onMessages) {
+ if (onMessage.isRestful()) {
+ getBpelRuntime().checkResourceRoute(_scopeFrame.resolve(onMessage.resource),
+ onMessage.messageExchangeId, pickResponseChannel, idx);
+ } else {
+ CorrelationKey key = null;
+ PartnerLinkInstance pLinkInstance = _scopeFrame.resolve(onMessage.partnerLink);
+ if (onMessage.matchCorrelation == null && !_opick.createInstanceFlag) {
+ // Adding a route for opaque correlation. In this case,
+ // correlation is on "out-of-band" session-id
+ String sessionId = getBpelRuntime().fetchMySessionId(pLinkInstance);
+ key = new CorrelationKey(-1, new String[] { sessionId });
+ } else if (onMessage.matchCorrelation != null) {
+ if (!getBpelRuntime().isCorrelationInitialized(
+ _scopeFrame.resolve(onMessage.matchCorrelation))) {
+ // the following should really test if this is a "join" type correlation...
+ if (!_opick.createInstanceFlag)
+ throw new FaultException(_opick.getOwner().constants.qnCorrelationViolation,
+ "Correlation not initialized.");
+ } else {
+ key = getBpelRuntime().readCorrelation(_scopeFrame.resolve(onMessage.matchCorrelation));
+ assert key != null;
+ }
+ }
+
+ selectors[idx] = new Selector(idx, pLinkInstance, onMessage.operation.getName(), onMessage.operation
+ .getOutput() == null, onMessage.messageExchangeId, key);
+ idx++;
+ }
+ }
+
+ if (selectors[0] != null)
+ getBpelRuntime().select(pickResponseChannel, timeout, _opick.createInstanceFlag, selectors);
} catch (FaultException e) {
__log.error(e);
FaultData fault = createFault(e.getQName(), _opick, e.getMessage());
@@ -161,56 +168,77 @@
return;
}
- Collection<String> partNames = (Collection<String>) onMessage.operation.getInput().getMessage().getParts().keySet();
-
- // Let's do some sanity checks here so that we don't get weird errors in assignment later.
- // The engine should have checked to make sure that the messages that are delivered conform
- // to the correct format; but you know what they say, don't trust anyone.
- if (!(onMessage.variable.type instanceof OMessageVarType)) {
- String errmsg = "Non-message variable for receive: should have been picked up by static analysis.";
- __log.fatal(errmsg);
- throw new InvalidProcessException(errmsg);
- }
+ if (!onMessage.isRestful()) {
+ Collection<String> partNames = (Collection<String>) onMessage.operation.getInput().getMessage().getParts().keySet();
- OMessageVarType vartype = (OMessageVarType) onMessage.variable.type;
-
- // Check that each part contains what we expect.
- for (String pName : partNames) {
- QName partName = new QName(null, pName);
- Element msgPart = DOMUtils.findChildByName(msgEl, partName);
- OMessageVarType.Part part = vartype.parts.get(pName);
- if (part == null) {
- String errmsg = "Inconsistent WSDL, part " + pName + " not found in message type " + vartype.messageType;
+ // Let's do some sanity checks here so that we don't get weird errors in assignment later.
+ // The engine should have checked to make sure that the messages that are delivered conform
+ // to the correct format; but you know what they say, don't trust anyone.
+ if (!(onMessage.variable.type instanceof OMessageVarType)) {
+ String errmsg = "Non-message variable for receive: should have been picked up by static analysis.";
__log.fatal(errmsg);
throw new InvalidProcessException(errmsg);
}
- if (msgPart == null) {
- String errmsg = "Message missing part: " + pName;
- __log.fatal(errmsg);
- throw new InvalidContextException(errmsg);
- }
-
- if (part.type instanceof OElementVarType) {
- OElementVarType ptype = (OElementVarType) part.type;
- Element e = DOMUtils.getFirstChildElement(msgPart);
- if (e == null) {
- String errmsg = "Message (element) part " + pName + " did not contain child element.";
+
+ OMessageVarType vartype = (OMessageVarType) onMessage.variable.type;
+ // Check that each part contains what we expect.
+ for (String pName : partNames) {
+ QName partName = new QName(null, pName);
+ Element msgPart = DOMUtils.findChildByName(msgEl, partName);
+ OMessageVarType.Part part = vartype.parts.get(pName);
+ if (part == null) {
+ String errmsg = "Inconsistent WSDL, part " + pName + " not found in message type " + vartype.messageType;
+ __log.fatal(errmsg);
+ throw new InvalidProcessException(errmsg);
+ }
+ if (msgPart == null) {
+ String errmsg = "Message missing part: " + pName;
__log.fatal(errmsg);
throw new InvalidContextException(errmsg);
}
- // Relaxing that check a bit for SimPEL
- if (!ptype.elementType.getLocalPart().equals("simpelWrapper")) {
- QName qn = new QName(e.getNamespaceURI(), e.getLocalName());
- if(!qn.equals(ptype.elementType)) {
- String errmsg = "Message (element) part " + pName + " did not contain correct child element: expected "
- + ptype.elementType + " but got " + qn;
+ if (part.type instanceof OElementVarType) {
+ OElementVarType ptype = (OElementVarType) part.type;
+ Element e = DOMUtils.getFirstChildElement(msgPart);
+ if (e == null) {
+ String errmsg = "Message (element) part " + pName + " did not contain child element.";
__log.fatal(errmsg);
throw new InvalidContextException(errmsg);
}
+
+ // Relaxing that check a bit for SimPEL
+ if (!ptype.elementType.getLocalPart().equals("simpelWrapper")) {
+ QName qn = new QName(e.getNamespaceURI(), e.getLocalName());
+ if(!qn.equals(ptype.elementType)) {
+ String errmsg = "Message (element) part " + pName + " did not contain correct child element: expected "
+ + ptype.elementType + " but got " + qn;
+ __log.fatal(errmsg);
+ throw new InvalidContextException(errmsg);
+ }
+ }
+ }
+
+ }
+ } else {
+ // Retrieving the map of properties associated with a RESTful mex. Those are the query
+ // parameters and are used to set implicit variables associated with these parameters
+ // in the process.
+ Map<String,String> props = getBpelRuntime().getProperties(mexId);
+ for (Map.Entry<String, String> entry : props.entrySet()) {
+ VariableInstance vi = _scopeFrame.resolveVariable(entry.getKey());
+ if (vi != null) {
+ // Always expected to be a string
+ Document doc = DOMUtils.newDocument();
+ Node textNode = doc.createTextNode(entry.getValue());
+ try {
+ initializeVariable(vi, textNode);
+ } catch (ExternalVariableModuleException e) {
+ __log.error("Exception while initializing external variable", e);
+ _self.parent.failure(e.toString(), null);
+ return;
+ }
}
}
-
}
VariableInstance vinst = _scopeFrame.resolve(onMessage.variable);
@@ -226,8 +254,7 @@
// Generating event
VariableModificationEvent se = new VariableModificationEvent(vinst.declaration.name);
se.setNewValue(msgEl);
- if (_opick.debugInfo != null)
- se.setLineNo(_opick.debugInfo.startLine);
+ if (_opick.debugInfo != null) se.setLineNo(_opick.debugInfo.startLine);
sendEvent(se);
}
@@ -269,13 +296,14 @@
FaultData fault;
initVariable(mexId, onMessage);
try {
+ if (onMessage.isRestful() && onMessage.getResource().isInstantiateResource())
+ getBpelRuntime().setInstantiatingMex(mexId);
+
for (OScope.CorrelationSet cset : onMessage.initCorrelations) {
initializeCorrelation(_scopeFrame.resolve(cset), _scopeFrame.resolve(onMessage.variable));
}
- if (onMessage.partnerLink.hasPartnerRole()) {
- // Trying to initialize partner epr based on a
- // message-provided epr/session.
-
+ if (onMessage.partnerLink != null && onMessage.partnerLink.hasPartnerRole()) {
+ // Trying to initialize partner epr based on a message-provided epr/session.
if (!getBpelRuntime().isPartnerRoleEndpointInitialized(
_scopeFrame.resolve(onMessage.partnerLink))
|| !onMessage.partnerLink.initializePartnerRole) {