svn commit: r700429 - in /webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/workers: Sender.java SenderWorker.java

Author: gatfora
Date: Tue Sep 30 05:11:10 2008
New Revision: 700429

URL: http://svn.apache.org/viewvc?rev=700429&view=rev
Log:
If we fail to send a terminate sequence because we never got a MakeConnection message for the response sequence, terminate the sequence and clean up

Modified:
    webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/workers/Sender.java
    webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/workers/SenderWorker.java

Modified: webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/workers/Sender.java
URL: http://svn.apache.org/viewvc/webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/workers/Sender.java?rev=700429&r1=700428&r2=700429&view=diff
==============================================================================
--- webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/workers/Sender.java (original)
+++ webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/workers/Sender.java Tue Sep 30 05:11:10 2008
@@ -51,6 +51,7 @@
 import org.apache.sandesha2.util.MsgInitializer;
 import org.apache.sandesha2.util.SandeshaUtil;
 import org.apache.sandesha2.util.SequenceManager;
+import org.apache.sandesha2.util.TerminateManager;
 
 /**
  * This is responsible for sending and re-sending messages of Sandesha2. This
@@ -574,6 +575,13 @@
 			finder.setTimeToSend(System.currentTimeMillis() - Sandesha2Constants.TRANSPORT_WAIT_TIME);
 
 			List beans = manager.getSenderBeanMgr().find(finder);
+			
+			//Commit this transaction
+			tran.commit();
+			
+			// Create a new transaction
+			tran = manager.getTransaction();
+
 			Iterator beanIter = beans.iterator();
 			while (beanIter.hasNext()) {
 				SenderBean bean = (SenderBean) beanIter.next();
@@ -605,9 +613,14 @@
 				int messageType = bean.getMessageType();
 				if(MessageTypes.TERMINATE_SEQ ==  messageType || MessageTypes.TERMINATE_SEQ_RESPONSE ==  messageType){
 					String id = bean.getSequenceID(); // get this again as it is an error case
-					if(log.isDebugEnabled()) log.debug("Sender::checkForOrphanMessages.  Orphaned message of type TERMINATE_SEQ or TERMINATE_SEQ_RESPONSE found.  Deleting this message with a sequence ID of : " + id);
-					manager.getSenderBeanMgr().delete(id);
 					
+					// Mark the sequence as terminated
+					RMSBean rmsBean = SandeshaUtil.getRMSBeanFromSequenceId(manager, id);
+					TerminateManager.terminateSendingSide(rmsBean, storageManager, false);
+					
+					if(log.isDebugEnabled()) log.debug("Sender::checkForOrphanMessages.  Orphaned message of type TERMINATE_SEQ or TERMINATE_SEQ_RESPONSE found.  Deleting this message with a sequence ID of : " + id);
+					// Delete the terminate sender bean.
+					manager.getSenderBeanMgr().delete(bean.getMessageID());		
 					
 				} else {					
 					// Update the bean so that we won't emit another message for another TRANSPORT_WAIT_TIME

Modified: webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/workers/SenderWorker.java
URL: http://svn.apache.org/viewvc/webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/workers/SenderWorker.java?rev=700429&r1=700428&r2=700429&view=diff
==============================================================================
--- webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/workers/SenderWorker.java (original)
+++ webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/workers/SenderWorker.java Tue Sep 30 05:11:10 2008
@@ -23,7 +23,6 @@
 import java.util.HashMap;
 import java.util.Iterator;
 
-import org.apache.axiom.om.impl.builder.StAXBuilder;
 import org.apache.axiom.soap.SOAPEnvelope;
 import org.apache.axis2.AxisFault;
 import org.apache.axis2.Constants;
@@ -231,12 +230,6 @@
 				transaction.commit();			
 			
 			transaction = storageManager.getTransaction();
-
-			
-			if (transaction != null && transaction.isActive()) 
-				transaction.commit();			
-			
-			transaction = storageManager.getTransaction();
 			
 			senderBean = updateMessage(rmMsgCtx,senderBean,storageManager);