| /* |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, |
| * software distributed under the License is distributed on an |
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| * KIND, either express or implied. See the License for the |
| * specific language governing permissions and limitations |
| * under the License. |
| */ |
| package org.apache.ode.bpel.engine; |
| |
| import java.io.ByteArrayInputStream; |
| import java.io.ByteArrayOutputStream; |
| import java.net.URI; |
| import java.util.ArrayList; |
| import java.util.Calendar; |
| import java.util.Collection; |
| import java.util.Collections; |
| import java.util.Date; |
| import java.util.List; |
| |
| import javax.wsdl.Operation; |
| import javax.xml.namespace.QName; |
| |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| import org.apache.ode.bpel.common.CorrelationKey; |
| import org.apache.ode.bpel.common.CorrelationKeySet; |
| import org.apache.ode.bpel.common.FaultException; |
| import org.apache.ode.bpel.common.ProcessState; |
| import org.apache.ode.bpel.dao.CorrelationSetDAO; |
| import org.apache.ode.bpel.dao.CorrelatorDAO; |
| import org.apache.ode.bpel.dao.MessageDAO; |
| import org.apache.ode.bpel.dao.MessageExchangeDAO; |
| import org.apache.ode.bpel.dao.MessageRouteDAO; |
| import org.apache.ode.bpel.dao.PartnerLinkDAO; |
| import org.apache.ode.bpel.dao.ProcessDAO; |
| import org.apache.ode.bpel.dao.ProcessInstanceDAO; |
| import org.apache.ode.bpel.dao.ScopeDAO; |
| import org.apache.ode.bpel.dao.ScopeStateEnum; |
| import org.apache.ode.bpel.dao.XmlDataDAO; |
| import org.apache.ode.bpel.evar.ExternalVariableModule.Value; |
| import org.apache.ode.bpel.evar.ExternalVariableModuleException; |
| import org.apache.ode.bpel.evt.CorrelationSetWriteEvent; |
| import org.apache.ode.bpel.evt.ProcessCompletionEvent; |
| import org.apache.ode.bpel.evt.ProcessInstanceEvent; |
| import org.apache.ode.bpel.evt.ProcessInstanceStateChangeEvent; |
| import org.apache.ode.bpel.evt.ProcessMessageExchangeEvent; |
| import org.apache.ode.bpel.evt.ProcessTerminationEvent; |
| import org.apache.ode.bpel.evt.ScopeCompletionEvent; |
| import org.apache.ode.bpel.evt.ScopeEvent; |
| import org.apache.ode.bpel.evt.ScopeFaultEvent; |
| import org.apache.ode.bpel.evt.ScopeStartEvent; |
| 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; |
| import org.apache.ode.bpel.iapi.Message; |
| import org.apache.ode.bpel.iapi.MessageExchange; |
| import org.apache.ode.bpel.iapi.MessageExchange.FailureType; |
| import org.apache.ode.bpel.iapi.MessageExchange.MessageExchangePattern; |
| import org.apache.ode.bpel.iapi.MyRoleMessageExchange; |
| import org.apache.ode.bpel.iapi.ProcessConf.CLEANUP_CATEGORY; |
| import org.apache.ode.bpel.iapi.ProcessConf.PartnerRoleConfig; |
| import org.apache.ode.bpel.iapi.Scheduler; |
| import org.apache.ode.bpel.iapi.Scheduler.JobDetails; |
| import org.apache.ode.bpel.iapi.Scheduler.JobType; |
| import org.apache.ode.bpel.memdao.ProcessInstanceDaoImpl; |
| import org.apache.ode.bpel.obj.OMessageVarType; |
| import org.apache.ode.bpel.obj.OPartnerLink; |
| import org.apache.ode.bpel.obj.OProcess; |
| import org.apache.ode.bpel.obj.OScope; |
| import org.apache.ode.bpel.obj.OScope.Variable; |
| import org.apache.ode.bpel.runtime.BpelJacobRunnable; |
| import org.apache.ode.bpel.runtime.BpelRuntimeContext; |
| import org.apache.ode.bpel.runtime.CorrelationSetInstance; |
| import org.apache.ode.bpel.runtime.ExpressionLanguageRuntimeRegistry; |
| import org.apache.ode.bpel.runtime.PROCESS; |
| import org.apache.ode.bpel.runtime.PartnerLinkInstance; |
| import org.apache.ode.bpel.runtime.Selector; |
| import org.apache.ode.bpel.runtime.VariableInstance; |
| import org.apache.ode.bpel.runtime.channels.ActivityRecovery; |
| import org.apache.ode.bpel.runtime.channels.FaultData; |
| import org.apache.ode.bpel.runtime.channels.InvokeResponse; |
| import org.apache.ode.bpel.runtime.channels.PickResponse; |
| import org.apache.ode.bpel.runtime.channels.TimerResponse; |
| import org.apache.ode.jacob.JacobRunnable; |
| import org.apache.ode.jacob.ProcessUtil; |
| import org.apache.ode.jacob.vpu.ExecutionQueueImpl; |
| import org.apache.ode.jacob.vpu.JacobVPU; |
| import org.apache.ode.utils.DOMUtils; |
| import org.apache.ode.utils.GUID; |
| import org.apache.ode.utils.Namespaces; |
| import org.apache.ode.utils.ObjectPrinter; |
| import org.w3c.dom.Document; |
| import org.w3c.dom.Element; |
| import org.w3c.dom.Node; |
| import org.w3c.dom.NodeList; |
| |
| public class BpelRuntimeContextImpl implements BpelRuntimeContext { |
| |
| private static final Logger __log = LoggerFactory.getLogger(BpelRuntimeContextImpl.class); |
| |
| /** Data-access object for process instance. */ |
| protected ProcessInstanceDAO _dao; |
| |
| /** Process Instance ID */ |
| private final Long _iid; |
| |
| /** JACOB VPU */ |
| protected JacobVPU _vpu; |
| |
| /** JACOB ExecutionQueue (state) */ |
| protected ExecutionQueueImpl _soup; |
| |
| private MyRoleMessageExchangeImpl _instantiatingMessageExchange; |
| |
| protected OutstandingRequestManager _outstandingRequests; |
| |
| protected IMAManager2 _imaManager; |
| |
| protected BpelProcess _bpelProcess; |
| |
| private Date _currentEventDateTime; |
| |
| private boolean _forceFlush; |
| |
| /** Five second maximum for continous execution. */ |
| private long _maxReductionTimeMs = 2000000; |
| |
| public BpelRuntimeContextImpl(BpelProcess bpelProcess, ProcessInstanceDAO dao, PROCESS PROCESS, |
| MyRoleMessageExchangeImpl instantiatingMessageExchange) { |
| _bpelProcess = bpelProcess; |
| _dao = dao; |
| _iid = dao.getInstanceId(); |
| _instantiatingMessageExchange = instantiatingMessageExchange; |
| _vpu = new JacobVPU(); |
| _vpu.registerExtension(BpelRuntimeContext.class, this); |
| |
| _soup = new ExecutionQueueImpl(null); |
| _soup.setReplacementMap(_bpelProcess.getReplacementMap(dao.getProcess().getProcessId())); |
| _outstandingRequests = null; |
| _imaManager = new IMAManager2(); |
| _vpu.setContext(_soup); |
| |
| if (bpelProcess.isInMemory()) { |
| ProcessInstanceDaoImpl inmem = (ProcessInstanceDaoImpl) _dao; |
| if (inmem.getSoup() != null) { |
| _soup = (ExecutionQueueImpl) inmem.getSoup(); |
| _imaManager = (IMAManager2) _soup.getGlobalData(); |
| _vpu.setContext(_soup); |
| } |
| } else { |
| byte[] daoState = dao.getExecutionState(); |
| if (daoState != null) { |
| ByteArrayInputStream iis = new ByteArrayInputStream(daoState); |
| try { |
| _soup.read(iis); |
| } catch (Exception ex) { |
| throw new RuntimeException(ex); |
| } |
| _imaManager = (IMAManager2) _soup.getGlobalData(); |
| } |
| } |
| |
| if (PROCESS != null) { |
| _vpu.inject(PROCESS); |
| } |
| |
| if (BpelProcess.__log.isDebugEnabled()) { |
| __log.debug("BpelRuntimeContextImpl created for instance " + _iid + ". INDEXED STATE=" + _soup.getIndex()); |
| } |
| } |
| |
| public Long getPid() { |
| return _iid; |
| } |
| |
| public long genId() { |
| return _dao.genMonotonic(); |
| } |
| |
| /** |
| * @see BpelRuntimeContext#isCorrelationInitialized(org.apache.ode.bpel.runtime.CorrelationSetInstance) |
| */ |
| public boolean isCorrelationInitialized(CorrelationSetInstance correlationSet) { |
| ScopeDAO scopeDAO = _dao.getScope(correlationSet.scopeInstance); |
| CorrelationSetDAO cs = scopeDAO.getCorrelationSet(correlationSet.declaration.getName()); |
| |
| return cs.getValue() != null; |
| } |
| |
| /** |
| * @see BpelRuntimeContext#isVariableInitialized(org.apache.ode.bpel.runtime.VariableInstance) |
| */ |
| public boolean isVariableInitialized(VariableInstance var) { |
| if (var == null) { |
| return false; |
| } |
| ScopeDAO scopeDAO = _dao.getScope(var.scopeInstance); |
| XmlDataDAO dataDAO = scopeDAO.getVariable(var.declaration.getName()); |
| return !dataDAO.isNull(); |
| } |
| |
| public boolean isPartnerRoleEndpointInitialized(PartnerLinkInstance pLink) { |
| PartnerLinkDAO spl = fetchPartnerLinkDAO(pLink); |
| |
| return spl.getPartnerEPR() != null || _bpelProcess.getInitialPartnerRoleEPR(pLink.partnerLink) != null; |
| } |
| |
| /** |
| * @see BpelRuntimeContext#completedFault(org.apache.ode.bpel.runtime.channels.FaultData) |
| */ |
| public void completedFault(FaultData faultData) { |
| BpelProcess.__log.warn("Instance " + _dao.getInstanceId() + " of " + _bpelProcess.getPID() + " has completed with fault: " + faultData); |
| |
| _dao.setFault(faultData.getFaultName(), faultData.getExplanation(), faultData.getFaultLineNo(), faultData |
| .getActivityId(), faultData.getFaultMessage()); |
| |
| // send event |
| ProcessInstanceStateChangeEvent evt = new ProcessInstanceStateChangeEvent(); |
| evt.setOldState(_dao.getState()); |
| _dao.setState(ProcessState.STATE_COMPLETED_WITH_FAULT); |
| evt.setNewState(ProcessState.STATE_COMPLETED_WITH_FAULT); |
| sendEvent(evt); |
| |
| sendEvent(new ProcessCompletionEvent(faultData.getFaultName())); |
| _dao.finishCompletion(); |
| |
| faultOutstandingMessageExchanges(faultData); |
| |
| _bpelProcess._engine._contexts.scheduler.registerSynchronizer(new Scheduler.Synchronizer() { |
| public void afterCompletion(boolean success) { |
| } |
| public void beforeCompletion() { |
| _dao.delete(_bpelProcess.getCleanupCategories(false), false); |
| } |
| }); |
| } |
| |
| /** |
| * @see BpelRuntimeContext#completedOk() |
| */ |
| public void completedOk() { |
| if (BpelProcess.__log.isDebugEnabled()) { |
| BpelProcess.__log.debug("ProcessImpl " + _bpelProcess.getPID() + " completed OK."); |
| } |
| |
| // send event |
| ProcessInstanceStateChangeEvent evt = new ProcessInstanceStateChangeEvent(); |
| evt.setOldState(_dao.getState()); |
| _dao.setState(ProcessState.STATE_COMPLETED_OK); |
| evt.setNewState(ProcessState.STATE_COMPLETED_OK); |
| sendEvent(evt); |
| |
| sendEvent(new ProcessCompletionEvent(null)); |
| _dao.finishCompletion(); |
| |
| completeOutstandingMessageExchanges(); |
| |
| _bpelProcess._engine._contexts.scheduler.registerSynchronizer(new Scheduler.Synchronizer() { |
| public void afterCompletion(boolean success) { |
| } |
| public void beforeCompletion() { |
| _dao.delete(_bpelProcess.getCleanupCategories(true), false); |
| } |
| }); |
| } |
| |
| /** |
| * @see BpelRuntimeContext#createScopeInstance(Long, |
| * org.apache.ode.bpel.obj.OScope) |
| */ |
| public Long createScopeInstance(Long parentScopeId, OScope scope) { |
| if (BpelProcess.__log.isTraceEnabled()) { |
| BpelProcess.__log.trace(ObjectPrinter.stringifyMethodEnter("createScopeInstance", new Object[] { |
| "parentScopeId", parentScopeId, "scope", scope })); |
| } |
| |
| ScopeDAO parent = null; |
| |
| if (parentScopeId != null) { |
| parent = _dao.getScope(parentScopeId); |
| } |
| |
| ScopeDAO scopeDao = _dao.createScope(parent, scope.getName(), scope.getId()); |
| return scopeDao.getScopeInstanceId(); |
| } |
| |
| public void initializePartnerLinks(Long parentScopeId, Collection<OPartnerLink> partnerLinks) { |
| |
| if (BpelProcess.__log.isTraceEnabled()) { |
| BpelProcess.__log.trace(ObjectPrinter.stringifyMethodEnter("initializeEndpointReferences", new Object[] { |
| "parentScopeId", parentScopeId, "partnerLinks", partnerLinks })); |
| } |
| |
| ScopeDAO parent = _dao.getScope(parentScopeId); |
| for (OPartnerLink partnerLink : partnerLinks) { |
| PartnerLinkDAO pdao = parent.createPartnerLink(partnerLink.getId(), partnerLink.getName(), |
| partnerLink.getMyRoleName(), partnerLink.getPartnerRoleName()); |
| // If there is a myrole on the link, initialize the session id so it |
| // is always |
| // available for opaque correlations. The myrole session id should |
| // never be changed. |
| if (partnerLink.hasMyRole()) |
| pdao.setMySessionId(new GUID().toString()); |
| } |
| } |
| |
| public void select(PickResponse pickResponseChannel, Date timeout, boolean createInstance, |
| Selector[] selectors) throws FaultException { |
| if (BpelProcess.__log.isTraceEnabled()) |
| BpelProcess.__log.trace(ObjectPrinter.stringifyMethodEnter("select", new Object[] { "pickResponseChannel", |
| pickResponseChannel, "timeout", timeout, "createInstance", createInstance, |
| "selectors", selectors })); |
| |
| ProcessDAO processDao = _dao.getProcess(); |
| |
| // check if this is first pick |
| if (_dao.getState() == ProcessState.STATE_NEW) { |
| assert createInstance; |
| // send event |
| ProcessInstanceStateChangeEvent evt = new ProcessInstanceStateChangeEvent(); |
| evt.setOldState(ProcessState.STATE_NEW); |
| _dao.setState(ProcessState.STATE_READY); |
| evt.setNewState(ProcessState.STATE_READY); |
| sendEvent(evt); |
| } |
| |
| final String pickResponseChannelStr = ProcessUtil.exportChannel(pickResponseChannel); |
| |
| List<CorrelatorDAO> correlators = new ArrayList<CorrelatorDAO>(selectors.length); |
| for (Selector selector : selectors) { |
| String correlatorId = BpelProcess.genCorrelatorId(selector.plinkInstance.partnerLink, selector.opName); |
| if (BpelProcess.__log.isDebugEnabled()) { |
| BpelProcess.__log.debug("SELECT: " + pickResponseChannel + ": USING CORRELATOR " + correlatorId); |
| } |
| correlators.add(processDao.getCorrelator(correlatorId)); |
| } |
| |
| // Checking conflicts |
| int conflict = _imaManager.findConflict(selectors); |
| if (conflict != -1) |
| throw new FaultException(_bpelProcess.getOProcess().getConstants().getQnConflictingReceive(), selectors[conflict] |
| .toString()); |
| |
| // Check for ambiguous receive |
| for (int i = 0; i < selectors.length; ++i) { |
| CorrelatorDAO correlator = correlators.get(i); |
| Selector selector = selectors[i]; |
| |
| if (!correlator.checkRoute(selector.correlationKeySet)) { |
| throw new FaultException(_bpelProcess.getOProcess().getConstants().qnAmbiguousReceive(), selector.toString()); |
| } |
| } |
| |
| //Registering |
| _imaManager.register(pickResponseChannelStr, selectors); |
| |
| // First check if we match to a new instance. |
| if (_instantiatingMessageExchange != null && _dao.getState() == ProcessState.STATE_READY) { |
| if (BpelProcess.__log.isDebugEnabled()) { |
| BpelProcess.__log.debug("SELECT: " + pickResponseChannel + ": CHECKING for NEW INSTANCE match"); |
| } |
| for (int i = 0; i < correlators.size(); ++i) { |
| CorrelatorDAO ci = correlators.get(i); |
| if (ci.equals(_dao.getInstantiatingCorrelator())) { |
| inputMsgMatch(pickResponseChannelStr, i, _instantiatingMessageExchange); |
| if (BpelProcess.__log.isDebugEnabled()) { |
| BpelProcess.__log.debug("SELECT: " + pickResponseChannel |
| + ": FOUND match for NEW instance mexRef=" + _instantiatingMessageExchange); |
| } |
| return; |
| } |
| } |
| } |
| |
| if (timeout != null) { |
| registerTimer(pickResponseChannel, timeout); |
| if (BpelProcess.__log.isDebugEnabled()) { |
| BpelProcess.__log.debug("SELECT: " + pickResponseChannel + "REGISTERED TIMEOUT for " + timeout); |
| } |
| } |
| |
| for (int i = 0; i < selectors.length; ++i) { |
| CorrelatorDAO correlator = correlators.get(i); |
| Selector selector = selectors[i]; |
| |
| correlator.addRoute(ProcessUtil.exportChannel(pickResponseChannel), _dao, i, selector.correlationKeySet, selector.route); |
| scheduleCorrelatorMatcher(correlator.getCorrelatorId(), selector.correlationKeySet); |
| |
| if (BpelProcess.__log.isDebugEnabled()) { |
| BpelProcess.__log.debug("SELECT: " + pickResponseChannel + ": ADDED ROUTE " + correlator.getCorrelatorId() + ": " |
| + selector.correlationKeySet + " --> " + _dao.getInstanceId()); |
| } |
| } |
| } |
| |
| /** |
| * @see BpelRuntimeContext#readCorrelation(org.apache.ode.bpel.runtime.CorrelationSetInstance) |
| */ |
| public CorrelationKey readCorrelation(CorrelationSetInstance cset) { |
| ScopeDAO scopeDAO = _dao.getScope(cset.scopeInstance); |
| CorrelationSetDAO cs = scopeDAO.getCorrelationSet(cset.declaration.getName()); |
| return cs.getValue(); |
| } |
| |
| |
| public Element fetchPartnerRoleEndpointReferenceData(PartnerLinkInstance pLink) throws FaultException { |
| PartnerLinkDAO pl = fetchPartnerLinkDAO(pLink); |
| Element epr = pl.getPartnerEPR(); |
| |
| if (epr == null) { |
| EndpointReference e = _bpelProcess.getInitialPartnerRoleEPR(pLink.partnerLink); |
| if (e != null) |
| epr = e.toXML().getDocumentElement(); |
| } |
| |
| if (epr == null) { |
| throw new FaultException(_bpelProcess.getOProcess().getConstants().getQnUninitializedPartnerRole()); |
| } |
| |
| return epr; |
| } |
| |
| public Element fetchMyRoleEndpointReferenceData(PartnerLinkInstance pLink) { |
| return _bpelProcess.getInitialMyRoleEPR(pLink.partnerLink).toXML().getDocumentElement(); |
| } |
| |
| protected PartnerLinkDAO fetchPartnerLinkDAO(PartnerLinkInstance pLink) { |
| ScopeDAO scopeDAO = _dao.getScope(pLink.scopeInstanceId); |
| return scopeDAO.getPartnerLink(pLink.partnerLink.getId()); |
| } |
| |
| /** |
| * Evaluate a property alias query expression against a variable, returning |
| * the normalized {@link String} representation of the property value. |
| * |
| * @param variable |
| * variable to read |
| * @param property |
| * property to read |
| * @return value of property for variable, in String form |
| * @throws org.apache.ode.bpel.common.FaultException |
| * in case of selection or other fault |
| */ |
| public String readProperty(VariableInstance variable, OProcess.OProperty property) throws FaultException { |
| Node varData = readVariable(variable.scopeInstance, variable.declaration.getName(), false); |
| |
| OProcess.OPropertyAlias alias = property.getAlias(variable.declaration.getType()); |
| String val = _bpelProcess.extractProperty((Element) varData, Collections.EMPTY_MAP, alias, variable.declaration.getDescription()); |
| |
| if (BpelProcess.__log.isTraceEnabled()) { |
| BpelProcess.__log.trace("readPropertyAlias(variable=" + variable + ", alias=" + alias + ") = " |
| + val.toString()); |
| } |
| |
| return val; |
| } |
| |
| public void writeEndpointReference(PartnerLinkInstance variable, Element data) throws FaultException { |
| if (__log.isDebugEnabled()) { |
| __log.debug("Writing endpoint reference " + variable.partnerLink.getName() + " with value " |
| + DOMUtils.domToString(data)); |
| } |
| |
| PartnerLinkDAO eprDAO = fetchPartnerLinkDAO(variable); |
| eprDAO.setPartnerEPR(data); |
| } |
| |
| public String fetchEndpointSessionId(PartnerLinkInstance pLink, boolean isMyEPR) throws FaultException { |
| PartnerLinkDAO dao = fetchPartnerLinkDAO(pLink); |
| return isMyEPR ? dao.getMySessionId() : dao.getPartnerSessionId(); |
| } |
| |
| public Node convertEndpointReference(Element sourceNode, Node targetNode) { |
| QName nodeQName; |
| if (targetNode.getNodeType() == Node.TEXT_NODE) { |
| nodeQName = new QName(Namespaces.XML_SCHEMA, "string"); |
| } else { |
| // We have an element |
| nodeQName = new QName(targetNode.getNamespaceURI(), targetNode.getLocalName()); |
| } |
| return _bpelProcess._engine._contexts.eprContext.convertEndpoint(nodeQName, sourceNode).toXML(); |
| } |
| |
| |
| public Node readVariable(Long scopeInstanceId, String varname, boolean forWriting) throws FaultException { |
| ScopeDAO scopedao = _dao.getScope(scopeInstanceId); |
| XmlDataDAO var = scopedao.getVariable(varname); |
| return (var == null || var.isNull()) ? null : var.get(); |
| } |
| |
| public Node writeVariable(VariableInstance variable, Node changes) { |
| ScopeDAO scopeDAO = _dao.getScope(variable.scopeInstance); |
| XmlDataDAO dataDAO = scopeDAO.getVariable(variable.declaration.getName()); |
| dataDAO.set(changes); |
| |
| writeProperties(variable, changes, dataDAO); |
| return dataDAO.get(); |
| } |
| |
| public void cancelOutstandingRequests(String channelId) { |
| _imaManager.cancel(channelId, false); |
| } |
| |
| public void processOutstandingRequest(PartnerLinkInstance partnerLink, String opName, String bpelMexId, String odeMexId) throws FaultException { |
| String mexRef = _imaManager.processOutstandingRequest(partnerLink, opName, bpelMexId, odeMexId); |
| if (mexRef != null) { |
| reply2(partnerLink, opName, bpelMexId, null, _bpelProcess.getOProcess().getConstants().getQnConflictingRequest(), false, mexRef); |
| throw new FaultException(_bpelProcess.getOProcess().getConstants().getQnConflictingRequest()); |
| } |
| } |
| |
| public void reply(final PartnerLinkInstance plinkInstnace, final String opName, final String mexId, Element msg, |
| QName fault) throws FaultException { |
| String mexRef = _imaManager.release(plinkInstnace, opName, mexId); |
| reply2(plinkInstnace, opName, mexId, msg, fault, false, mexRef); |
| } |
| |
| public void reply2(final PartnerLinkInstance plinkInstnace, final String opName, final String mexId, Element msg, |
| QName fault, boolean failure, final String mexRef) throws FaultException { |
| |
| // prepare event |
| ProcessMessageExchangeEvent evt = new ProcessMessageExchangeEvent(); |
| evt.setMexId(mexId); |
| evt.setOperation(opName); |
| evt.setPortType(plinkInstnace.partnerLink.getMyRolePortType().getQName()); |
| |
| MessageExchangeDAO mex = _dao.getConnection().getMessageExchange(mexRef); |
| |
| MessageDAO message = mex.createMessage(plinkInstnace.partnerLink.getMyRoleOperation(opName).getOutput() |
| .getMessage().getQName()); |
| buildOutgoingMessage(message, msg); |
| |
| MyRoleMessageExchangeImpl m = new MyRoleMessageExchangeImpl(_bpelProcess, _bpelProcess._engine, mex); |
| _bpelProcess.initMyRoleMex(m); |
| m.setResponse(new MessageImpl(message)); |
| |
| if (failure) { |
| mex.setStatus(MessageExchange.Status.FAILURE.toString()); |
| } else if (fault != null) { |
| mex.setStatus(MessageExchange.Status.FAULT.toString()); |
| mex.setFault(fault); |
| evt.setAspect(ProcessMessageExchangeEvent.PROCESS_FAULT); |
| } else { |
| mex.setStatus(MessageExchange.Status.RESPONSE.toString()); |
| evt.setAspect(ProcessMessageExchangeEvent.PROCESS_OUTPUT); |
| } |
| |
| _bpelProcess.doAsyncReply(m, this); |
| |
| // send event |
| sendEvent(evt); |
| } |
| |
| /** |
| * @see BpelRuntimeContext#writeCorrelation(org.apache.ode.bpel.runtime.CorrelationSetInstance, |
| * org.apache.ode.bpel.common.CorrelationKey) |
| */ |
| public void writeCorrelation(CorrelationSetInstance cset, CorrelationKey correlation) { |
| ScopeDAO scopeDAO = _dao.getScope(cset.scopeInstance); |
| CorrelationSetDAO cs = scopeDAO.getCorrelationSet(cset.declaration.getName()); |
| QName[] propNames = new QName[cset.declaration.getProperties().size()]; |
| for (int m = 0; m < cset.declaration.getProperties().size(); m++) { |
| OProcess.OProperty oProperty = cset.declaration.getProperties().get(m); |
| propNames[m] = oProperty.getName(); |
| } |
| cs.setValue(propNames, correlation); |
| |
| CorrelationSetWriteEvent cswe = new CorrelationSetWriteEvent(cset.declaration.getName(), correlation); |
| cswe.setScopeId(cset.scopeInstance); |
| sendEvent(cswe); |
| |
| } |
| |
| /** |
| * Common functionality to initialize a correlation set based on data |
| * available in a variable. |
| * |
| * @param cset |
| * the correlation set instance |
| * @param variable |
| * variable instance |
| * |
| * @throws IllegalStateException |
| * DOCUMENTME |
| */ |
| public void initializeCorrelation(CorrelationSetInstance cset, VariableInstance variable) throws FaultException { |
| if (BpelProcess.__log.isDebugEnabled()) { |
| BpelProcess.__log.debug("Initializing correlation set " + cset.declaration.getName()); |
| } |
| // if correlation set is already initialized, then skip |
| if (isCorrelationInitialized(cset)) { |
| // if already set, we ignore |
| if (BpelProcess.__log.isDebugEnabled()) { |
| BpelProcess.__log.debug("OCorrelation set " + cset + " is already set: ignoring"); |
| } |
| return; |
| } |
| |
| String[] propNames = new String[cset.declaration.getProperties().size()]; |
| String[] propValues = new String[cset.declaration.getProperties().size()]; |
| |
| for (int i = 0; i < cset.declaration.getProperties().size(); ++i) { |
| OProcess.OProperty property = cset.declaration.getProperties().get(i); |
| propValues[i] = readProperty(variable, property); |
| propNames[i] = property.getName().toString(); |
| } |
| |
| CorrelationKey ckeyVal = new CorrelationKey(cset.declaration.getName(), propValues); |
| writeCorrelation(cset, ckeyVal); |
| } |
| |
| public ExpressionLanguageRuntimeRegistry getExpLangRuntime() { |
| return _bpelProcess._expLangRuntimeRegistry; |
| } |
| |
| /** |
| * @see BpelRuntimeContext#terminate() |
| */ |
| public void terminate() { |
| // send event |
| ProcessInstanceStateChangeEvent evt = new ProcessInstanceStateChangeEvent(); |
| evt.setOldState(_dao.getState()); |
| _dao.setState(ProcessState.STATE_TERMINATED); |
| evt.setNewState(ProcessState.STATE_TERMINATED); |
| sendEvent(evt); |
| sendEvent(new ProcessTerminationEvent()); |
| |
| _dao.finishCompletion(); |
| failOutstandingMessageExchanges(); |
| |
| _bpelProcess._engine._contexts.scheduler.registerSynchronizer(new Scheduler.Synchronizer() { |
| public void afterCompletion(boolean success) { |
| } |
| public void beforeCompletion() { |
| _dao.delete(_bpelProcess.getCleanupCategories(false), false); |
| } |
| }); |
| } |
| |
| public void registerTimer(TimerResponse timerChannel, Date timeToFire) { |
| JobDetails we = new JobDetails(); |
| we.setInstanceId(_dao.getInstanceId()); |
| we.setChannel(ProcessUtil.exportChannel(timerChannel)); |
| we.setType(JobType.TIMER); |
| we.setInMem(_bpelProcess.isInMemory()); |
| if(_bpelProcess.isInMemory()){ |
| _bpelProcess._engine._contexts.scheduler.scheduleVolatileJob(true, we, timeToFire); |
| }else{ |
| _bpelProcess._engine._contexts.scheduler.schedulePersistedJob(we, timeToFire); |
| } |
| } |
| |
| private void scheduleCorrelatorMatcher(String correlatorId, CorrelationKeySet keySet) { |
| JobDetails we = new JobDetails(); |
| we.setInstanceId(_dao.getInstanceId()); |
| we.setType(JobType.MATCHER); |
| we.setCorrelatorId(correlatorId); |
| we.setCorrelationKeySet(keySet); |
| we.setInMem(_bpelProcess.isInMemory()); |
| if(_bpelProcess.isInMemory()){ |
| _bpelProcess._engine._contexts.scheduler.scheduleVolatileJob(true, we); |
| }else{ |
| _bpelProcess._engine._contexts.scheduler.schedulePersistedJob(we, null); |
| } |
| } |
| |
| public void checkInvokeExternalPermission() {} |
| |
| /** |
| * Called back when the process executes an invokation. |
| * |
| * @param activityId The activity id in the process definition (id of OInvoke) |
| * @param partnerLinkInstance The partner link variable instance |
| * @param operation The wsdl operation. |
| * @param outboundMsg The message sent outside as a DOM |
| * @param invokeResponseChannel Object called back when the response is received. |
| * @return The instance id of the message exchange. |
| * @throws FaultException When the response is a fault or when the invoke could not be executed |
| * in which case it is one of the bpel standard fault. |
| */ |
| public String invoke(int aid, PartnerLinkInstance partnerLink, Operation operation, Element outgoingMessage, |
| InvokeResponse channel) throws FaultException { |
| |
| PartnerLinkDAO plinkDAO = fetchPartnerLinkDAO(partnerLink); |
| // The target (partner endpoint) -- if it has not been explicitly |
| // initialized |
| // then use the value from bthe deployment descriptor .. |
| Element partnerEPR = plinkDAO.getPartnerEPR(); |
| EndpointReference partnerEpr; |
| |
| if (partnerEPR == null) { |
| partnerEpr = _bpelProcess.getInitialPartnerRoleEPR(partnerLink.partnerLink); |
| // In this case, the partner link has not been initialized. |
| if (partnerEpr == null) |
| throw new FaultException(partnerLink.partnerLink.getOwner().getConstants().getQnUninitializedPartnerRole()); |
| } else { |
| partnerEpr = _bpelProcess._engine._contexts.eprContext.resolveEndpointReference(partnerEPR); |
| } |
| |
| if (BpelProcess.__log.isDebugEnabled()) { |
| BpelProcess.__log.debug("INVOKING PARTNER: partnerLink=" + partnerLink + |
| ", op=" + operation.getName() + " channel=" + channel + ")"); |
| } |
| |
| // prepare event |
| ProcessMessageExchangeEvent evt = new ProcessMessageExchangeEvent(); |
| evt.setOperation(operation.getName()); |
| evt.setPortType(partnerLink.partnerLink.getPartnerRolePortType().getQName()); |
| evt.setAspect(ProcessMessageExchangeEvent.PARTNER_INPUT); |
| |
| MessageExchangeDAO mexDao = _dao.getConnection().createMessageExchange( |
| MessageExchangeDAO.DIR_BPEL_INVOKES_PARTNERROLE); |
| mexDao.setCreateTime(getCurrentEventDateTime()); |
| mexDao.setStatus(MessageExchange.Status.NEW.toString()); |
| mexDao.setOperation(operation.getName()); |
| mexDao.setPortType(partnerLink.partnerLink.getPartnerRolePortType().getQName()); |
| mexDao.setPartnerLinkModelId(partnerLink.partnerLink.getId()); |
| mexDao.setPartnerLink(plinkDAO); |
| mexDao.setProcess(_dao.getProcess()); |
| mexDao.setInstance(_dao); |
| mexDao.setPattern((operation.getOutput() != null ? MessageExchangePattern.REQUEST_RESPONSE |
| : MessageExchangePattern.REQUEST_ONLY).toString()); |
| mexDao.setChannel(channel == null ? null : ProcessUtil.exportChannel(channel)); |
| |
| // Properties used by stateful-exchange protocol. |
| String mySessionId = plinkDAO.getMySessionId(); |
| String partnerSessionId = plinkDAO.getPartnerSessionId(); |
| |
| if ( mySessionId != null ) |
| mexDao.setProperty(MessageExchange.PROPERTY_SEP_MYROLE_SESSIONID, mySessionId); |
| if ( partnerSessionId != null ) |
| mexDao.setProperty(MessageExchange.PROPERTY_SEP_PARTNERROLE_SESSIONID, partnerSessionId); |
| |
| if (__log.isDebugEnabled()) |
| __log.debug("INVOKE PARTNER (SEP): sessionId=" + mySessionId + " partnerSessionId=" + partnerSessionId); |
| |
| MessageDAO message = null; |
| |
| if (operation.getInput() != null) |
| message = mexDao.createMessage(operation.getInput().getMessage().getQName()); |
| else |
| message = mexDao.createMessage(null); |
| |
| mexDao.setRequest(message); |
| |
| if (operation.getInput() != null) |
| message.setType(operation.getInput().getMessage().getQName()); |
| else |
| message.setType(null); |
| buildOutgoingMessage(message, outgoingMessage); |
| |
| // Get he my-role EPR (if myrole exists) for optional use by partner |
| // (for callback mechanism). |
| EndpointReference myRoleEndpoint = partnerLink.partnerLink.hasMyRole() ? _bpelProcess |
| .getInitialMyRoleEPR(partnerLink.partnerLink) : null; |
| PartnerRoleMessageExchangeImpl mex = |
| createPartnerRoleMessageExchangeImpl(mexDao, partnerLink, |
| operation, partnerEpr, myRoleEndpoint); |
| mex.setProperty("activityId", ""+aid); |
| |
| List<BpelProcess> p2pProcesses = null; |
| |
| Endpoint partnerEndpoint = _bpelProcess.getInitialPartnerRoleEndpoint(partnerLink.partnerLink); |
| if (getConfigForPartnerLink(partnerLink.partnerLink).usePeer2Peer && partnerEndpoint != null) |
| p2pProcesses = _bpelProcess.getEngine().route(partnerEndpoint.serviceName, mex.getRequest()); |
| |
| if (p2pProcesses != null && !p2pProcesses.isEmpty()) { |
| // Creating a my mex using the same message id as partner mex to "pipe" them |
| MyRoleMessageExchange myRoleMex = _bpelProcess.getEngine().createMessageExchange( |
| mex.getMessageExchangeId(), partnerEndpoint.serviceName, |
| operation.getName(), mex.getMessageExchangeId()); |
| |
| if (myRoleMex instanceof BrokeredMyRoleMessageExchangeImpl) { |
| mex.setSubscriberCount(((BrokeredMyRoleMessageExchangeImpl) myRoleMex).getSubscriberCount()); |
| } |
| |
| if (BpelProcess.__log.isDebugEnabled()) { |
| __log.debug("Invoking in a p2p interaction, partnerrole " + mex + " - myrole " + myRoleMex); |
| } |
| |
| Message odeRequest = myRoleMex.createMessage(operation.getInput().getMessage().getQName()); |
| odeRequest.setMessage(outgoingMessage); |
| ((MessageImpl)odeRequest)._dao.setHeader(message.getHeader()); |
| |
| if (BpelProcess.__log.isDebugEnabled()) { |
| __log.debug("Setting myRoleMex session ids for p2p interaction, mySession " |
| + partnerSessionId + " - partnerSess " + mySessionId); |
| } |
| if ( partnerSessionId != null ) |
| myRoleMex.setProperty(MessageExchange.PROPERTY_SEP_MYROLE_SESSIONID, partnerSessionId); |
| if ( mySessionId != null ) |
| myRoleMex.setProperty(MessageExchange.PROPERTY_SEP_PARTNERROLE_SESSIONID, mySessionId); |
| |
| mex.setStatus(MessageExchange.Status.REQUEST); |
| myRoleMex.invoke(odeRequest); |
| |
| // Can't expect any sync response |
| scheduleInvokeCheck(mex, partnerLink.partnerLink, true); |
| mex.replyAsync(); |
| } else { |
| // If we couldn't find the endpoint, then there is no sense |
| // in asking the IL to invoke. |
| if (partnerEpr != null) { |
| checkInvokeExternalPermission(); |
| mexDao.setEPR(partnerEpr.toXML().getDocumentElement()); |
| mex.setStatus(MessageExchange.Status.REQUEST); |
| // Assuming an unreliable protocol, we schedule a task to check if recovery mode will be needed |
| scheduleInvokeCheck(mex, partnerLink.partnerLink, false); |
| _bpelProcess._engine._contexts.mexContext.invokePartner(mex); |
| } else { |
| __log.error("Couldn't find endpoint for partner EPR " + DOMUtils.domToString(partnerEPR)); |
| mex.setFailure(FailureType.UNKNOWN_ENDPOINT, "UnknownEndpoint", partnerEPR); |
| } |
| } |
| |
| evt.setMexId(mexDao.getMessageExchangeId()); |
| sendEvent(evt); |
| |
| // MEX pattern is request only, at this point the status can only be a one way |
| if (mexDao.getPattern().equals(MessageExchangePattern.REQUEST_ONLY.toString())) { |
| mexDao.setStatus(MessageExchange.Status.ASYNC.toString()); |
| // This mex can now be released |
| boolean succeeded = mex.getStatus() != MessageExchange.Status.FAILURE && mex.getStatus() != MessageExchange.Status.FAULT; |
| mexDao.release(_bpelProcess.isCleanupCategoryEnabled(succeeded, CLEANUP_CATEGORY.MESSAGES)); |
| } |
| // Check if there is a synchronous response, if so, we need to inject the |
| // message on the response channel. |
| switch (mex.getStatus()) { |
| case NEW: |
| throw new AssertionError("Impossible!"); |
| case ASYNC: |
| break; |
| case RESPONSE: |
| case FAULT: |
| case FAILURE: |
| invocationResponse(mex); |
| break; |
| default: |
| __log.error("Partner did not acknowledge message exchange: " + mex); |
| mex.setFailure(FailureType.NO_RESPONSE, "Partner did not acknowledge.", null); |
| invocationResponse(mex); |
| } |
| |
| return mexDao.getMessageExchangeId(); |
| } |
| |
| // enable extensibility |
| protected PartnerRoleMessageExchangeImpl createPartnerRoleMessageExchangeImpl(MessageExchangeDAO mexDao, |
| PartnerLinkInstance partnerLink, Operation operation, EndpointReference partnerEpr, |
| EndpointReference myRoleEndpoint) { |
| return new PartnerRoleMessageExchangeImpl(getBpelProcess().getEngine(), mexDao, |
| partnerLink.partnerLink.getPartnerRolePortType(), operation, partnerEpr, myRoleEndpoint, |
| getBpelProcess().getPartnerRoleChannel(partnerLink.partnerLink)); |
| } |
| |
| protected BpelProcess getBpelProcess() { |
| return _bpelProcess; |
| } |
| |
| private void scheduleInvokeCheck(PartnerRoleMessageExchangeImpl mex, OPartnerLink partnerLink, boolean p2p) { |
| boolean isTwoWay = mex.getMessageExchangePattern() == |
| org.apache.ode.bpel.iapi.MessageExchange.MessageExchangePattern.REQUEST_RESPONSE; |
| if (!_bpelProcess.isInMemory() && isTwoWay) { |
| JobDetails event = new JobDetails(); |
| event.setMexId(mex.getMessageExchangeId()); |
| event.setProcessId(_bpelProcess.getPID()); |
| event.setInMem(false); |
| event.setType(JobType.INVOKE_CHECK); |
| // use a greater timeout to make sure the check job does not get executed while the service invocation is still waiting for a response |
| long timeout = getBpelProcess().getTimeout(partnerLink, p2p); |
| if (__log.isDebugEnabled()) __log.debug("Creating invocation check event in "+timeout+"ms for mexid " + mex.getMessageExchangeId()); |
| Date future = new Date(System.currentTimeMillis() + timeout); |
| String jobId = _bpelProcess._engine._contexts.scheduler.schedulePersistedJob(event, future); |
| mex.setProperty("invokeCheckJobId", jobId); |
| } |
| } |
| |
| protected void buildOutgoingMessage(MessageDAO message, Element outgoingElmt) { |
| if (outgoingElmt == null) return; |
| |
| Document doc = DOMUtils.newDocument(); |
| Element header = doc.createElement("header"); |
| NodeList parts = outgoingElmt.getChildNodes(); |
| for (int m = 0; m < parts.getLength(); m++) { |
| if (parts.item(m).getNodeType() == Node.ELEMENT_NODE) { |
| Element part = (Element) parts.item(m); |
| if (part.getAttribute("headerPart") != null && part.getAttribute("headerPart").length() > 0) { |
| header.appendChild(doc.importNode(part, true)); |
| // remove the element from the list AND decrement the index to avoid skipping the next element!! |
| outgoingElmt.removeChild(part); |
| m--; |
| } |
| } |
| } |
| message.setData(outgoingElmt); |
| message.setHeader(header); |
| } |
| |
| public void execute() { |
| long maxTime = System.currentTimeMillis() + _maxReductionTimeMs; |
| boolean canReduce = true; |
| assert _outstandingRequests == null && _imaManager != null; |
| while (ProcessState.canExecute(_dao.getState()) && System.currentTimeMillis() < maxTime && canReduce && !_forceFlush) { |
| canReduce = _vpu.execute(); |
| } |
| _dao.setLastActiveTime(new Date()); |
| if (!ProcessState.isFinished(_dao.getState())) { |
| if (__log.isDebugEnabled()) __log.debug("Setting execution state on instance " + _iid); |
| _soup.setGlobalData(_imaManager); |
| |
| if (_bpelProcess.isInMemory()) { |
| // don't serialize in-memory processes |
| ((ProcessInstanceDaoImpl) _dao).setSoup(_soup); |
| } else { |
| ByteArrayOutputStream bos = new ByteArrayOutputStream(10000); |
| try { |
| _soup.write(bos); |
| bos.close(); |
| } catch (Exception ex) { |
| throw new RuntimeException(ex); |
| } |
| _dao.setExecutionState(bos.toByteArray()); |
| } |
| |
| if (ProcessState.canExecute(_dao.getState()) && canReduce) { |
| // Max time exceeded (possibly an infinite loop). |
| if (__log.isDebugEnabled()) |
| __log.debug("MaxTime exceeded for instance # " + _iid); |
| try { |
| JobDetails we = new JobDetails(); |
| we.setInstanceId(_iid); |
| we.setType(JobType.RESUME); |
| we.setInMem(_bpelProcess.isInMemory()); |
| if (_bpelProcess.isInMemory()) |
| _bpelProcess._engine._contexts.scheduler.scheduleVolatileJob(true, we); |
| else |
| _bpelProcess._engine._contexts.scheduler.schedulePersistedJob(we, new Date()); |
| } catch (ContextException e) { |
| __log.error("Failed to schedule resume task.", e); |
| throw new BpelEngineException(e); |
| } |
| } |
| } |
| } |
| |
| public void inputMsgMatch(final String responsechannel, final int idx, MyRoleMessageExchangeImpl mex) { |
| // if we have a message match, this instance should be marked |
| // active if it isn't already |
| if (_dao.getState() == ProcessState.STATE_READY) { |
| if (BpelProcess.__log.isDebugEnabled()) { |
| BpelProcess.__log.debug("INPUTMSGMATCH: Changing process instance state from ready to active"); |
| } |
| |
| _dao.setState(ProcessState.STATE_ACTIVE); |
| |
| // send event |
| ProcessInstanceStateChangeEvent evt = new ProcessInstanceStateChangeEvent(); |
| evt.setOldState(ProcessState.STATE_READY); |
| evt.setNewState(ProcessState.STATE_ACTIVE); |
| sendEvent(evt); |
| } |
| |
| final String mexId = mex.getMessageExchangeId(); |
| _vpu.inject(new JacobRunnable() { |
| private static final long serialVersionUID = 3168964409165899533L; |
| |
| public void run() { |
| PickResponse pickResponse = importChannel(responsechannel, PickResponse.class); |
| pickResponse.onRequestRcvd(idx, mexId); |
| } |
| }); |
| } |
| |
| protected void timerEvent(final String timerResponseChannel) { |
| // In case this is a pick event, we remove routes, |
| // and cancel the outstanding requests. |
| _dao.getProcess().removeRoutes(timerResponseChannel, _dao); |
| _imaManager.cancel(timerResponseChannel, true); |
| |
| // Ignore timer events after the process is finished. |
| if (ProcessState.isFinished(_dao.getState())) { |
| return; |
| } |
| |
| _vpu.inject(new JacobRunnable() { |
| private static final long serialVersionUID = -7767141033611036745L; |
| |
| public void run() { |
| TimerResponse responseChannel = importChannel(timerResponseChannel, TimerResponse.class); |
| responseChannel.onTimeout(); |
| } |
| }); |
| execute(); |
| } |
| |
| public void cancel(final TimerResponse timerResponseChannel) { |
| // In case this is a pick response channel, we need to cancel routes and |
| // receive/reply association. |
| final String id = ProcessUtil.exportChannel(timerResponseChannel); |
| _dao.getProcess().removeRoutes(id, _dao); |
| _imaManager.cancel(id, true); |
| |
| _vpu.inject(new JacobRunnable() { |
| private static final long serialVersionUID = 6157913683737696396L; |
| |
| public void run() { |
| TimerResponse timerResponse = importChannel(id, TimerResponse.class); |
| timerResponse.onCancel(); |
| } |
| }); |
| } |
| |
| void invocationResponse(PartnerRoleMessageExchangeImpl mex) { |
| invocationResponse(mex.getDAO().getMessageExchangeId(), mex.getDAO().getChannel()); |
| } |
| |
| public void invocationResponse(final String mexid, final String responseChannelId) { |
| if (responseChannelId == null) |
| throw new NullPointerException("Null responseChannelId"); |
| if (mexid == null) |
| throw new NullPointerException("Null mexId"); |
| |
| if (BpelProcess.__log.isDebugEnabled()) { |
| __log.debug("Invoking message response for mexid " + mexid + " and channel " + responseChannelId); |
| } |
| _vpu.inject(new BpelJacobRunnable() { |
| private static final long serialVersionUID = -1095444335740879981L; |
| |
| public void run() { |
| ((BpelRuntimeContextImpl) getBpelRuntimeContext()).invocationResponse2( |
| mexid, importChannel(responseChannelId, InvokeResponse.class)); |
| } |
| }); |
| } |
| |
| /** |
| * Continuation of the above. |
| * |
| * @param mexid |
| * @param responseChannel |
| */ |
| private void invocationResponse2(String mexid, InvokeResponse responseChannel) { |
| __log.debug("Triggering response"); |
| MessageExchangeDAO mex = _dao.getConnection().getMessageExchange(mexid); |
| |
| ProcessMessageExchangeEvent evt = new ProcessMessageExchangeEvent(); |
| evt.setPortType(mex.getPortType()); |
| evt.setMexId(mexid); |
| evt.setOperation(mex.getOperation()); |
| |
| MessageExchange.Status status = MessageExchange.Status.valueOf(mex.getStatus()); |
| |
| switch (status) { |
| case FAULT: |
| evt.setAspect(ProcessMessageExchangeEvent.PARTNER_FAULT); |
| responseChannel.onFault(); |
| break; |
| case RESPONSE: |
| evt.setAspect(ProcessMessageExchangeEvent.PARTNER_OUTPUT); |
| responseChannel.onResponse(); |
| break; |
| case FAILURE: |
| evt.setAspect(ProcessMessageExchangeEvent.PARTNER_FAILURE); |
| responseChannel.onFailure(); |
| break; |
| default: |
| __log.error("Invalid response state for mex " + mexid + ": " + status); |
| } |
| sendEvent(evt); |
| } |
| |
| private void saveScopeState(Long scopeId, ScopeStateEnum scopeState) { |
| ScopeDAO scope = _dao.getScope(scopeId); |
| scope.setState(scopeState); |
| } |
| |
| /** |
| * @see BpelRuntimeContext#sendEvent(org.apache.ode.bpel.evt.ProcessInstanceEvent) |
| */ |
| public void sendEvent(ProcessInstanceEvent event) { |
| // fill in missing pieces |
| event.setProcessId(_dao.getProcess().getProcessId()); |
| event.setProcessName(_dao.getProcess().getType()); |
| event.setProcessInstanceId(_dao.getInstanceId()); |
| _bpelProcess._debugger.onEvent(event); |
| |
| // filter scopes |
| List<String> scopeNames = null; |
| if (event instanceof ScopeEvent) { |
| ScopeEvent sevent = (ScopeEvent) event; |
| |
| scopeNames = sevent.getParentScopesNames(); |
| |
| if (sevent instanceof ScopeStartEvent) { |
| saveScopeState(sevent.getScopeId(), ScopeStateEnum.ACTIVE); |
| } else if (sevent instanceof ScopeCompletionEvent) { |
| saveScopeState(sevent.getScopeId(), ScopeStateEnum.COMPLETED); |
| } else if (sevent instanceof ScopeFaultEvent) { |
| saveScopeState(sevent.getScopeId(), ScopeStateEnum.FAULT); |
| } |
| } |
| |
| // saving |
| _bpelProcess.saveEvent(event, _dao, scopeNames); |
| } |
| |
| public static String debugInfoToString(org.apache.ode.bpel.obj.DebugInfo debugInfo) { |
| if (debugInfo == null) return ""; |
| else return " at " + debugInfo.getSourceURI() + ":" + debugInfo.getStartLine(); |
| } |
| |
| /** |
| * We record all values of properties of a 'MessageType' variable for |
| * efficient lookup. |
| */ |
| private void writeProperties(VariableInstance variable, Node value, XmlDataDAO dao) { |
| if (variable.declaration.getType() instanceof OMessageVarType) { |
| for (OProcess.OProperty property : variable.declaration.getOwner().getProperties()) { |
| OProcess.OPropertyAlias alias = property.getAlias(variable.declaration.getType()); |
| if (alias != null) { |
| try { |
| String val = _bpelProcess.extractProperty((Element) value, Collections.EMPTY_MAP, alias, variable.declaration |
| .getDescription()); |
| if (val != null) { |
| dao.setProperty(property.getName().toString(), val); |
| } |
| } catch (FaultException e) { |
| // This will fail as we're basically trying to extract properties on all |
| // received messages for optimization purposes. |
| if (__log.isWarnEnabled()) |
| __log.warn("Couldn't extract property '" + property.toString() |
| + "' and variable " + variable.declaration + debugInfoToString(variable.declaration.getDebugInfo()) + " in property pre-extraction: " + e.toString()); |
| } |
| } |
| } |
| } |
| } |
| |
| private void completeOutstandingMessageExchanges() { |
| String[] mexRefs = _imaManager.releaseAll(); |
| for (String mexId : mexRefs) { |
| MessageExchangeDAO mexDao = _dao.getConnection().getMessageExchange(mexId); |
| if (mexDao != null) { |
| MyRoleMessageExchangeImpl mex = new MyRoleMessageExchangeImpl(_bpelProcess, _bpelProcess._engine, mexDao); |
| switch (mex.getStatus()) { |
| case ASYNC: |
| case RESPONSE: |
| mex.setStatus(MessageExchange.Status.COMPLETED_OK); |
| break; |
| case REQUEST: |
| if (mex.getPattern().equals(MessageExchange.MessageExchangePattern.REQUEST_ONLY)) { |
| mex.setStatus(MessageExchange.Status.COMPLETED_OK); |
| break; |
| } |
| default: |
| mex.setFailure(FailureType.OTHER, "No response.", null); |
| _bpelProcess.doAsyncReply(mex, this); |
| } |
| } |
| } |
| } |
| |
| private void faultOutstandingMessageExchanges(FaultData faultData) { |
| String[] mexRefs = _imaManager.releaseAll(); |
| for (String mexId : mexRefs) { |
| MessageExchangeDAO mexDao = _dao.getConnection().getMessageExchange(mexId); |
| if (mexDao != null) { |
| MyRoleMessageExchangeImpl mex = new MyRoleMessageExchangeImpl(_bpelProcess, _bpelProcess._engine, mexDao); |
| _bpelProcess.initMyRoleMex(mex); |
| |
| Message message = mex.createMessage(faultData.getFaultName()); |
| if (faultData.getFaultMessage() != null) |
| message.setMessage(faultData.getFaultMessage()); |
| mex.setResponse(message); |
| |
| mex.setFault(faultData.getFaultName(), message); |
| mex.setFaultExplanation(faultData.getExplanation()); |
| _bpelProcess.doAsyncReply(mex, this); |
| } |
| } |
| } |
| |
| private void failOutstandingMessageExchanges() { |
| String[] mexRefs = _imaManager.releaseAll(); |
| for (String mexId : mexRefs) { |
| MessageExchangeDAO mexDao = _dao.getConnection().getMessageExchange(mexId); |
| if (mexDao != null) { |
| MyRoleMessageExchangeImpl mex = new MyRoleMessageExchangeImpl(_bpelProcess, _bpelProcess._engine, mexDao); |
| _bpelProcess.initMyRoleMex(mex); |
| mex.setFailure(FailureType.OTHER, "No response.", null); |
| _bpelProcess.doAsyncReply(mex, this); |
| } |
| } |
| } |
| |
| public Element getPartnerResponse(String mexId) { |
| return mergeHeaders(_getPartnerResponse(mexId)); |
| } |
| |
| public Element getMyRequest(String mexId) { |
| MessageExchangeDAO dao = _dao.getConnection().getMessageExchange(mexId); |
| if (dao == null) { |
| // this should not happen.... |
| String msg = "Engine requested non-existent message exchange: " + mexId; |
| __log.error(msg); |
| throw new BpelEngineException(msg); |
| } |
| |
| if (dao.getDirection() != MessageExchangeDAO.DIR_PARTNER_INVOKES_MYROLE) { |
| // this should not happen.... |
| String msg = "Engine requested my-role request for a partner-role mex: " + mexId; |
| __log.error(msg); |
| throw new BpelEngineException(msg); |
| } |
| |
| MessageExchange.Status status = MessageExchange.Status.valueOf(dao.getStatus()); |
| switch (status) { |
| case ASYNC: |
| case REQUEST: |
| // In the case of pub-sub, the status may already be OK. |
| case COMPLETED_OK: |
| MessageDAO request = dao.getRequest(); |
| if (request == null) { |
| // this also should not happen |
| String msg = "Engine requested request for message exchange that did not have one: " + mexId; |
| __log.error(msg); |
| throw new BpelEngineException(msg); |
| } |
| return mergeHeaders(request); |
| default: |
| // We should not be in any other state when requesting this. |
| String msg = "Engine requested response while the message exchange " + mexId + " was in the state " |
| + status; |
| __log.error(msg); |
| throw new BpelEngineException(msg); |
| } |
| } |
| |
| private Element mergeHeaders(MessageDAO msg) { |
| if(msg==null) return null; |
| // Merging header data, it's all stored in the same variable |
| Element data = msg.getData(); |
| if (msg.getHeader() != null) { |
| if (data == null) { |
| Document doc = DOMUtils.newDocument(); |
| data = doc.createElement("message"); |
| doc.appendChild(data); |
| } |
| NodeList headerParts = msg.getHeader().getChildNodes(); |
| for (int m = 0; m < headerParts.getLength(); m++) { |
| if (headerParts.item(m).getNodeType() == Node.ELEMENT_NODE) { |
| Element headerPart = (Element) headerParts.item(m); |
| headerPart.setAttribute("headerPart", "true"); |
| data.appendChild(data.getOwnerDocument().importNode(headerPart, true)); |
| } |
| } |
| } |
| return data; |
| } |
| |
| public QName getPartnerFault(String mexId) { |
| MessageExchangeDAO mex = _getPartnerResponse(mexId).getMessageExchange(); |
| return mex.getFault(); |
| } |
| |
| public QName getPartnerResponseType(String mexId) { |
| return _getPartnerResponse(mexId).getType(); |
| } |
| |
| public String getPartnerFaultExplanation(String mexId) { |
| MessageExchangeDAO dao = _dao.getConnection().getMessageExchange(mexId); |
| return dao != null ? dao.getFaultExplanation() : null; |
| } |
| |
| private MessageDAO _getPartnerResponse(String mexId) { |
| MessageExchangeDAO dao = _dao.getConnection().getMessageExchange(mexId); |
| if (dao == null) { |
| // this should not happen.... |
| String msg = "Engine requested non-existent message exchange: " + mexId; |
| __log.error(msg); |
| throw new BpelEngineException(msg); |
| } |
| if (dao.getDirection() != MessageExchangeDAO.DIR_BPEL_INVOKES_PARTNERROLE) { |
| // this should not happen.... |
| String msg = "Engine requested partner response for a my-role mex: " + mexId; |
| __log.error(msg); |
| throw new BpelEngineException(msg); |
| } |
| |
| MessageDAO response; |
| MessageExchange.Status status = MessageExchange.Status.valueOf(dao.getStatus()); |
| switch (status) { |
| case FAULT: |
| case RESPONSE: |
| response = dao.getResponse(); |
| if (response == null) { |
| // this also should not happen |
| String msg = "Engine requested response for message exchange that did not have one: " + mexId; |
| __log.error(msg); |
| throw new BpelEngineException(msg); |
| } |
| break; |
| case FAILURE: |
| response = dao.getResponse(); |
| break; |
| default: |
| // We should not be in any other state when requesting this. |
| String msg = "Engine requested response while the message exchange " + mexId + " was in the state " |
| + status; |
| __log.error(msg); |
| throw new BpelEngineException(msg); |
| } |
| return response; |
| } |
| |
| public void releasePartnerMex(String mexId, boolean instanceSucceeded) { |
| MessageExchangeDAO dao = _dao.getConnection().getMessageExchange(mexId); |
| dao.release(_bpelProcess.isCleanupCategoryEnabled(instanceSucceeded, CLEANUP_CATEGORY.MESSAGES) ); |
| |
| // We used to cancel the invoke check job here but it turns out |
| // it creates more contention on the ODE_JOB table. It's better |
| // just to let the job get scheduled and discarded quietly |
| /* |
| String jobId = dao.getProperty("invokeCheckJobId"); |
| if (jobId != null) |
| _bpelProcess._engine._contexts.scheduler.cancelJob(jobId); |
| */ |
| } |
| |
| |
| public Element getSourceEPR(String mexId) { |
| MessageExchangeDAO dao = _dao.getConnection().getMessageExchange(mexId); |
| String epr = dao.getProperty(MessageExchange.PROPERTY_SEP_PARTNERROLE_EPR); |
| if (epr == null) |
| return null; |
| try { |
| Element eepr = DOMUtils.stringToDOM(epr); |
| return eepr; |
| } catch (Exception ex) { |
| __log.error("Invalid value for SEP property " + MessageExchange.PROPERTY_SEP_PARTNERROLE_EPR + ": " + epr); |
| } |
| |
| return null; |
| } |
| |
| public String getSourceSessionId(String mexId) { |
| MessageExchangeDAO dao = _dao.getConnection().getMessageExchange(mexId); |
| return dao.getProperty(MessageExchange.PROPERTY_SEP_PARTNERROLE_SESSIONID); |
| } |
| |
| public void registerActivityForRecovery(ActivityRecovery channel, long activityId, String reason, |
| Date dateTime, Element details, String[] actions, int retries) { |
| if (reason == null) |
| reason = "Unspecified"; |
| if (dateTime == null) |
| dateTime = new Date(); |
| __log.info("ActivityRecovery: Registering activity " + activityId + ", failure reason: " + reason + |
| " on channel " + ProcessUtil.exportChannel(channel)); |
| _dao.createActivityRecovery(ProcessUtil.exportChannel(channel), (int) activityId, reason, dateTime, details, actions, retries); |
| } |
| |
| public void unregisterActivityForRecovery(ActivityRecovery channel) { |
| _dao.deleteActivityRecovery(ProcessUtil.exportChannel(channel)); |
| } |
| |
| public void recoverActivity(final String channel, final long activityId, final String action, final FaultData fault) { |
| _vpu.inject(new JacobRunnable() { |
| private static final long serialVersionUID = 3168964409165899533L; |
| |
| public void run() { |
| ActivityRecovery recovery = importChannel(channel, ActivityRecovery.class); |
| __log.info("ActivityRecovery: Recovering activity " + activityId + " with action " + action + |
| " on channel " + recovery); |
| if (recovery != null) { |
| if ("cancel".equals(action)) |
| recovery.cancel(); |
| else if ("retry".equals(action)) |
| recovery.retry(); |
| else if ("fault".equals(action)) |
| recovery.fault(fault); |
| } |
| } |
| }); |
| //_dao.deleteActivityRecovery(channel); |
| execute(); |
| } |
| |
| /** |
| * Fetch the session-identifier for the partner link from the database. |
| */ |
| public String fetchMySessionId(PartnerLinkInstance pLink) { |
| String sessionId = fetchPartnerLinkDAO(pLink).getMySessionId(); |
| assert sessionId != null : "Session ID should always be set!"; |
| return sessionId; |
| } |
| |
| public String fetchPartnersSessionId(PartnerLinkInstance pLink) { |
| return fetchPartnerLinkDAO(pLink).getPartnerSessionId(); |
| } |
| |
| public void initializePartnersSessionId(PartnerLinkInstance pLink, String session) { |
| if (__log.isDebugEnabled()) |
| __log.debug("initializing partner " + pLink + " sessionId to " + session); |
| fetchPartnerLinkDAO(pLink).setPartnerSessionId(session); |
| |
| } |
| |
| /** |
| * Attempt to match message exchanges on a correlator. |
| * |
| */ |
| public void matcherEvent(String correlatorId, CorrelationKeySet ckeySet) { |
| if (BpelProcess.__log.isDebugEnabled()) { |
| __log.debug("MatcherEvent handling: correlatorId=" + correlatorId + ", ckeySet=" + ckeySet); |
| } |
| CorrelatorDAO correlator = _dao.getProcess().getCorrelator(correlatorId); |
| |
| // Find the route first, this is a SELECT FOR UPDATE on the "selector" row, |
| // So we want to acquire the lock before we do anthing else. |
| List<MessageRouteDAO> mroutes = correlator.findRoute(ckeySet); |
| if (mroutes == null || mroutes.size() == 0) { |
| // Ok, this means that a message arrived before we did, so nothing to do. |
| __log.debug("MatcherEvent handling: nothing to do, route no longer in DB"); |
| return; |
| } |
| |
| // Now see if there is a message that matches this selector. |
| MessageExchangeDAO mexdao = correlator.dequeueMessage(ckeySet); |
| if (mexdao != null) { |
| __log.debug("MatcherEvent handling: found matching message in DB (i.e. message arrived before <receive>)"); |
| if( MessageExchangePattern.REQUEST_RESPONSE.toString().equals(mexdao.getPattern())) { |
| __log.warn("A message arrived before a receive is ready for a request/response pattern. This may be processed to success. However, you should consider revising your process since a TCP port and a container thread will be held for a longer time and the process will not scale under heavy load."); |
| } |
| |
| for (MessageRouteDAO mroute : mroutes) { |
| // We have a match, so we can get rid of the routing entries. |
| correlator.removeRoutes(mroute.getGroupId(), _dao); |
| } |
| |
| // Selecting first route to proceed, other matching entries are ignored |
| MessageRouteDAO mroute = mroutes.get(0); |
| |
| // Found message matching one of our selectors. |
| if (BpelProcess.__log.isDebugEnabled()) { |
| BpelProcess.__log.debug("SELECT: " + mroute.getGroupId() + ": matched to MESSAGE " + mexdao |
| + " on CKEYSET " + ckeySet); |
| } |
| |
| MyRoleMessageExchangeImpl mex = new MyRoleMessageExchangeImpl(_bpelProcess, _bpelProcess._engine, mexdao); |
| |
| inputMsgMatch(mroute.getGroupId(), mroute.getIndex(), mex); |
| execute(); |
| |
| // Do not release yet if the process is suspended, the mex will be used again |
| if (_dao.getState() != ProcessState.STATE_SUSPENDED) |
| mexdao.releasePremieMessages(); |
| } else { |
| __log.debug("MatcherEvent handling: nothing to do, no matching message in DB"); |
| |
| } |
| } |
| |
| public Node readExtVar(Variable variable, Node reference) throws ExternalVariableModuleException { |
| Value val = _bpelProcess.getEVM().read(variable, reference, _iid); |
| return val.value; |
| } |
| |
| public ValueReferencePair writeExtVar(Variable variable, Node reference, Node value) throws ExternalVariableModuleException { |
| ValueReferencePair vrp = new ValueReferencePair(); |
| |
| Value val = _bpelProcess.getEVM().write(variable, reference, value, _iid); |
| vrp.reference = val.locator.reference; |
| vrp.value = val.value; |
| |
| return vrp; |
| } |
| |
| public URI getBaseResourceURI() { |
| return _bpelProcess.getBaseResourceURI(); |
| } |
| |
| public Node getProcessProperty(QName propertyName) { |
| return _bpelProcess.getProcessProperty(propertyName); |
| } |
| |
| public QName getProcessQName() { |
| return _bpelProcess.getProcessType(); |
| } |
| |
| public Date getCurrentEventDateTime() { |
| if (_currentEventDateTime == null) |
| return Calendar.getInstance().getTime(); |
| else |
| return _currentEventDateTime; |
| } |
| |
| public void setCurrentEventDateTime(Date eventDateTime) { |
| _currentEventDateTime = eventDateTime; |
| } |
| |
| public ClassLoader getProcessClassLoader() { |
| return _bpelProcess._classLoader; |
| } |
| |
| public PartnerRoleConfig getConfigForPartnerLink(OPartnerLink pLink) { |
| PartnerRoleConfig c = _bpelProcess.getConf().getPartnerRoleConfig().get(pLink.getName()); |
| if (c == null) { |
| return new PartnerRoleConfig(null, true); |
| } else { |
| return c; |
| } |
| } |
| |
| public void forceFlush() { |
| _forceFlush = true; |
| } |
| } |
