Modified: webservices/sandesha/trunk/src/org/apache/sandesha2/util/SandeshaUtil.java URL: http://svn.apache.org/viewcvs/webservices/sandesha/trunk/src/org/apache/sandesha2/util/SandeshaUtil.java?rev=391959&r1=391958&r2=391959&view=diff ============================================================================== --- webservices/sandesha/trunk/src/org/apache/sandesha2/util/SandeshaUtil.java (original) +++ webservices/sandesha/trunk/src/org/apache/sandesha2/util/SandeshaUtil.java Thu Apr 6 03:50:20 2006 @@ -91,12 +91,6 @@ // private static Hashtable storedMsgContexts = new Hashtable(); - private static StorageManager storageManager = null; - - private static Sender sender = new Sender(); - - private static InOrderInvoker invoker = new InOrderInvoker(); - private static Log log = LogFactory.getLog(SandeshaUtil.class); /** @@ -214,21 +208,41 @@ } public static void startSenderForTheSequence(ConfigurationContext context, String sequenceID) { - sender.runSenderForTheSequence(context, sequenceID); + + Sender sender = (Sender) context.getProperty(Sandesha2Constants.SENDER); + + if (sender!=null) + sender.runSenderForTheSequence(context, sequenceID); + else { + sender = new Sender (); + context.setProperty(Sandesha2Constants.SENDER,sender); + sender.runSenderForTheSequence(context, sequenceID); + } } public static void stopSenderForTheSequence(String sequenceID) { - sender.stopSenderForTheSequence(sequenceID); +// sender.stopSenderForTheSequence(sequenceID); } public static void startInvokerForTheSequence(ConfigurationContext context, String sequenceID) { - if (!invoker.isInvokerStarted()) { - invoker.runInvokerForTheSequence(context, sequenceID); +// if (!invoker.isInvokerStarted()) { +// invoker.runInvokerForTheSequence(context, sequenceID); +// } + + InOrderInvoker invoker = (InOrderInvoker) context.getProperty(Sandesha2Constants.INVOKER); + if (invoker!=null) + invoker.runInvokerForTheSequence(context,sequenceID); + else { + invoker = new InOrderInvoker (); + context.setProperty(Sandesha2Constants.INVOKER,invoker); + invoker.runInvokerForTheSequence(context,sequenceID); } + + } public static void stopInvokerForTheSequence(String sequenceID) { - invoker.stopInvokerForTheSequence(sequenceID); +// invoker.stopInvokerForTheSequence(sequenceID); } public static String getMessageTypeString(int messageType) { @@ -310,14 +324,12 @@ */ public static StorageManager getSandeshaStorageManager(ConfigurationContext context) throws SandeshaException { + StorageManager storageManager = (StorageManager) context.getProperty(Sandesha2Constants.STORAGE_MANAGER); if (storageManager != null) return storageManager; String srotageManagerClassStr = PropertyManager.getInstance().getStorageManagerClass(); - if (storageManager != null) - return storageManager; - try { Class c = Class.forName(srotageManagerClassStr); Class configContextClass = Class.forName(context.getClass().getName()); @@ -329,8 +341,9 @@ StorageManager mgr = (StorageManager) obj; storageManager = mgr; + context.setProperty(Sandesha2Constants.STORAGE_MANAGER,storageManager); return storageManager; - + } catch (Exception e) { String message = "Cannot load the given storage manager"; log.error(message); @@ -621,7 +634,7 @@ return Sandesha2Constants.INTERNAL_SEQUENCE_PREFIX + ":" + to + ":" + sequenceKey; } - public static String getInternalSequenceID(String sequenceID) { + public static String getOutgoingSideInternalSequenceID(String sequenceID) { return Sandesha2Constants.INTERNAL_SEQUENCE_PREFIX + ":" + sequenceID; }
Modified: webservices/sandesha/trunk/src/org/apache/sandesha2/util/SequenceManager.java URL: http://svn.apache.org/viewcvs/webservices/sandesha/trunk/src/org/apache/sandesha2/util/SequenceManager.java?rev=391959&r1=391958&r2=391959&view=diff ============================================================================== --- webservices/sandesha/trunk/src/org/apache/sandesha2/util/SequenceManager.java (original) +++ webservices/sandesha/trunk/src/org/apache/sandesha2/util/SequenceManager.java Thu Apr 6 03:50:20 2006 @@ -27,9 +27,9 @@ import org.apache.sandesha2.RMMsgContext; import org.apache.sandesha2.Sandesha2Constants; import org.apache.sandesha2.SandeshaException; -import org.apache.sandesha2.SpecSpecificConstants; import org.apache.sandesha2.Sandesha2Constants.WSA; -import org.apache.sandesha2.client.Sandesha2ClientAPI; +import org.apache.sandesha2.client.RMClientAPI; +import org.apache.sandesha2.client.RMClientConstants; import org.apache.sandesha2.policy.RMPolicyBean; import org.apache.sandesha2.storage.StorageManager; import org.apache.sandesha2.storage.Transaction; @@ -207,7 +207,7 @@ EndpointReference toEPR = firstAplicationMsgCtx.getTo(); String acksTo = (String) firstAplicationMsgCtx - .getProperty(Sandesha2ClientAPI.AcksTo); + .getProperty(RMClientConstants.AcksTo); if (toEPR == null) { String message = "WS-Addressing To is null"; @@ -323,7 +323,7 @@ String transportInProtocol = messageContext.getOptions().getTransportInProtocol(); - String acksTo = (String) messageContext.getProperty(Sandesha2ClientAPI.AcksTo); + String acksTo = (String) messageContext.getProperty(RMClientConstants.AcksTo); String mep = messageContext.getAxisOperation().getMessageExchangePattern(); boolean startListnerForAsyncAcks = false; Added: webservices/sandesha/trunk/src/org/apache/sandesha2/util/SpecSpecificConstants.java URL: http://svn.apache.org/viewcvs/webservices/sandesha/trunk/src/org/apache/sandesha2/util/SpecSpecificConstants.java?rev=391959&view=auto ============================================================================== --- webservices/sandesha/trunk/src/org/apache/sandesha2/util/SpecSpecificConstants.java (added) +++ webservices/sandesha/trunk/src/org/apache/sandesha2/util/SpecSpecificConstants.java Thu Apr 6 03:50:20 2006 @@ -0,0 +1,222 @@ +package org.apache.sandesha2.util; + +import org.apache.axis2.addressing.AddressingConstants; +import org.apache.sandesha2.Sandesha2Constants; +import org.apache.sandesha2.SandeshaException; +import org.apache.sandesha2.Sandesha2Constants.SPEC_2005_02; +import org.apache.sandesha2.Sandesha2Constants.SPEC_2005_10; +import org.apache.sandesha2.Sandesha2Constants.SPEC_VERSIONS; +import org.apache.sandesha2.Sandesha2Constants.SPEC_2005_02.Actions; + +public class SpecSpecificConstants { + + private static String unknownSpecErrorMessage = "Unknown specification version"; + + public static String getSpecVersionString (String namespaceValue) throws SandeshaException { + if (Sandesha2Constants.SPEC_2005_02.NS_URI.equals(namespaceValue)) + return Sandesha2Constants.SPEC_VERSIONS.WSRM; + else if (Sandesha2Constants.SPEC_2005_10.NS_URI.equals(namespaceValue)) + return Sandesha2Constants.SPEC_VERSIONS.WSRX; + else + throw new SandeshaException ("Unknows rm namespace value"); + } + + public static String getRMNamespaceValue (String specVersion) throws SandeshaException { + if (Sandesha2Constants.SPEC_VERSIONS.WSRM.equals(specVersion)) + return Sandesha2Constants.SPEC_2005_02.NS_URI; + else if (Sandesha2Constants.SPEC_VERSIONS.WSRX.equals(specVersion)) + return Sandesha2Constants.SPEC_2005_10.NS_URI; + else + throw new SandeshaException (unknownSpecErrorMessage); + } + + public static String getCreateSequenceAction (String specVersion) throws SandeshaException { + if (Sandesha2Constants.SPEC_VERSIONS.WSRM.equals(specVersion)) + return Sandesha2Constants.SPEC_2005_02.Actions.ACTION_CREATE_SEQUENCE; + else if (Sandesha2Constants.SPEC_VERSIONS.WSRX.equals(specVersion)) + return Sandesha2Constants.SPEC_2005_10.Actions.ACTION_CREATE_SEQUENCE; + else + throw new SandeshaException (unknownSpecErrorMessage); + } + + public static String getCreateSequenceResponseAction (String specVersion) throws SandeshaException { + if (Sandesha2Constants.SPEC_VERSIONS.WSRM.equals(specVersion)) + return Sandesha2Constants.SPEC_2005_02.Actions.ACTION_CREATE_SEQUENCE_RESPONSE; + else if (Sandesha2Constants.SPEC_VERSIONS.WSRX.equals(specVersion)) + return Sandesha2Constants.SPEC_2005_10.Actions.ACTION_CREATE_SEQUENCE_RESPONSE; + else + throw new SandeshaException (unknownSpecErrorMessage); + } + + public static String getTerminateSequenceAction (String specVersion) throws SandeshaException { + if (Sandesha2Constants.SPEC_VERSIONS.WSRM.equals(specVersion)) + return Sandesha2Constants.SPEC_2005_02.Actions.ACTION_TERMINATE_SEQUENCE; + else if (Sandesha2Constants.SPEC_VERSIONS.WSRX.equals(specVersion)) + return Sandesha2Constants.SPEC_2005_10.Actions.ACTION_TERMINATE_SEQUENCE; + else + throw new SandeshaException (unknownSpecErrorMessage); + } + + public static String getTerminateSequenceResponseAction (String specVersion) throws SandeshaException { + if (Sandesha2Constants.SPEC_VERSIONS.WSRX.equals(specVersion)) + return Sandesha2Constants.SPEC_2005_10.Actions.ACTION_TERMINATE_SEQUENCE_RESPONSE; + else + throw new SandeshaException (unknownSpecErrorMessage); + } + + public static String getCloseSequenceAction (String specVersion) throws SandeshaException { + if (Sandesha2Constants.SPEC_VERSIONS.WSRM.equals(specVersion)) + throw new SandeshaException ("This rm spec version does not define a sequenceClose action"); + else if (Sandesha2Constants.SPEC_VERSIONS.WSRX.equals(specVersion)) + return Sandesha2Constants.SPEC_2005_10.Actions.ACTION_CLOSE_SEQUENCE; + else + throw new SandeshaException (unknownSpecErrorMessage); + } + + public static String getCloseSequenceResponseAction (String specVersion) throws SandeshaException { + if (Sandesha2Constants.SPEC_VERSIONS.WSRM.equals(specVersion)) + throw new SandeshaException ("This rm spec version does not define a sequenceClose action"); + else if (Sandesha2Constants.SPEC_VERSIONS.WSRX.equals(specVersion)) + return Sandesha2Constants.SPEC_2005_10.Actions.ACTION_CLOSE_SEQUENCE_RESPONSE; + else + throw new SandeshaException (unknownSpecErrorMessage); + } + + public static String getAckRequestAction (String specVersion) throws SandeshaException { + if (Sandesha2Constants.SPEC_VERSIONS.WSRM.equals(specVersion)) + throw new SandeshaException ("this spec version does not define a ackRequest action"); + else if (Sandesha2Constants.SPEC_VERSIONS.WSRX.equals(specVersion)) + return Sandesha2Constants.SPEC_2005_10.Actions.ACTION_ACK_REQUEST; + else + throw new SandeshaException (unknownSpecErrorMessage); + } + + public static String getSequenceAcknowledgementAction (String specVersion) throws SandeshaException { + if (Sandesha2Constants.SPEC_VERSIONS.WSRM.equals(specVersion)) + return Sandesha2Constants.SPEC_2005_02.Actions.ACTION_SEQUENCE_ACKNOWLEDGEMENT; + else if (Sandesha2Constants.SPEC_VERSIONS.WSRX.equals(specVersion)) + return Sandesha2Constants.SPEC_2005_10.Actions.ACTION_SEQUENCE_ACKNOWLEDGEMENT; + else + throw new SandeshaException (unknownSpecErrorMessage); + } + + public static String getCreateSequenceSOAPAction (String specVersion) throws SandeshaException { + if (Sandesha2Constants.SPEC_VERSIONS.WSRM.equals(specVersion)) + return Sandesha2Constants.SPEC_2005_02.Actions.SOAP_ACTION_CREATE_SEQUENCE; + else if (Sandesha2Constants.SPEC_VERSIONS.WSRX.equals(specVersion)) + return Sandesha2Constants.SPEC_2005_10.Actions.SOAP_ACTION_CREATE_SEQUENCE; + else + throw new SandeshaException (unknownSpecErrorMessage); + } + + public static String getCreateSequenceResponseSOAPAction (String specVersion) throws SandeshaException { + if (Sandesha2Constants.SPEC_VERSIONS.WSRM.equals(specVersion)) + return Sandesha2Constants.SPEC_2005_02.Actions.SOAP_ACTION_CREATE_SEQUENCE_RESPONSE; + else if (Sandesha2Constants.SPEC_VERSIONS.WSRX.equals(specVersion)) + return Sandesha2Constants.SPEC_2005_10.Actions.SOAP_ACTION_CREATE_SEQUENCE_RESPONSE; + else + throw new SandeshaException (unknownSpecErrorMessage); + } + + public static String getTerminateSequenceSOAPAction (String specVersion) throws SandeshaException { + if (Sandesha2Constants.SPEC_VERSIONS.WSRM.equals(specVersion)) + return Sandesha2Constants.SPEC_2005_02.Actions.SOAP_ACTION_TERMINATE_SEQUENCE; + else if (Sandesha2Constants.SPEC_VERSIONS.WSRX.equals(specVersion)) + return Sandesha2Constants.SPEC_2005_10.Actions.SOAP_ACTION_TERMINATE_SEQUENCE; + else + throw new SandeshaException (unknownSpecErrorMessage); + } + + public static String getTerminateSequenceResponseSOAPAction (String specVersion) throws SandeshaException { + if (Sandesha2Constants.SPEC_VERSIONS.WSRX.equals(specVersion)) + return Sandesha2Constants.SPEC_2005_10.Actions.SOAP_ACTION_TERMINATE_SEQUENCE_RESPONSE; + else + throw new SandeshaException (unknownSpecErrorMessage); + } + + public static String getAckRequestSOAPAction (String specVersion) throws SandeshaException { + if (Sandesha2Constants.SPEC_VERSIONS.WSRM.equals(specVersion)) + throw new SandeshaException ("this spec version does not define a ackRequest SOAP action"); + else if (Sandesha2Constants.SPEC_VERSIONS.WSRX.equals(specVersion)) + return Sandesha2Constants.SPEC_2005_10.Actions.SOAP_ACTION_ACK_REQUEST; + else + throw new SandeshaException (unknownSpecErrorMessage); + } + + public static String getSequenceAcknowledgementSOAPAction (String specVersion) throws SandeshaException { + if (Sandesha2Constants.SPEC_VERSIONS.WSRM.equals(specVersion)) + return Sandesha2Constants.SPEC_2005_02.Actions.SOAP_ACTION_SEQUENCE_ACKNOWLEDGEMENT; + else if (Sandesha2Constants.SPEC_VERSIONS.WSRX.equals(specVersion)) + return Sandesha2Constants.SPEC_2005_10.Actions.SOAP_ACTION_SEQUENCE_ACKNOWLEDGEMENT; + else + throw new SandeshaException (unknownSpecErrorMessage); + } + + public static boolean isTerminateSequenceResponseRequired (String specVersion) throws SandeshaException { + if (Sandesha2Constants.SPEC_VERSIONS.WSRM.equals(specVersion)) + return false; + else if (Sandesha2Constants.SPEC_VERSIONS.WSRX.equals(specVersion)) + return true; + else + throw new SandeshaException (unknownSpecErrorMessage); + } + + public static boolean isLastMessageIndicatorRequired (String specVersion) throws SandeshaException { + if (Sandesha2Constants.SPEC_VERSIONS.WSRM.equals(specVersion)) + return true; + else if (Sandesha2Constants.SPEC_VERSIONS.WSRX.equals(specVersion)) + return false; + else + throw new SandeshaException (unknownSpecErrorMessage); + } + + public static boolean isAckFinalAllowed (String specVersion) throws SandeshaException { + if (Sandesha2Constants.SPEC_VERSIONS.WSRM.equals(specVersion)) + return false; + else if (Sandesha2Constants.SPEC_VERSIONS.WSRX.equals(specVersion)) + return true; + else + throw new SandeshaException (unknownSpecErrorMessage); + } + + public static boolean isAckNoneAllowed (String specVersion) throws SandeshaException { + if (Sandesha2Constants.SPEC_VERSIONS.WSRM.equals(specVersion)) + return false; + else if (Sandesha2Constants.SPEC_VERSIONS.WSRX.equals(specVersion)) + return true; + else + throw new SandeshaException (unknownSpecErrorMessage); + } + + public static boolean isSequenceClosingAllowed (String specVersion) throws SandeshaException { + if (Sandesha2Constants.SPEC_VERSIONS.WSRM.equals(specVersion)) + return false; + else if (Sandesha2Constants.SPEC_VERSIONS.WSRX.equals(specVersion)) + return true; + else + throw new SandeshaException (unknownSpecErrorMessage); + } + + public static String getDefaultSpecVersion () { + return Sandesha2Constants.SPEC_VERSIONS.WSRM; + } + + public static String getAddressingAnonymousURI (String addressingNSURI) throws SandeshaException { + if (AddressingConstants.Submission.WSA_NAMESPACE.equals(addressingNSURI)) + return AddressingConstants.Submission.WSA_ANONYMOUS_URL; + else if (AddressingConstants.Final.WSA_NAMESPACE.equals(addressingNSURI)) + return AddressingConstants.Final.WSA_ANONYMOUS_URL; + else + throw new SandeshaException ("Unknown addressing version"); + } + + public static String getAddressingFaultAction (String addressingNSURI) throws SandeshaException { + if (AddressingConstants.Submission.WSA_NAMESPACE.equals(addressingNSURI)) + return "http://schemas.xmlsoap.org/ws/2004/08/addressing/fault";; //this is not available in addressing constants )-: + else if (AddressingConstants.Final.WSA_NAMESPACE.equals(addressingNSURI)) + return AddressingConstants.Final.WSA_FAULT_ACTION; + else + throw new SandeshaException ("Unknown addressing version"); + } + +} Added: webservices/sandesha/trunk/src/org/apache/sandesha2/util/TerminateManager.java URL: http://svn.apache.org/viewcvs/webservices/sandesha/trunk/src/org/apache/sandesha2/util/TerminateManager.java?rev=391959&view=auto ============================================================================== --- webservices/sandesha/trunk/src/org/apache/sandesha2/util/TerminateManager.java (added) +++ webservices/sandesha/trunk/src/org/apache/sandesha2/util/TerminateManager.java Thu Apr 6 03:50:20 2006 @@ -0,0 +1,449 @@ +/* + * Copyright 1999-2004 The Apache Software Foundation. + * + * Licensed 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.sandesha2.util; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.HashMap; +import java.util.Iterator; + +import org.apache.axis2.AxisFault; +import org.apache.axis2.addressing.EndpointReference; +import org.apache.axis2.context.ConfigurationContext; +import org.apache.axis2.context.MessageContext; +import org.apache.axis2.context.MessageContextConstants; +import org.apache.axis2.description.TransportOutDescription; +import org.apache.axis2.engine.AxisEngine; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.apache.sandesha2.RMMsgContext; +import org.apache.sandesha2.Sandesha2Constants; +import org.apache.sandesha2.SandeshaException; +import org.apache.sandesha2.Sandesha2Constants.SequenceProperties; +import org.apache.sandesha2.storage.StorageManager; +import org.apache.sandesha2.storage.Transaction; +import org.apache.sandesha2.storage.beanmanagers.CreateSeqBeanMgr; +import org.apache.sandesha2.storage.beanmanagers.InvokerBeanMgr; +import org.apache.sandesha2.storage.beanmanagers.NextMsgBeanMgr; +import org.apache.sandesha2.storage.beanmanagers.SenderBeanMgr; +import org.apache.sandesha2.storage.beanmanagers.SequencePropertyBeanMgr; +import org.apache.sandesha2.storage.beans.CreateSeqBean; +import org.apache.sandesha2.storage.beans.InvokerBean; +import org.apache.sandesha2.storage.beans.NextMsgBean; +import org.apache.sandesha2.storage.beans.SenderBean; +import org.apache.sandesha2.storage.beans.SequencePropertyBean; +import org.apache.sandesha2.transport.Sandesha2TransportOutDesc; + +/** + * Contains logic to remove all the storad data of a sequence. + * Methods of this are called by sending side and the receiving side when appropriate + * + * @author Chamikara Jayalath <[EMAIL PROTECTED]> + */ + +public class TerminateManager { + + private static Log log = LogFactory.getLog(TerminateManager.class); + + private static String CLEANED_ON_TERMINATE_MSG = "CleanedOnTerminateMsg"; + private static String CLEANED_AFTER_INVOCATION = "CleanedAfterInvocation"; + + public static HashMap receivingSideCleanMap = new HashMap (); + /** + * Called by the receiving side to remove data related to a sequence. + * e.g. After sending the TerminateSequence message. Calling this methods will complete all + * the data if InOrder invocation is not sequired. + * + * @param configContext + * @param sequenceID + * @throws SandeshaException + */ + public static void cleanReceivingSideOnTerminateMessage (ConfigurationContext configContext, String sequenceID) throws SandeshaException { + StorageManager storageManager = SandeshaUtil.getSandeshaStorageManager(configContext); + + //clean senderMap + + boolean inOrderInvocation = PropertyManager.getInstance().isInOrderInvocation(); + //SandeshaPropertyBean propertyBean = (SandeshaPropertyBean) msgContext.getParameter(Sandesha2Constants.SANDESHA2_POLICY_BEAN); + + + if(!inOrderInvocation) { + //there is no invoking by Sandesha2. So clean invocations storages. + cleanReceivingSideAfterInvocation(configContext,sequenceID); + } + + String cleanStatus = (String) receivingSideCleanMap.get(sequenceID); + if (cleanStatus!=null && CLEANED_AFTER_INVOCATION.equals(cleanStatus)) + completeTerminationOfReceivingSide(configContext,sequenceID); + else { + receivingSideCleanMap.put(sequenceID,CLEANED_ON_TERMINATE_MSG); + } + } + + /** + * When InOrder invocation is anabled this had to be called to clean the data left by the + * above method. This had to be called after the Invocation of the Last Message. + * + * @param configContext + * @param sequenceID + * @throws SandeshaException + */ + public static void cleanReceivingSideAfterInvocation (ConfigurationContext configContext, String sequenceID) throws SandeshaException { + StorageManager storageManager = SandeshaUtil.getSandeshaStorageManager(configContext); + InvokerBeanMgr storageMapBeanMgr = storageManager.getStorageMapBeanMgr(); + + //removing storageMap entries + InvokerBean findStorageMapBean = new InvokerBean (); + findStorageMapBean.setSequenceID(sequenceID); + findStorageMapBean.setInvoked(true); + Collection collection = storageMapBeanMgr.find(findStorageMapBean); + Iterator iterator = collection.iterator(); + while (iterator.hasNext()) { + InvokerBean storageMapBean = (InvokerBean) iterator.next(); + storageMapBeanMgr.delete(storageMapBean.getMessageContextRefKey()); + } + + String cleanStatus = (String) receivingSideCleanMap.get(sequenceID); + if (cleanStatus!=null && CLEANED_ON_TERMINATE_MSG.equals(cleanStatus)) + completeTerminationOfReceivingSide(configContext,sequenceID); + else { + receivingSideCleanMap.put(sequenceID,CLEANED_AFTER_INVOCATION); + } + } + + /** + * This has to be called by the lastly invocated one of the above two methods. + * + */ + private static void completeTerminationOfReceivingSide (ConfigurationContext configContext, String sequenceID) throws SandeshaException { + StorageManager storageManager = SandeshaUtil.getSandeshaStorageManager(configContext); + InvokerBeanMgr storageMapBeanMgr = storageManager.getStorageMapBeanMgr(); + NextMsgBeanMgr nextMsgBeanMgr = storageManager.getNextMsgBeanMgr(); + + //removing nextMsgMgr entries + NextMsgBean findNextMsgBean = new NextMsgBean (); + findNextMsgBean.setSequenceID(sequenceID); + Collection collection = nextMsgBeanMgr.find(findNextMsgBean); + Iterator iterator = collection.iterator(); + while (iterator.hasNext()) { + NextMsgBean nextMsgBean = (NextMsgBean) iterator.next(); + //nextMsgBeanMgr.delete(nextMsgBean.getSequenceID()); + } + + removeReceivingSideProperties(configContext,sequenceID); + } + + private static void removeReceivingSideProperties (ConfigurationContext configContext, String sequenceID) throws SandeshaException { + StorageManager storageManager = SandeshaUtil.getSandeshaStorageManager(configContext); + SequencePropertyBeanMgr sequencePropertyBeanMgr = storageManager.getSequencePropretyBeanMgr(); + SequencePropertyBean allSequenceBean = sequencePropertyBeanMgr.retrieve(Sandesha2Constants.SequenceProperties.ALL_SEQUENCES,Sandesha2Constants.SequenceProperties.INCOMING_SEQUENCE_LIST); + + if (allSequenceBean!=null) { + log.debug("AllSequence bean is null"); + + ArrayList allSequenceList = SandeshaUtil.getArrayListFromString(allSequenceBean.getValue()); + allSequenceList.remove(sequenceID); + + //updating + allSequenceBean.setValue(allSequenceList.toString()); + sequencePropertyBeanMgr.update(allSequenceBean); + } + } + + private static boolean isRequiredForResponseSide (String name) { + if (name==null && name.equals(Sandesha2Constants.SequenceProperties.LAST_OUT_MESSAGE_NO)) + return false; + + if (name.equals(Sandesha2Constants.SequenceProperties.LAST_OUT_MESSAGE_NO)) + return false; + + return false; + } + + + /** + * This is called by the sending side to clean data related to a sequence. + * e.g. after sending the TerminateSequence message. + * + * @param configContext + * @param sequenceID + * @throws SandeshaException + */ + public static void terminateSendingSide (ConfigurationContext configContext, String sequenceID,boolean serverSide) throws SandeshaException { + + StorageManager storageManager = SandeshaUtil.getSandeshaStorageManager(configContext); + + SequencePropertyBeanMgr seqPropMgr = storageManager.getSequencePropretyBeanMgr(); + SequencePropertyBean seqTerminatedBean = new SequencePropertyBean (sequenceID,Sandesha2Constants.SequenceProperties.SEQUENCE_TERMINATED,Sandesha2Constants.VALUE_TRUE); + seqPropMgr.insert(seqTerminatedBean); + + cleanSendingSideData(configContext,sequenceID,serverSide); + + + } + + + + private static void doUpdatesIfNeeded (String sequenceID, SequencePropertyBean propertyBean, SequencePropertyBeanMgr seqPropMgr) throws SandeshaException { + if (propertyBean.getName().equals(Sandesha2Constants.SequenceProperties.CLIENT_COMPLETED_MESSAGES)) { + + //this value cannot be completely deleted since this data will be needed by SequenceReports + //so saving it with the sequenceID value being the out sequenceID. + + SequencePropertyBean newBean = new SequencePropertyBean (); + newBean.setSequenceID(sequenceID); + newBean.setName(propertyBean.getName()); + newBean.setValue(propertyBean.getValue()); + + seqPropMgr.insert(newBean); + //TODO amazingly this property does not seem to get deleted without following - in the hibernate impl + //(even though the lines efter current methodcall do this). + seqPropMgr.delete (propertyBean.getSequenceID(),propertyBean.getName()); + } + } + + private static boolean isProportyDeletable (String name) { + boolean deleatable = true; + + if (Sandesha2Constants.SequenceProperties.TERMINATE_ADDED.equals(name)) + deleatable = false; + + if (Sandesha2Constants.SequenceProperties.NO_OF_OUTGOING_MSGS_ACKED.equals(name)) + deleatable = false; + + if (Sandesha2Constants.SequenceProperties.INTERNAL_SEQUENCE_ID.equals(name)) + deleatable = false; + + if (Sandesha2Constants.SequenceProperties.RM_SPEC_VERSION.equals(name)) + deleatable = false; + + if (Sandesha2Constants.SequenceProperties.SEQUENCE_TERMINATED.equals(name)) + deleatable = false; + + if (Sandesha2Constants.SequenceProperties.SEQUENCE_CLOSED.equals(name)) + deleatable = false; + + if (Sandesha2Constants.SequenceProperties.SEQUENCE_TIMED_OUT.equals(name)) + deleatable = false; + + return deleatable; + } + + public static void timeOutSendingSideSequence (ConfigurationContext context,String sequenceID, boolean serverside) throws SandeshaException { + StorageManager storageManager = SandeshaUtil.getSandeshaStorageManager(context); + + SequencePropertyBeanMgr seqPropMgr = storageManager.getSequencePropretyBeanMgr(); + SequencePropertyBean seqTerminatedBean = new SequencePropertyBean (sequenceID,Sandesha2Constants.SequenceProperties.SEQUENCE_TIMED_OUT,Sandesha2Constants.VALUE_TRUE); + seqPropMgr.insert(seqTerminatedBean); + + + cleanSendingSideData(context,sequenceID,serverside); + } + + private static void cleanSendingSideData (ConfigurationContext configContext,String sequenceID, boolean serverSide) throws SandeshaException { + StorageManager storageManager = SandeshaUtil.getSandeshaStorageManager(configContext); + + SequencePropertyBeanMgr sequencePropertyBeanMgr = storageManager.getSequencePropretyBeanMgr(); + SenderBeanMgr retransmitterBeanMgr = storageManager.getRetransmitterBeanMgr(); + CreateSeqBeanMgr createSeqBeanMgr = storageManager.getCreateSeqBeanMgr(); + + SequencePropertyBean sequenceTerminatedBean = new SequencePropertyBean (sequenceID,Sandesha2Constants.SequenceProperties.SEQUENCE_TERMINATED,Sandesha2Constants.VALUE_TRUE); + sequencePropertyBeanMgr.insert(sequenceTerminatedBean); + + if (!serverSide) { + //stpoing the listner for the client side. + + //SequencePropertyBean outGoingAcksToBean = sequencePropertyBeanMgr.retrieve(sequenceID,Sandesha2Constants.SequenceProperties.OUT_SEQ_ACKSTO); + + boolean stopListnerForAsyncAcks = false; + SequencePropertyBean internalSequenceBean = sequencePropertyBeanMgr.retrieve(sequenceID,Sandesha2Constants.SequenceProperties.INTERNAL_SEQUENCE_ID); + if (internalSequenceBean!=null) { + String internalSequenceID = internalSequenceBean.getValue(); + SequencePropertyBean acksToBean = sequencePropertyBeanMgr.retrieve(internalSequenceID,Sandesha2Constants.SequenceProperties.ACKS_TO_EPR); + + String addressingNamespace = SandeshaUtil.getSequenceProperty(internalSequenceID,Sandesha2Constants.SequenceProperties.ADDRESSING_NAMESPACE_VALUE,configContext); + String anonymousURI = SpecSpecificConstants.getAddressingAnonymousURI(addressingNamespace); + + if (acksToBean!=null) { + String acksTo = acksToBean.getValue(); + if (acksTo!=null && !anonymousURI.equals(acksTo)) { + stopListnerForAsyncAcks = true; + } + } + } + } + + SequencePropertyBean internalSequenceBean = sequencePropertyBeanMgr.retrieve(sequenceID,Sandesha2Constants.SequenceProperties.INTERNAL_SEQUENCE_ID); + if (internalSequenceBean==null) + throw new SandeshaException ("TempSequence entry not found"); + + String internalSequenceId = (String) internalSequenceBean.getValue(); + + //removing retransmitterMgr entries + //SenderBean findRetransmitterBean = new SenderBean (); + //findRetransmitterBean.setInternalSequenceID(internalSequenceId); + Collection collection = retransmitterBeanMgr.find(internalSequenceId); + Iterator iterator = collection.iterator(); + while (iterator.hasNext()) { + SenderBean retransmitterBean = (SenderBean) iterator.next(); + retransmitterBeanMgr.delete(retransmitterBean.getMessageID()); + } + + //removing the createSeqMgrEntry + CreateSeqBean findCreateSequenceBean = new CreateSeqBean (); + findCreateSequenceBean.setInternalSequenceID(internalSequenceId); + collection = createSeqBeanMgr.find(findCreateSequenceBean); + iterator = collection.iterator(); + while (iterator.hasNext()) { + CreateSeqBean createSeqBean = (CreateSeqBean) iterator.next(); + createSeqBeanMgr.delete(createSeqBean.getCreateSeqMsgID()); + } + + //removing sequence properties + SequencePropertyBean findSequencePropertyBean1 = new SequencePropertyBean (); + findSequencePropertyBean1.setSequenceID(internalSequenceId); + collection = sequencePropertyBeanMgr.find(findSequencePropertyBean1); + iterator = collection.iterator(); + while (iterator.hasNext()) { + SequencePropertyBean sequencePropertyBean = (SequencePropertyBean) iterator.next(); + doUpdatesIfNeeded (sequenceID,sequencePropertyBean,sequencePropertyBeanMgr); + + if (isProportyDeletable(sequencePropertyBean.getName())) { + sequencePropertyBeanMgr.delete(sequencePropertyBean.getSequenceID(),sequencePropertyBean.getName()); + } + } + + SandeshaUtil.stopSenderForTheSequence(internalSequenceId); + } + + public static void addTerminateSequenceMessage(RMMsgContext referenceMessage, + String outSequenceId, String internalSequenceId) + throws SandeshaException { + + + ConfigurationContext configurationContext = referenceMessage.getMessageContext().getConfigurationContext(); + StorageManager storageManager = SandeshaUtil + .getSandeshaStorageManager(configurationContext); + + Transaction addTerminateSeqTransaction = storageManager.getTransaction(); + + SequencePropertyBeanMgr seqPropMgr = storageManager + .getSequencePropretyBeanMgr(); + + SequencePropertyBean terminated = seqPropMgr.retrieve(outSequenceId, + Sandesha2Constants.SequenceProperties.TERMINATE_ADDED); + + if (terminated != null && terminated.getValue() != null + && "true".equals(terminated.getValue())) { + String message = "Terminate was added previously."; + log.info(message); +// return; + } + + RMMsgContext terminateRMMessage = RMMsgCreator + .createTerminateSequenceMessage(referenceMessage, outSequenceId,internalSequenceId); + terminateRMMessage.setFlow(MessageContext.OUT_FLOW); + terminateRMMessage.setProperty(Sandesha2Constants.APPLICATION_PROCESSING_DONE,"true"); + + SequencePropertyBean toBean = seqPropMgr.retrieve(internalSequenceId, + Sandesha2Constants.SequenceProperties.TO_EPR); + + EndpointReference toEPR = new EndpointReference ( toBean.getValue()); + if (toEPR == null) { + String message = "To EPR has an invalid value"; + throw new SandeshaException(message); + } + + terminateRMMessage.setTo(new EndpointReference(toEPR.getAddress())); + + String addressingNamespaceURI = SandeshaUtil.getSequenceProperty(internalSequenceId,Sandesha2Constants.SequenceProperties.ADDRESSING_NAMESPACE_VALUE,configurationContext); + String anonymousURI = SpecSpecificConstants.getAddressingAnonymousURI(addressingNamespaceURI); + + String rmVersion = SandeshaUtil.getRMVersion(internalSequenceId,configurationContext); + if (rmVersion==null) + throw new SandeshaException ("Cant find the rmVersion of the given message"); + terminateRMMessage.setWSAAction(SpecSpecificConstants.getTerminateSequenceAction(rmVersion)); + terminateRMMessage.setSOAPAction(SpecSpecificConstants.getTerminateSequenceSOAPAction(rmVersion)); + + SequencePropertyBean transportToBean = seqPropMgr.retrieve(internalSequenceId,Sandesha2Constants.SequenceProperties.TRANSPORT_TO); + if (transportToBean!=null) { + terminateRMMessage.setProperty(MessageContextConstants.TRANSPORT_URL,transportToBean.getValue()); + } + + try { + terminateRMMessage.addSOAPEnvelope(); + } catch (AxisFault e) { + throw new SandeshaException(e.getMessage()); + } + + String key = SandeshaUtil.getUUID(); + + SenderBean terminateBean = new SenderBean(); + terminateBean.setMessageContextRefKey(key); + + + storageManager.storeMessageContext(key,terminateRMMessage.getMessageContext()); + + + //Set a retransmitter lastSentTime so that terminate will be send with + // some delay. + //Otherwise this get send before return of the current request (ack). + //TODO: refine the terminate delay. + terminateBean.setTimeToSend(System.currentTimeMillis() + + Sandesha2Constants.TERMINATE_DELAY); + + terminateBean.setMessageID(terminateRMMessage.getMessageId()); + + //this will be set to true at the sender. + terminateBean.setSend(true); + + terminateRMMessage.getMessageContext().setProperty(Sandesha2Constants.QUALIFIED_FOR_SENDING, + Sandesha2Constants.VALUE_FALSE); + + terminateBean.setReSend(false); + + SenderBeanMgr retramsmitterMgr = storageManager + .getRetransmitterBeanMgr(); + + retramsmitterMgr.insert(terminateBean); + + SequencePropertyBean terminateAdded = new SequencePropertyBean(); + terminateAdded.setName(Sandesha2Constants.SequenceProperties.TERMINATE_ADDED); + terminateAdded.setSequenceID(outSequenceId); + terminateAdded.setValue("true"); + + seqPropMgr.insert(terminateAdded); + + //This should be dumped to the storage by the sender + TransportOutDescription transportOut = terminateRMMessage.getMessageContext().getTransportOut(); + terminateRMMessage.setProperty(Sandesha2Constants.ORIGINAL_TRANSPORT_OUT_DESC,transportOut); + terminateRMMessage.setProperty(Sandesha2Constants.MESSAGE_STORE_KEY,key); + terminateRMMessage.setProperty(Sandesha2Constants.SET_SEND_TO_TRUE,Sandesha2Constants.VALUE_TRUE); + terminateRMMessage.getMessageContext().setTransportOut(new Sandesha2TransportOutDesc ()); + addTerminateSeqTransaction.commit(); + + AxisEngine engine = new AxisEngine (configurationContext); + try { + engine.send(terminateRMMessage.getMessageContext()); + } catch (AxisFault e) { + throw new SandeshaException (e.getMessage()); + } + + } + +} Modified: webservices/sandesha/trunk/src/org/apache/sandesha2/workers/InOrderInvoker.java URL: http://svn.apache.org/viewcvs/webservices/sandesha/trunk/src/org/apache/sandesha2/workers/InOrderInvoker.java?rev=391959&r1=391958&r2=391959&view=diff ============================================================================== --- webservices/sandesha/trunk/src/org/apache/sandesha2/workers/InOrderInvoker.java (original) +++ webservices/sandesha/trunk/src/org/apache/sandesha2/workers/InOrderInvoker.java Thu Apr 6 03:50:20 2006 @@ -30,7 +30,6 @@ import org.apache.sandesha2.RMMsgContext; import org.apache.sandesha2.Sandesha2Constants; import org.apache.sandesha2.SandeshaException; -import org.apache.sandesha2.TerminateManager; import org.apache.sandesha2.storage.StorageManager; import org.apache.sandesha2.storage.Transaction; import org.apache.sandesha2.storage.beanmanagers.InvokerBeanMgr; @@ -41,6 +40,7 @@ import org.apache.sandesha2.storage.beans.SequencePropertyBean; import org.apache.sandesha2.util.MsgInitializer; import org.apache.sandesha2.util.SandeshaUtil; +import org.apache.sandesha2.util.TerminateManager; import org.apache.sandesha2.wsrm.Sequence; /** @@ -79,9 +79,9 @@ workingSequences.add(sequenceID); if (!isInvokerStarted()) { + this.context = context; runInvoker = true; //so that isSenderStarted()=true. super.start(); - this.context = context; } } @@ -188,11 +188,7 @@ if (!AxisOperationFactory.MEP_URI_IN_ONLY.equals(msgToInvoke.getAxisOperation().getMessageExchangePattern())) { invocationTransaction = storageManager.getTransaction(); - } - - log.info("Invoker invoking a '" + SandeshaUtil.getMessageTypeString(rmMsg - .getMessageType()) + "' message."); - + } storageMapMgr.delete(key); } catch (AxisFault e) { Modified: webservices/sandesha/trunk/src/org/apache/sandesha2/workers/Sender.java URL: http://svn.apache.org/viewcvs/webservices/sandesha/trunk/src/org/apache/sandesha2/workers/Sender.java?rev=391959&r1=391958&r2=391959&view=diff ============================================================================== --- webservices/sandesha/trunk/src/org/apache/sandesha2/workers/Sender.java (original) +++ webservices/sandesha/trunk/src/org/apache/sandesha2/workers/Sender.java Thu Apr 6 03:50:20 2006 @@ -37,20 +37,20 @@ import org.apache.axis2.util.threadpool.ThreadPool; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; -import org.apache.sandesha2.AcknowledgementManager; import org.apache.sandesha2.RMMsgContext; import org.apache.sandesha2.Sandesha2Constants; import org.apache.sandesha2.SandeshaException; -import org.apache.sandesha2.TerminateManager; import org.apache.sandesha2.storage.StorageManager; import org.apache.sandesha2.storage.Transaction; import org.apache.sandesha2.storage.beanmanagers.SenderBeanMgr; import org.apache.sandesha2.storage.beans.SenderBean; +import org.apache.sandesha2.util.AcknowledgementManager; import org.apache.sandesha2.util.MessageRetransmissionAdjuster; import org.apache.sandesha2.util.MsgInitializer; import org.apache.sandesha2.util.PropertyManager; import org.apache.sandesha2.util.SandeshaUtil; import org.apache.sandesha2.util.SequenceManager; +import org.apache.sandesha2.util.TerminateManager; import org.apache.sandesha2.wsrm.Sequence; import org.apache.sandesha2.wsrm.TerminateSequence; @@ -122,6 +122,7 @@ continue; } + MessageRetransmissionAdjuster retransmitterAdjuster = new MessageRetransmissionAdjuster(); retransmitterAdjuster.adjustRetransmittion(senderBean, context); @@ -161,8 +162,7 @@ } updateMessage(msgCtx); - log.info("Sender is sending a '" + SandeshaUtil.getMessageTypeString(rmMsgCtx.getMessageType()) + "' message."); - + Transaction preSendTransaction = storageManager.getTransaction(); int messageType = rmMsgCtx.getMessageType(); @@ -258,9 +258,9 @@ workingSequences.add(sequenceID); if (!isSenderStarted()) { + this.context = context; runSender = true; // so that isSenderStarted()=true. super.start(); - this.context = context; } } Modified: webservices/sandesha/trunk/src/org/apache/sandesha2/wsrm/RMElements.java URL: http://svn.apache.org/viewcvs/webservices/sandesha/trunk/src/org/apache/sandesha2/wsrm/RMElements.java?rev=391959&r1=391958&r2=391959&view=diff ============================================================================== --- webservices/sandesha/trunk/src/org/apache/sandesha2/wsrm/RMElements.java (original) +++ webservices/sandesha/trunk/src/org/apache/sandesha2/wsrm/RMElements.java Thu Apr 6 03:50:20 2006 @@ -20,9 +20,6 @@ import javax.xml.namespace.QName; -import org.apache.sandesha2.Sandesha2Constants; -import org.apache.sandesha2.SandeshaException; -import org.apache.sandesha2.util.SOAPAbstractFactory; import org.apache.axiom.om.OMElement; import org.apache.axiom.om.OMException; import org.apache.axiom.soap.SOAP11Constants; @@ -30,8 +27,9 @@ import org.apache.axiom.soap.SOAPFactory; import org.apache.axiom.soap.SOAPHeader; import org.apache.axis2.addressing.AddressingConstants; - -import sun.util.logging.resources.logging; +import org.apache.sandesha2.Sandesha2Constants; +import org.apache.sandesha2.SandeshaException; +import org.apache.sandesha2.util.SOAPAbstractFactory; /** * @author Chamikara Jayalath <[EMAIL PROTECTED]> Modified: webservices/sandesha/trunk/src/org/apache/sandesha2/wsrm/SequenceAcknowledgement.java URL: http://svn.apache.org/viewcvs/webservices/sandesha/trunk/src/org/apache/sandesha2/wsrm/SequenceAcknowledgement.java?rev=391959&r1=391958&r2=391959&view=diff ============================================================================== --- webservices/sandesha/trunk/src/org/apache/sandesha2/wsrm/SequenceAcknowledgement.java (original) +++ webservices/sandesha/trunk/src/org/apache/sandesha2/wsrm/SequenceAcknowledgement.java Thu Apr 6 03:50:20 2006 @@ -31,7 +31,7 @@ import org.apache.axiom.soap.SOAPHeaderBlock; import org.apache.sandesha2.Sandesha2Constants; import org.apache.sandesha2.SandeshaException; -import org.apache.sandesha2.SpecSpecificConstants; +import org.apache.sandesha2.util.SpecSpecificConstants; /** * Adds the SequenceAcknowledgement header block. --------------------------------------------------------------------- To unsubscribe, e-mail: [EMAIL PROTECTED] For additional commands, e-mail: [EMAIL PROTECTED]
