pick.java

来自「bpel执行引擎用来执行bpel业务流程」· Java 代码 · 共 344 行 · 第 1/2 页

        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);
        }

        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);
            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);
            }

            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);
                }

                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);
                }
            }

        }

        VariableInstance vinst = _scopeFrame.resolve(onMessage.variable);

        try {
            initializeVariable(vinst, msgEl);
        } catch (ExternalVariableModuleException e) {
            __log.error("Exception while initializing external variable", e);
            _self.parent.failure(e.toString(), null);
            return;
        }

        // Generating event
        VariableModificationEvent se = new VariableModificationEvent(vinst.declaration.name);
        se.setNewValue(msgEl);
        if (_opick.debugInfo != null)
            se.setLineNo(_opick.debugInfo.startLine);
        sendEvent(se);
    }

    private class WAITING extends BpelJacobRunnable {
        private static final long serialVersionUID = 1L;

        private PickResponseChannel _pickResponseChannel;

        private WAITING(PickResponseChannel pickResponseChannel) {
            this._pickResponseChannel = pickResponseChannel;
        }

        public void run() {

            object(false, new PickResponseChannelListener(_pickResponseChannel) {
                private static final long serialVersionUID = -8237296827418738011L;

                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);
                    }

                    FaultData fault;
                    initVariable(mexId, onMessage);
                    try {
                        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 (!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);

                        }
                    } 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 TerminationChannelListener(_self.self) {
                private static final long serialVersionUID = 4399496341785922396L;

                public void terminate() {
                    getBpelRuntimeContext().cancel(_pickResponseChannel);
                    instance(WAITING.this);
                }
            }));
        }
    }
}