You are viewing a plain text version of this content. The canonical link for it is here.
Posted to sandesha-dev@ws.apache.org by mc...@apache.org on 2007/12/12 17:12:56 UTC
svn commit: r603654 - in
/webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/msgprocessors:
CloseSequenceProcessor.java TerminateSeqMsgProcessor.java
Author: mckierna
Date: Wed Dec 12 08:12:56 2007
New Revision: 603654
URL: http://svn.apache.org/viewvc?rev=603654&view=rev
Log:
Ensure RMD can close sequences
Modified:
webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/msgprocessors/CloseSequenceProcessor.java
webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/msgprocessors/TerminateSeqMsgProcessor.java
Modified: webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/msgprocessors/CloseSequenceProcessor.java
URL: http://svn.apache.org/viewvc/webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/msgprocessors/CloseSequenceProcessor.java?rev=603654&r1=603653&r2=603654&view=diff
==============================================================================
--- webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/msgprocessors/CloseSequenceProcessor.java (original)
+++ webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/msgprocessors/CloseSequenceProcessor.java Wed Dec 12 08:12:56 2007
@@ -38,6 +38,8 @@
import org.apache.sandesha2.storage.StorageManager;
import org.apache.sandesha2.storage.Transaction;
import org.apache.sandesha2.storage.beans.RMDBean;
+import org.apache.sandesha2.storage.beans.RMSBean;
+import org.apache.sandesha2.storage.beans.RMSequenceBean;
import org.apache.sandesha2.util.AcknowledgementManager;
import org.apache.sandesha2.util.FaultManager;
import org.apache.sandesha2.util.RMMsgCreator;
@@ -71,10 +73,13 @@
StorageManager storageManager = SandeshaUtil.getSandeshaStorageManager(configCtx, configCtx
.getAxisConfiguration());
- RMDBean rmdBean = SandeshaUtil.getRMDBeanFromSequenceId(storageManager, sequenceId);
+ RMSequenceBean rmBean = SandeshaUtil.getRMDBeanFromSequenceId(storageManager, sequenceId);
+ if(rmBean==null){
+ rmBean = SandeshaUtil.getRMSBeanFromSequenceId(storageManager, sequenceId);
+ }
//check the security credentials
- SandeshaUtil.assertProofOfPossession(rmdBean, msgCtx, msgCtx.getEnvelope().getBody());
+ SandeshaUtil.assertProofOfPossession(rmBean, msgCtx, msgCtx.getEnvelope().getBody());
if (FaultManager.checkForUnknownSequence(rmMsgCtx, sequenceId, storageManager, false)) {
if (log.isDebugEnabled())
@@ -83,32 +88,35 @@
}
// throwing a fault if the sequence is terminated
- if (FaultManager.checkForSequenceTerminated(rmMsgCtx, sequenceId, rmdBean, false)) {
+ if (FaultManager.checkForSequenceTerminated(rmMsgCtx, sequenceId, rmBean, false)) {
if (log.isDebugEnabled())
log.debug("Exit: CloseSequenceProcessor::processInMessage, Sequence terminated");
return false;
}
- rmdBean.setClosed(true);
- storageManager.getRMDBeanMgr().update(rmdBean);
-
- RMMsgContext ackRMMsgCtx = AcknowledgementManager.generateAckMessage(rmMsgCtx, rmdBean, sequenceId, storageManager, true);
- // adding the ack part(s) to the envelope.
- Iterator sequenceAckIter = ackRMMsgCtx.getSequenceAcknowledgements();
+ rmBean.setClosed(true);
+ Iterator sequenceAckIter = null;
+ if(rmBean instanceof RMDBean){
+ storageManager.getRMDBeanMgr().update((RMDBean)rmBean);
+ RMMsgContext ackRMMsgCtx = AcknowledgementManager.generateAckMessage(rmMsgCtx, (RMDBean)rmBean, sequenceId, storageManager, true);
+ // adding the ack part(s) to the envelope.
+ sequenceAckIter = ackRMMsgCtx.getSequenceAcknowledgements();
+ }
+ else{
+ storageManager.getRMSBeanMgr().update((RMSBean)rmBean);
+ }
- RMMsgContext closeSeqResponseRMMsg = RMMsgCreator.createCloseSeqResponseMsg(rmMsgCtx, rmdBean);
+ RMMsgContext closeSeqResponseRMMsg = RMMsgCreator.createCloseSeqResponseMsg(rmMsgCtx, rmBean);
MessageContext closeSequenceResponseMsg = closeSeqResponseRMMsg.getMessageContext();
- while (sequenceAckIter.hasNext()) {
+ while (sequenceAckIter!=null && sequenceAckIter.hasNext()) {
SequenceAcknowledgement sequenceAcknowledgement = (SequenceAcknowledgement) sequenceAckIter.next();
closeSeqResponseRMMsg.addSequenceAcknowledgement(sequenceAcknowledgement);
}
closeSeqResponseRMMsg.setFlow(MessageContext.OUT_FLOW);
closeSeqResponseRMMsg.setProperty(Sandesha2Constants.APPLICATION_PROCESSING_DONE, "true");
-
closeSequenceResponseMsg.setResponseWritten(true);
-
closeSeqResponseRMMsg.addSOAPEnvelope();
//
Modified: webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/msgprocessors/TerminateSeqMsgProcessor.java
URL: http://svn.apache.org/viewvc/webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/msgprocessors/TerminateSeqMsgProcessor.java?rev=603654&r1=603653&r2=603654&view=diff
==============================================================================
--- webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/msgprocessors/TerminateSeqMsgProcessor.java (original)
+++ webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/msgprocessors/TerminateSeqMsgProcessor.java Wed Dec 12 08:12:56 2007
@@ -43,6 +43,7 @@
import org.apache.sandesha2.storage.beanmanagers.RMDBeanMgr;
import org.apache.sandesha2.storage.beans.RMDBean;
import org.apache.sandesha2.storage.beans.RMSBean;
+import org.apache.sandesha2.storage.beans.RMSequenceBean;
import org.apache.sandesha2.storage.beans.SenderBean;
import org.apache.sandesha2.util.AcknowledgementManager;
import org.apache.sandesha2.util.FaultManager;
@@ -91,10 +92,13 @@
StorageManager storageManager = SandeshaUtil.getSandeshaStorageManager(context,context.getAxisConfiguration());
// Check that the sender of this TerminateSequence holds the correct token
- RMDBean rmdBean = SandeshaUtil.getRMDBeanFromSequenceId(storageManager, sequenceId);
+ RMSequenceBean rmBean = SandeshaUtil.getRMDBeanFromSequenceId(storageManager, sequenceId);
+ if(rmBean==null){
+ rmBean = SandeshaUtil.getRMSBeanFromSequenceId(storageManager, sequenceId);
+ }
//check security credentials
- SandeshaUtil.assertProofOfPossession(rmdBean, terminateSeqMsg,
+ SandeshaUtil.assertProofOfPossession(rmBean, terminateSeqMsg,
terminateSeqMsg.getEnvelope().getBody());
if (FaultManager.checkForUnknownSequence(terminateSeqRMMsg, sequenceId, storageManager, false)) {
@@ -106,24 +110,21 @@
// add the terminate sequence response if required.
RMMsgContext terminateSequenceResponse = null;
if (SpecSpecificConstants.isTerminateSequenceResponseRequired(terminateSeqRMMsg.getRMSpecVersion()))
- terminateSequenceResponse = getTerminateSequenceResponse(terminateSeqRMMsg, rmdBean, sequenceId, storageManager);
+ terminateSequenceResponse = getTerminateSequenceResponse(terminateSeqRMMsg, rmBean, sequenceId, storageManager);
setUpHighestMsgNumbers(context, storageManager, sequenceId, terminateSeqRMMsg);
-
-
boolean inOrderInvocation = SandeshaUtil.getDefaultPropertyBean(context.getAxisConfiguration()).isInOrder();
-
- //if the invocation is inOrder and if this is RM 1.1 there is a posibility of all the messages having eleady being invoked.
+ //if the invocation is inOrder and if this is RM 1.1 there is a posibility of all the messages having aleady being invoked.
//In this case we should do the full termination.
boolean doFullTermination = false;
- if (inOrderInvocation) {
+ if (inOrderInvocation && rmBean instanceof RMDBean) {
- long highestMsgNo = rmdBean.getHighestInMessageNumber();
- long nextMsgToProcess = rmdBean.getNextMsgNoToProcess();
+ long highestMsgNo = ((RMDBean)rmBean).getHighestInMessageNumber();
+ long nextMsgToProcess = ((RMDBean)rmBean).getNextMsgNoToProcess();
if (nextMsgToProcess>highestMsgNo) {
//all the messages have been invoked, u can do the full termination
@@ -140,10 +141,14 @@
} else
TerminateManager.cleanReceivingSideOnTerminateMessage(context, sequenceId, storageManager);
- rmdBean.setTerminated(true);
- rmdBean.setLastActivatedTime(System.currentTimeMillis());
- storageManager.getRMDBeanMgr().update(rmdBean);
-
+ rmBean.setTerminated(true);
+ rmBean.setLastActivatedTime(System.currentTimeMillis());
+ if(rmBean instanceof RMDBean){
+ storageManager.getRMDBeanMgr().update((RMDBean)rmBean);
+ }
+ else{
+ storageManager.getRMSBeanMgr().update((RMSBean)rmBean);
+ }
//sending the terminate sequence response
if (terminateSequenceResponse != null) {
@@ -313,25 +318,26 @@
log.debug("Exit: TerminateSeqMsgProcessor::setUpHighestMsgNumbers");
}
- private RMMsgContext getTerminateSequenceResponse(RMMsgContext terminateSeqRMMsg, RMDBean rmdBean, String sequenceId,
+ private RMMsgContext getTerminateSequenceResponse(RMMsgContext terminateSeqRMMsg, RMSequenceBean rmBean, String sequenceId,
StorageManager storageManager) throws AxisFault {
if (log.isDebugEnabled())
log.debug("Enter: TerminateSeqMsgProcessor::addTerminateSequenceResponse, " + sequenceId);
- RMMsgContext terminateSeqResponseRMMsg = RMMsgCreator.createTerminateSeqResponseMsg(terminateSeqRMMsg, rmdBean);
+ RMMsgContext terminateSeqResponseRMMsg = RMMsgCreator.createTerminateSeqResponseMsg(terminateSeqRMMsg, rmBean);
MessageContext outMessage = terminateSeqResponseRMMsg.getMessageContext();
-
- RMMsgContext ackRMMessage = AcknowledgementManager.generateAckMessage(terminateSeqRMMsg, rmdBean,
- sequenceId, storageManager, true);
-
- // copy over the ack parts
- Iterator iter = ackRMMessage.getSequenceAcknowledgements();
- while (iter.hasNext()) {
- SequenceAcknowledgement seqAck = (SequenceAcknowledgement) iter.next();
- terminateSeqResponseRMMsg.addSequenceAcknowledgement(seqAck);
- }
+ if(rmBean instanceof RMDBean){
+ RMMsgContext ackRMMessage = AcknowledgementManager.generateAckMessage(terminateSeqRMMsg, (RMDBean)rmBean,
+ sequenceId, storageManager, true);
+
+ // copy over the ack parts
+ Iterator iter = ackRMMessage.getSequenceAcknowledgements();
+ while (iter.hasNext()) {
+ SequenceAcknowledgement seqAck = (SequenceAcknowledgement) iter.next();
+ terminateSeqResponseRMMsg.addSequenceAcknowledgement(seqAck);
+ }
+ }
terminateSeqResponseRMMsg.addSOAPEnvelope();
terminateSeqResponseRMMsg.setFlow(MessageContext.OUT_FLOW);
---------------------------------------------------------------------
To unsubscribe, e-mail: sandesha-dev-unsubscribe@ws.apache.org
For additional commands, e-mail: sandesha-dev-help@ws.apache.org