Author: amilas
Date: Tue Oct 28 23:35:47 2008
New Revision: 708804
URL: http://svn.apache.org/viewvc?rev=708804&view=rev
Log:
apply the patch for SANDESHA2-179
Modified:
webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/msgprocessors/AcknowledgementProcessor.java
webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/workers/WorkerLock.java
Modified: webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/msgprocessors/AcknowledgementProcessor.java
URL: http://svn.apache.org/viewvc/webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/msgprocessors/AcknowledgementProcessor.java?rev=708804&r1=708803&r2=708804&view=diff
==============================================================================
--- webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/msgprocessors/AcknowledgementProcessor.java (original)
+++ webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/msgprocessors/AcknowledgementProcessor.java Tue Oct 28 23:35:47 2008
@@ -103,15 +103,15 @@
log.debug("Exit: AcknowledgementProcessor::processAckHeader, Sequence bean not found");
return;
}
-
+
if (outSequenceId == null || outSequenceId.length()==0) {
String message = SandeshaMessageHelper.getMessage(SandeshaMessageKeys.outSeqIDIsNull);
log.debug(message);
throw new SandeshaException(message);
}
- // If the message type is terminate sequence, then there may be a piggy backed ACK for a
+ // If the message type is terminate sequence, then there may be a piggy backed ACK for a
// sequence that has been terminated
- if (rmMsgCtx.getMessageType()!=Sandesha2Constants.MessageTypes.TERMINATE_SEQ &&
+ if (rmMsgCtx.getMessageType()!=Sandesha2Constants.MessageTypes.TERMINATE_SEQ &&
FaultManager.checkForSequenceTerminated(rmMsgCtx, outSequenceId, rmsBean, piggybackedAck)) {
if (log.isDebugEnabled())
log.debug("Exit: AcknowledgementProcessor::processAckHeader, Sequence terminated");
@@ -121,32 +121,44 @@
// Check that the sender of this Ack holds the correct token
String internalSequenceId = rmsBean.getInternalSequenceID();
SandeshaUtil.assertProofOfPossession(rmsBean, msgCtx, soapHeader);
-
+
if(log.isDebugEnabled()) log.debug("Got Ack for RM Sequence: " + outSequenceId + ", internalSeqId: " + internalSequenceId);
Iterator ackRangeIterator = sequenceAck.getAcknowledgementRanges().iterator();
-
+
if (FaultManager.checkForInvalidAcknowledgement(rmMsgCtx, sequenceAck, storageManager, rmsBean, piggybackedAck)) {
if (log.isDebugEnabled())
log.debug("Exit: AcknowledgementProcessor::processAckHeader, Invalid Ack range ");
return;
}
-
+
EndpointReference replyTo = rmsBean.getReplyToEndpointReference();
boolean anonReplyTo = replyTo==null || replyTo.isWSAddressingAnonymous(); //if this is wsa anonymous
//then we might be using replay
-
+
String rmVersion = rmMsgCtx.getRMSpecVersion();
-
+
// Compare the clientCompletedMessages with the range we just got, to work out if there
// is any new information in this ack message
RangeString completedMessages = rmsBean.getClientCompletedMessages();
long numberOfNewMessagesAcked = 0;
-
+
boolean ackNeedsToSendInvalidFault = false; //if this ack includes a msg that we have not sent then
//we should try to send a fault back to the client
Range firstInvalidRange = null; //If there is a single invalid range then we set it here.
//If there is more than one we report the first invalid range
+ //adding a MakeConnection for the response sequence if needed.
+ if (rmsBean.getOfferedSequence() != null) {
+
+ RMDBeanMgr rMDBeanMgr = storageManager.getRMDBeanMgr();
+ RMDBean rMDBean = rMDBeanMgr.retrieve(outSequenceId);
+
+ if (rMDBean!=null && rMDBean.isPollingMode()) {
+ PollingManager manager = storageManager.getPollingManager();
+ if(manager != null) manager.schedulePollingRequest(rMDBean.getSequenceID(), false);
+ }
+ }
+
while(ackRangeIterator.hasNext()) {
Range ackRange = (Range) ackRangeIterator.next();
long lower = ackRange.lowerValue;
@@ -157,14 +169,14 @@
//we now know that this range is complete so we update it. This should aggregate the
//ranges together and tell us which numbers are newly acked
Range[] newRanges = completedMessages.addRange(ackedRange).getRanges();
-
+
// We now take each newly acked message in turn and see if we need to update a sender bean
for (int rangeIndex=0; rangeIndex < newRanges.length; rangeIndex++) {
//now work on each newly acked message in this range
for(long messageNo = newRanges[rangeIndex].lowerValue; messageNo<=newRanges[rangeIndex].upperValue; messageNo++){
-
+
numberOfNewMessagesAcked++;
- SenderBean retransmitterBean = retransmitterMgr.retrieve(outSequenceId, messageNo);
+ SenderBean retransmitterBean = retransmitterMgr.retrieve(outSequenceId, messageNo);
if (retransmitterBean != null && retransmitterBean.getMessageType()==Sandesha2Constants.MessageTypes.APPLICATION) {
// Check we haven't got an Ack for an application message that hasn't been sent yet !
if (retransmitterBean.getSentCount() == 0 ) {
@@ -180,7 +192,7 @@
//delete the sender bean that has been validly acknowledged (unless
//we use replay model)
String storageKey = retransmitterBean.getMessageContextRefKey();
-
+
boolean syncResponseNeeded = false;
if (anonReplyTo) {
MessageContext applicationMessage = storageManager.retrieveMessageContext(storageKey, configCtx);
@@ -196,14 +208,14 @@
// removing the application message from the storage if there is no replay model
retransmitterMgr.delete(retransmitterBean.getMessageID());
storageManager.removeMessageContext(storageKey);
- }
+ }
}
}
}//end for
}//end for
} //end while
}
-
+
if(ackNeedsToSendInvalidFault){
//try to send an invalid ack
FaultManager.makeInvalidAcknowledgementFault(rmMsgCtx, sequenceAck, firstInvalidRange,
@@ -216,18 +228,6 @@
// updating the last activated time of the sequence.
rmsBean.setLastActivatedTime(System.currentTimeMillis());
- //adding a MakeConnection for the response sequence if needed.
- if (rmsBean.getOfferedSequence() != null) {
-
- RMDBeanMgr rMDBeanMgr = storageManager.getRMDBeanMgr();
- RMDBean rMDBean = rMDBeanMgr.retrieve(outSequenceId);
-
- if (rMDBean!=null && rMDBean.isPollingMode()) {
- PollingManager manager = storageManager.getPollingManager();
- if(manager != null) manager.schedulePollingRequest(rMDBean.getSequenceID(), false);
- }
- }
-
// We overwrite the previous client completed message ranges with the
// latest view, but only if it is an update i.e. contained a new
// ack range (which is because we do not previous acks arriving late
Modified: webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/workers/WorkerLock.java
URL: http://svn.apache.org/viewvc/webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/workers/WorkerLock.java?rev=708804&r1=708803&r2=708804&view=diff
==============================================================================
--- webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/workers/WorkerLock.java (original)
+++ webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/workers/WorkerLock.java Tue Oct 28 23:35:47 2008
@@ -82,7 +82,9 @@
if (LoggingControl.isAnyTracingEnabled() && log.isDebugEnabled())
log.debug("Enter: WorkerLock::removeWork " + work);
Holder h = (Holder) locks.remove(work);
- h.release();
+ if (h != null){
+ h.release();
+ }
if (LoggingControl.isAnyTracingEnabled() && log.isDebugEnabled())
log.debug("Exit: WorkerLock::removeWork");
}
lmpx.com only provides a reader for public news (NNTP) servers. It is not
affiliated with the servers or forums shown here and is not responsible for
the content of articles, which is written by their respective authors.