| /* |
| * 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.util.ArrayList; |
| import java.util.Iterator; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Set; |
| |
| import javax.wsdl.Operation; |
| import javax.wsdl.PortType; |
| 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.InvalidMessageException; |
| import org.apache.ode.bpel.common.OptionalCorrelationKey; |
| import org.apache.ode.bpel.dao.CorrelatorDAO; |
| import org.apache.ode.bpel.dao.MessageRouteDAO; |
| import org.apache.ode.bpel.dao.ProcessDAO; |
| import org.apache.ode.bpel.dao.ProcessInstanceDAO; |
| import org.apache.ode.bpel.evt.CorrelationMatchEvent; |
| import org.apache.ode.bpel.evt.CorrelationNoMatchEvent; |
| import org.apache.ode.bpel.evt.NewProcessInstanceEvent; |
| import org.apache.ode.bpel.iapi.Endpoint; |
| import org.apache.ode.bpel.iapi.MessageExchange; |
| import org.apache.ode.bpel.iapi.MyRoleMessageExchange; |
| import org.apache.ode.bpel.iapi.OdeGlobalConfig; |
| import org.apache.ode.bpel.iapi.ProcessState; |
| import org.apache.ode.bpel.iapi.Scheduler.JobDetails; |
| import org.apache.ode.bpel.iapi.Scheduler.JobType; |
| import org.apache.ode.bpel.intercept.InterceptorInvoker; |
| 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.runtime.InvalidProcessException; |
| import org.apache.ode.bpel.runtime.PROCESS; |
| import org.apache.ode.utils.ObjectPrinter; |
| import org.apache.ode.utils.msg.MessageBundle; |
| import org.w3c.dom.Element; |
| import org.w3c.dom.Node; |
| |
| /** |
| * @author Matthieu Riou <mriou at apache dot org> |
| */ |
| public class PartnerLinkMyRoleImpl extends PartnerLinkRoleImpl { |
| private static final Logger __log = LoggerFactory.getLogger(BpelProcess.class); |
| private static final Messages __msgs = MessageBundle.getMessages(Messages.class); |
| |
| /** The local endpoint for this "myrole". */ |
| public Endpoint _endpoint; |
| |
| PartnerLinkMyRoleImpl(BpelProcess process, OPartnerLink plink, Endpoint endpoint) { |
| super(process, plink); |
| _endpoint = endpoint; |
| } |
| |
| public String toString() { |
| StringBuffer buf = new StringBuffer("{PartnerLinkRole-"); |
| buf.append(_plinkDef.getName()); |
| buf.append('.'); |
| buf.append(_plinkDef.getMyRoleName()); |
| buf.append(" on "); |
| buf.append(_endpoint); |
| buf.append('}'); |
| |
| return buf.toString(); |
| } |
| |
| public boolean isCreateInstance(MyRoleMessageExchangeImpl mex) { |
| Operation operation = getMyRoleOperation(mex.getOperationName()); |
| if(operation == null) { |
| return false; |
| } |
| return _plinkDef.isCreateInstanceOperation(operation); |
| } |
| |
| public List<RoutingInfo> findRoute(MyRoleMessageExchangeImpl mex) { |
| List<RoutingInfo> routingInfos = new ArrayList<RoutingInfo>(); |
| |
| if (__log.isTraceEnabled()) { |
| __log.trace(ObjectPrinter.stringifyMethodEnter(this + ":inputMsgRcvd", new Object[] { |
| "messageExchange", mex })); |
| } |
| |
| Operation operation = getMyRoleOperation(mex.getOperationName()); |
| if (operation == null) { |
| // __log.error(__msgs.msgUnknownOperation(mex.getOperationName(), _plinkDef.myRolePortType.getQName())); |
| // mex.setFailure(MessageExchange.FailureType.UNKNOWN_OPERATION, mex.getOperationName(), null); |
| return null; |
| } |
| setMexRole(mex); |
| |
| // now, the tricks begin: when a message arrives we have to see if there |
| // is anyone waiting for it. Get the correlator, a persisted communication-reduction |
| // data structure supporting correlation correlationKey matching! |
| String correlatorId = BpelProcess.genCorrelatorId(_plinkDef, operation.getName()); |
| |
| CorrelatorDAO correlator = _process.getProcessDAO().getCorrelator(correlatorId); |
| |
| CorrelationKeySet keySet; |
| |
| // We need to compute the correlation keys (based on the operation |
| // we can infer which correlation keys to compute - this is merely a set |
| // consisting of each correlationKey used in each correlation sets |
| // that is ever referenced in an <receive>/<onMessage> on this |
| // partnerlink/operation. |
| try { |
| keySet = computeCorrelationKeys(mex); |
| } catch (InvalidMessageException ime) { |
| // We'd like to do a graceful exit here, no sense in rolling back due to a |
| // a message format problem. |
| __log.debug("Unable to evaluate correlation keys, invalid message format. ",ime); |
| mex.setFailure(MessageExchange.FailureType.FORMAT_ERROR, ime.getMessage(), null); |
| return null; |
| } |
| |
| String mySessionId = mex.getProperty(MessageExchange.PROPERTY_SEP_MYROLE_SESSIONID); |
| String partnerSessionId = mex.getProperty(MessageExchange.PROPERTY_SEP_PARTNERROLE_SESSIONID); |
| if (__log.isDebugEnabled()) { |
| __log.debug("INPUTMSG: " + correlatorId + ": MSG RCVD keys=" |
| + keySet + " mySessionId=" + mySessionId |
| + " partnerSessionId=" + partnerSessionId); |
| } |
| |
| // Try to find a route for one of our keys. |
| List<MessageRouteDAO> messageRoutes = correlator.findRoute(keySet); |
| if (messageRoutes != null && messageRoutes.size() > 0) { |
| for (MessageRouteDAO messageRoute : messageRoutes) { |
| if (__log.isDebugEnabled()) { |
| __log.debug("INPUTMSG: " + correlatorId + ": ckeySet " + messageRoute.getCorrelationKeySet() + " route is to " + messageRoute); |
| } |
| routingInfos.add(new RoutingInfo(messageRoute, messageRoute.getCorrelationKeySet(), correlator, keySet)); |
| } |
| } |
| |
| if (routingInfos.size() == 0) { |
| routingInfos.add(new RoutingInfo(null, null, correlator, keySet)); |
| } |
| |
| return routingInfos; |
| } |
| |
| public static class RoutingInfo { |
| public MessageRouteDAO messageRoute; |
| public CorrelationKeySet matchedKeySet; |
| public CorrelatorDAO correlator; |
| // CorrelationKey[] keys; |
| public CorrelationKeySet wholeKeySet; |
| |
| public RoutingInfo(MessageRouteDAO messageRoute, CorrelationKeySet matchedKeySet, |
| CorrelatorDAO correlator, CorrelationKeySet wholeKeySet) { |
| this.messageRoute = messageRoute; |
| this.matchedKeySet = matchedKeySet; |
| this.correlator = correlator; |
| this.wholeKeySet = wholeKeySet; |
| } |
| } |
| |
| public void invokeNewInstance(MyRoleMessageExchangeImpl mex, RoutingInfo routing) { |
| Operation operation = getMyRoleOperation(mex.getOperationName()); |
| |
| if (__log.isDebugEnabled()) { |
| __log.debug("INPUTMSG: " + routing.correlator.getCorrelatorId() + ": routing failed, CREATING NEW INSTANCE"); |
| } |
| ProcessDAO processDAO = _process.getProcessDAO(); |
| |
| if (_process._pconf.getState() == ProcessState.RETIRED) { |
| throw new InvalidProcessException("Process is retired.", InvalidProcessException.RETIRED_CAUSE_CODE); |
| } |
| |
| if (!_process.processInterceptors(mex, InterceptorInvoker.__onNewInstanceInvoked)) { |
| if (__log.isDebugEnabled()) { |
| __log.debug("Not creating a new instance for mex " + mex + "; interceptor prevented!"); |
| } |
| throw new InvalidProcessException("Cannot instantiate process '" + _process.getPID() + "' any more.", InvalidProcessException.TOO_MANY_INSTANCES_CAUSE_CODE); |
| } |
| |
| ProcessInstanceDAO newInstance = processDAO.createInstance(routing.correlator); |
| |
| BpelRuntimeContextImpl instance = _process |
| .createRuntimeContext(newInstance, new PROCESS(_process.getOProcess()), mex); |
| |
| // send process instance event |
| NewProcessInstanceEvent evt = new NewProcessInstanceEvent(new QName(_process.getOProcess().getTargetNamespace(), |
| _process.getOProcess().getName()), _process.getProcessDAO().getProcessId(), newInstance.getInstanceId()); |
| evt.setPortType(mex.getPortType().getQName()); |
| evt.setOperation(operation.getName()); |
| evt.setMexId(mex.getMessageExchangeId()); |
| _process._debugger.onEvent(evt); |
| _process.saveEvent(evt, newInstance); |
| mex.setCorrelationStatus(MyRoleMessageExchange.CorrelationStatus.CREATE_INSTANCE); |
| mex.getDAO().setInstance(newInstance); |
| if (mex.getDAO().getCreateTime() == null) |
| mex.getDAO().setCreateTime(instance.getCurrentEventDateTime()); |
| |
| _process._engine.acquireInstanceLock(newInstance.getInstanceId()); |
| instance.execute(); |
| } |
| |
| public void invokeInstance(MyRoleMessageExchangeImpl mex, RoutingInfo routing) { |
| Operation operation = getMyRoleOperation(mex.getOperationName()); |
| if (__log.isDebugEnabled()) { |
| __log.debug("INPUTMSG: " + routing.correlator.getCorrelatorId() + ": ROUTING to existing instance " |
| + routing.messageRoute.getTargetInstance().getInstanceId()); |
| } |
| |
| ProcessInstanceDAO instanceDao = routing.messageRoute.getTargetInstance(); |
| BpelProcess process2 = _process._engine._activeProcesses.get(instanceDao.getProcess().getProcessId()); |
| |
| // Reload process instance for DAO. |
| BpelRuntimeContextImpl instance = process2.createRuntimeContext(instanceDao, null, null); |
| instance.inputMsgMatch(routing.messageRoute.getGroupId(), routing.messageRoute.getIndex(), mex); |
| |
| // Kill the route so some new message does not get routed to |
| // same process instance. |
| routing.correlator.removeRoutes(routing.messageRoute.getGroupId(), instanceDao); |
| |
| // send process instance event |
| CorrelationMatchEvent evt = new CorrelationMatchEvent(new QName(process2.getOProcess().getTargetNamespace(), |
| process2.getOProcess().getName()), process2.getProcessDAO().getProcessId(), |
| instanceDao.getInstanceId(), routing.matchedKeySet); |
| evt.setPortType(mex.getPortType().getQName()); |
| evt.setOperation(operation.getName()); |
| evt.setMexId(mex.getMessageExchangeId()); |
| |
| process2._debugger.onEvent(evt); |
| // store event |
| process2.saveEvent(evt, instanceDao); |
| |
| mex.setCorrelationStatus(MyRoleMessageExchange.CorrelationStatus.MATCHED); |
| mex.getDAO().setInstance(routing.messageRoute.getTargetInstance()); |
| if (mex.getDAO().getCreateTime() == null) |
| mex.getDAO().setCreateTime(instance.getCurrentEventDateTime()); |
| |
| instance.execute(); |
| } |
| |
| public void noRoutingMatch(MyRoleMessageExchangeImpl mex, List<RoutingInfo> routings) { |
| if (!mex.isAsynchronous()) { |
| mex.setFailure(MessageExchange.FailureType.NOMATCH, "No process instance matching correlation keys.", null); |
| if (!OdeGlobalConfig.queueInOutMessages()) { |
| _process.doAsyncReply(mex, null); |
| } |
| } else { |
| // enqueue message with the last message route, as per the comments in caller (@see BpelProcess.invokeProcess()) |
| RoutingInfo routing = |
| (routings != null && routings.size() > 0) ? |
| routings.get(routings.size() - 1) : null; |
| if (routing != null) { |
| if (__log.isDebugEnabled()) { |
| __log.debug("INPUTMSG: " + routing.correlator.getCorrelatorId() + ": SAVING to DB (no match) "); |
| } |
| |
| // send event |
| CorrelationNoMatchEvent evt = new CorrelationNoMatchEvent(mex.getPortType().getQName(), mex |
| .getOperation().getName(), mex.getMessageExchangeId(), routing.wholeKeySet); |
| |
| evt.setProcessId(_process.getProcessDAO().getProcessId()); |
| evt.setProcessName(new QName(_process.getOProcess().getTargetNamespace(), _process.getOProcess().getName())); |
| _process._debugger.onEvent(evt); |
| |
| mex.setCorrelationStatus(MyRoleMessageExchange.CorrelationStatus.QUEUED); |
| |
| // No match, means we add message exchange to the queue. |
| routing.correlator.enqueueMessage(mex.getDAO(), routing.wholeKeySet); |
| |
| // Second matcher needs to be registered here |
| JobDetails we = new JobDetails(); |
| we.setType(JobType.MEX_MATCHER); |
| we.setProcessId(_process.getPID()); |
| we.setMexId(mex.getMessageExchangeId()); |
| we.setInMem(_process.isInMemory()); |
| if(_process.isInMemory()){ |
| _process._engine._contexts.scheduler.scheduleVolatileJob(true, we); |
| }else{ |
| _process._engine._contexts.scheduler.schedulePersistedJob(we, null); |
| } |
| } |
| } |
| } |
| |
| private void setMexRole(MyRoleMessageExchangeImpl mex) { |
| Operation operation = getMyRoleOperation(mex.getOperationName()); |
| mex.getDAO().setPartnerLinkModelId(_plinkDef.getId()); |
| mex.setPortOp(_plinkDef.getMyRolePortType(), operation); |
| mex.setPattern(operation.getOutput() == null ? MessageExchange.MessageExchangePattern.REQUEST_ONLY |
| : MessageExchange.MessageExchangePattern.REQUEST_RESPONSE); |
| } |
| |
| private Operation getMyRoleOperation(String operationName) { |
| return _plinkDef.getMyRoleOperation(operationName); |
| } |
| |
| private CorrelationKeySet computeCorrelationKeys(MyRoleMessageExchangeImpl mex) { |
| CorrelationKeySet keySet = new CorrelationKeySet(); |
| |
| Operation operation = mex.getOperation(); |
| Element msg = mex.getRequest().getMessage(); |
| Map<String, Node> headerParts = mex.getRequest().getHeaderParts(); |
| javax.wsdl.Message msgDescription = operation.getInput().getMessage(); |
| |
| Set<OScope.CorrelationSet> csets = _plinkDef.getNonInitiatingCorrelationSetsForOperation(operation); |
| for (OScope.CorrelationSet cset : csets) { |
| CorrelationKey key = computeCorrelationKey(cset, |
| _process.getOProcess().getMessageTypes().get(msgDescription.getQName()), msg, headerParts); |
| keySet.add(key); |
| } |
| |
| csets = _plinkDef.getJoinningCorrelationSetsForOperation(operation); |
| for (OScope.CorrelationSet cset : csets) { |
| CorrelationKey key = computeCorrelationKey(cset, |
| _process.getOProcess().getMessageTypes().get(msgDescription.getQName()), msg, headerParts); |
| keySet.add(key); |
| } |
| |
| // Let's creata a key based on the sessionId |
| String mySessionId = mex.getProperty(MessageExchange.PROPERTY_SEP_MYROLE_SESSIONID); |
| if (mySessionId != null) |
| keySet.add(new CorrelationKey("-1", new String[] { mySessionId })); |
| |
| return keySet; |
| } |
| |
| @SuppressWarnings("unchecked") |
| private CorrelationKey computeCorrelationKey(OScope.CorrelationSet cset, OMessageVarType messagetype, |
| Element msg, Map<String, Node> headerParts) { |
| CorrelationKey key = null; |
| |
| String[] values = new String[cset.getProperties().size()]; |
| |
| int jIdx = 0; |
| for (Iterator j = cset.getProperties().iterator(); j.hasNext(); ++jIdx) { |
| OProcess.OProperty property = (OProcess.OProperty) j.next(); |
| OProcess.OPropertyAlias alias = property.getAlias(messagetype); |
| |
| if (alias == null) { |
| // TODO: Throw a real exception! And catch this at compile |
| // time. |
| throw new IllegalArgumentException("No alias matching property '" + property.getName() |
| + "' with message type '" + messagetype + "'"); |
| } |
| |
| String value; |
| try { |
| value = _process.extractProperty(msg, headerParts, alias, msg.toString()); |
| } catch (FaultException fe) { |
| String emsg = __msgs.msgPropertyAliasDerefFailedOnMessage(alias.getDescription(), fe.getMessage()); |
| __log.error(emsg, fe); |
| throw new InvalidMessageException(emsg, fe); |
| } |
| values[jIdx] = value; |
| } |
| |
| if( cset.isHasJoinUseCases() ) { |
| key = new OptionalCorrelationKey(cset.getName(), values); |
| } else { |
| key = new CorrelationKey(cset.getName(), values); |
| } |
| |
| return key; |
| } |
| |
| @SuppressWarnings("unchecked") |
| public boolean isOneWayOnly() { |
| PortType portType = _plinkDef.getMyRolePortType(); |
| if (portType == null) { |
| return false; |
| } |
| for (Operation operation : (List<Operation>) portType.getOperations()) { |
| if (operation.getOutput() != null) { |
| return false; |
| } |
| } |
| return true; |
| } |
| } |
