private WAITING(PickResponse pickResponseChannel) {
this._pickResponseChannel = pickResponseChannel;
}
public void run() {
object(false, compose(new ReceiveProcess() {
private static final long serialVersionUID = -8237296827418738011L;
}.setChannel(_pickResponseChannel).setReceiver(new PickResponse() {
public void onRequestRcvd(int selectorIdx, String mexId) {
OPickReceive.OnMessage onMessage = _opick.onMessages.get(selectorIdx);
// dead path the non-selected onMessage blocks.
for (OPickReceive.OnMessage onmsg : _opick.onMessages) {
if (!onmsg.equals(onMessage)) {
dpe(onmsg.activity);
}
}
// dead-path the alarm (if any)
if (_alarm != null) {
dpe(_alarm.activity);
}
getBpelRuntimeContext().cancelOutstandingRequests(ProcessUtil.exportChannel(_pickResponseChannel));
FaultData fault;
initVariable(mexId, onMessage);
try {
VariableInstance vinst = _scopeFrame.resolve(onMessage.variable);
for (OScope.CorrelationSet cset : onMessage.initCorrelations) {
initializeCorrelation(_scopeFrame.resolve(cset), vinst);
}
for( OScope.CorrelationSet cset : onMessage.joinCorrelations ) {
// will be ignored if already initialized
initializeCorrelation(_scopeFrame.resolve(cset), vinst);
}
if (onMessage.partnerLink.hasPartnerRole()) {
// Trying to initialize partner epr based on a
// message-provided epr/session.
if (!getBpelRuntimeContext().isPartnerRoleEndpointInitialized(
_scopeFrame.resolve(onMessage.partnerLink))
|| !onMessage.partnerLink.initializePartnerRole) {
Node fromEpr = getBpelRuntimeContext().getSourceEPR(mexId);
if (fromEpr != null) {
if (__log.isDebugEnabled())
__log.debug("Received callback EPR " + DOMUtils.domToString(fromEpr)
+ " saving it on partner link " + onMessage.partnerLink.getName());
getBpelRuntimeContext().writeEndpointReference(
_scopeFrame.resolve(onMessage.partnerLink), (Element) fromEpr);
}
}
String partnersSessionId = getBpelRuntimeContext().getSourceSessionId(mexId);
if (partnersSessionId != null)
getBpelRuntimeContext().initializePartnersSessionId(
_scopeFrame.resolve(onMessage.partnerLink), partnersSessionId);
}
// this request is now waiting for a reply
getBpelRuntimeContext().processOutstandingRequest(_scopeFrame.resolve(onMessage.partnerLink),
onMessage.operation.getName(), onMessage.messageExchangeId, mexId);
} catch (FaultException e) {
__log.error(e);
fault = createFault(e.getQName(), onMessage);
_self.parent.completed(fault, CompensationHandler.emptySet());
dpe(onMessage.activity);
return;
}
// load 'onMessage' activity
// Because we are done with all the DPE, we can simply
// re-use our control
// channels for the child.
ActivityInfo child = new ActivityInfo(genMonotonic(), onMessage.activity, _self.self, _self.parent);
instance(createChild(child, _scopeFrame, _linkFrame));
}
public void onTimeout() {
// Dead path all the onMessage activiites (the other alarms
// have already been DPE'ed)
for (OPickReceive.OnMessage onMessage : _opick.onMessages) {
dpe(onMessage.activity);
}
// Because we are done with all the DPE, we can simply
// re-use our control
// channels for the child.
ActivityInfo child = new ActivityInfo(genMonotonic(), _alarm.activity, _self.self, _self.parent);
instance(createChild(child, _scopeFrame, _linkFrame));
}
public void onCancel() {
_self.parent.completed(null, CompensationHandler.emptySet());
}
})).or(new ReceiveProcess() {
private static final long serialVersionUID = 4399496341785922396L;
}.setChannel(_self.self).setReceiver(new Termination() {
public void terminate() {
getBpelRuntimeContext().cancel(_pickResponseChannel);
instance(WAITING.this);