You are viewing a plain text version of this content. The canonical link for it is here.
Posted to fx-dev@ws.apache.org by ga...@apache.org on 2007/01/12 16:45:35 UTC
svn commit: r495613 - in /webservices/sandesha/trunk/java:
src/org/apache/sandesha2/ src/org/apache/sandesha2/msgprocessors/
src/org/apache/sandesha2/storage/beans/ src/org/apache/sandesha2/util/
src/org/apache/sandesha2/workers/ test/src/org/apache/sa...
Author: gatfora
Date: Fri Jan 12 07:45:34 2007
New Revision: 495613
URL: http://svn.apache.org/viewvc?view=rev&rev=495613
Log:
Move the OutOfOrderRanges, TermianteReceived, lastActivatedTime, noOutGoingMsgs, transportTo, sequenceClosedClient and lastInMsgId to RMS/RMDBeans
Modified:
webservices/sandesha/trunk/java/src/org/apache/sandesha2/Sandesha2Constants.java
webservices/sandesha/trunk/java/src/org/apache/sandesha2/msgprocessors/AcknowledgementProcessor.java
webservices/sandesha/trunk/java/src/org/apache/sandesha2/msgprocessors/ApplicationMsgProcessor.java
webservices/sandesha/trunk/java/src/org/apache/sandesha2/msgprocessors/CloseSequenceProcessor.java
webservices/sandesha/trunk/java/src/org/apache/sandesha2/msgprocessors/CreateSeqMsgProcessor.java
webservices/sandesha/trunk/java/src/org/apache/sandesha2/msgprocessors/CreateSeqResponseMsgProcessor.java
webservices/sandesha/trunk/java/src/org/apache/sandesha2/msgprocessors/SequenceProcessor.java
webservices/sandesha/trunk/java/src/org/apache/sandesha2/msgprocessors/TerminateSeqMsgProcessor.java
webservices/sandesha/trunk/java/src/org/apache/sandesha2/storage/beans/RMDBean.java
webservices/sandesha/trunk/java/src/org/apache/sandesha2/storage/beans/RMSBean.java
webservices/sandesha/trunk/java/src/org/apache/sandesha2/storage/beans/RMSequenceBean.java
webservices/sandesha/trunk/java/src/org/apache/sandesha2/util/SequenceManager.java
webservices/sandesha/trunk/java/src/org/apache/sandesha2/util/TerminateManager.java
webservices/sandesha/trunk/java/src/org/apache/sandesha2/util/WSRMMessageSender.java
webservices/sandesha/trunk/java/src/org/apache/sandesha2/workers/Invoker.java
webservices/sandesha/trunk/java/test/src/org/apache/sandesha2/workers/ForceInboundDispatchTest.java
Modified: webservices/sandesha/trunk/java/src/org/apache/sandesha2/Sandesha2Constants.java
URL: http://svn.apache.org/viewvc/webservices/sandesha/trunk/java/src/org/apache/sandesha2/Sandesha2Constants.java?view=diff&rev=495613&r1=495612&r2=495613
==============================================================================
--- webservices/sandesha/trunk/java/src/org/apache/sandesha2/Sandesha2Constants.java (original)
+++ webservices/sandesha/trunk/java/src/org/apache/sandesha2/Sandesha2Constants.java Fri Jan 12 07:45:34 2007
@@ -253,11 +253,6 @@
// property. This is used as the
// sequenceId to share data b/w
// sequences
-
- //For IN_ORDER sequences, we can have finite ranges of messages that can be
- //delivered out of order. These are maintained as a String that is consistent
- //with the form described in org.apache.sandesha2.util.RangeString
- String OUT_OF_ORDER_RANGES = "OutOfOrderRanges";
String INTERNAL_SEQUENCE_ID = "TempSequenceId";
@@ -265,25 +260,10 @@
String OFFERED_SEQUENCE = "OfferedSequence";
- String TERMINATE_RECEIVED = "TerminateReceived";
-
- String LAST_ACTIVATED_TIME = "LastActivatedTime";
-
- String NO_OF_OUTGOING_MSGS_ACKED = "NoOfOutGoingMessagesAcked";
-
- String TRANSPORT_TO = "TransportTo";
-
String SEQUENCE_CLOSED = "SequenceClosed";
- String SEQUENCE_CLOSED_CLIENT = "SequenceClosedClient"; //indicates the client has sent a close sequence
-
String SEQUENCE_TIMED_OUT = "SequenceTimedOut";
-
- // Once an inbound sequence is closed, or we receive a message with the
- // 'LastMessage' marker, we record the message id of the highest message
- // in the sequence.
- String LAST_IN_MSG_ID = "LastInMessageId";
-
+
String SECURITY_TOKEN = "SecurityToken";
String SOAP_VERSION = "SOAPVersion";
Modified: webservices/sandesha/trunk/java/src/org/apache/sandesha2/msgprocessors/AcknowledgementProcessor.java
URL: http://svn.apache.org/viewvc/webservices/sandesha/trunk/java/src/org/apache/sandesha2/msgprocessors/AcknowledgementProcessor.java?view=diff&rev=495613&r1=495612&r2=495613
==============================================================================
--- webservices/sandesha/trunk/java/src/org/apache/sandesha2/msgprocessors/AcknowledgementProcessor.java (original)
+++ webservices/sandesha/trunk/java/src/org/apache/sandesha2/msgprocessors/AcknowledgementProcessor.java Fri Jan 12 07:45:34 2007
@@ -168,8 +168,9 @@
}
}
+ RMSBean rmsBean = SandeshaUtil.getRMSBeanFromSequenceId(storageManager, outSequenceId);
// updating the last activated time of the sequence.
- SequenceManager.updateLastActivatedTime(sequencePropertyKey, storageManager);
+ rmsBean.setLastActivatedTime(System.currentTimeMillis());
while (nackIterator.hasNext()) {
Nack nack = (Nack) nackIterator.next();
@@ -193,33 +194,14 @@
// setting acked message date.
// TODO add details specific to each message.
- long noOfMsgsAcked = getNoOfMessagesAcked(sequenceAck.getAcknowledgementRanges().iterator());
- SequencePropertyBean noOfMsgsAckedBean = seqPropMgr.retrieve(sequencePropertyKey,
- Sandesha2Constants.SequenceProperties.NO_OF_OUTGOING_MSGS_ACKED);
- boolean added = false;
-
- if (noOfMsgsAckedBean == null) {
- added = true;
- noOfMsgsAckedBean = new SequencePropertyBean();
- noOfMsgsAckedBean.setSequencePropertyKey(sequencePropertyKey);
- noOfMsgsAckedBean.setName(Sandesha2Constants.SequenceProperties.NO_OF_OUTGOING_MSGS_ACKED);
- }
-
- noOfMsgsAckedBean.setValue(Long.toString(noOfMsgsAcked));
-
- if (added)
- seqPropMgr.insert(noOfMsgsAckedBean);
- else
- seqPropMgr.update(noOfMsgsAckedBean);
-
- // setting the completed_messages list. This gives all the messages of
- // the sequence that were acked.
- RMSBean rmsBean = SandeshaUtil.getRMSBeanFromSequenceId(storageManager, outSequenceId);
// Set the completed message list, but only if we have actually removed a SenderBean
// It is possible for the ACK messages arrive out of sequence
- if (removedSenderBean)
+ if (removedSenderBean) {
rmsBean.setClientCompletedMessages(ackedMessagesList);
+ long noOfMsgsAcked = ackedMessagesList.size();
+ rmsBean.setNumberOfMessagesAcked(noOfMsgsAcked);
+ }
long highestOutMsgNo = rmsBean.getLastOutMessage();
@@ -255,20 +237,4 @@
return null;
}
-
- private static long getNoOfMessagesAcked(Iterator ackRangeIterator) {
- long noOfMsgs = 0;
- while (ackRangeIterator.hasNext()) {
- AcknowledgementRange acknowledgementRange = (AcknowledgementRange) ackRangeIterator.next();
- long lower = acknowledgementRange.getLowerValue();
- long upper = acknowledgementRange.getUpperValue();
-
- for (long i = lower; i <= upper; i++) {
- noOfMsgs++;
- }
- }
-
- return noOfMsgs;
- }
-
}
Modified: webservices/sandesha/trunk/java/src/org/apache/sandesha2/msgprocessors/ApplicationMsgProcessor.java
URL: http://svn.apache.org/viewvc/webservices/sandesha/trunk/java/src/org/apache/sandesha2/msgprocessors/ApplicationMsgProcessor.java?view=diff&rev=495613&r1=495612&r2=495613
==============================================================================
--- webservices/sandesha/trunk/java/src/org/apache/sandesha2/msgprocessors/ApplicationMsgProcessor.java (original)
+++ webservices/sandesha/trunk/java/src/org/apache/sandesha2/msgprocessors/ApplicationMsgProcessor.java Fri Jan 12 07:45:34 2007
@@ -137,10 +137,11 @@
internalSequenceId = SandeshaUtil.getOutgoingSideInternalSequenceID(inboundSequence);
- // Deciding weather this is the last message. We assume it is if it relates to
+ // Deciding whether this is the last message. We assume it is if it relates to
// a message which arrived with the LastMessage flag on it.
- String lastRequestId = SandeshaUtil.getSequenceProperty(inboundSequence,
- Sandesha2Constants.SequenceProperties.LAST_IN_MSG_ID, storageManager);
+ RMDBean rmdBean = SandeshaUtil.getRMDBeanFromSequenceId(storageManager, inboundSequence);
+ // Get the last in message
+ String lastRequestId = rmdBean.getLastInMessageId();
RelatesTo relatesTo = msgContext.getRelatesTo();
if(relatesTo != null && lastRequestId != null &&
lastRequestId.equals(relatesTo.getValue())) {
@@ -196,13 +197,13 @@
if (dummyMessageString != null && Sandesha2Constants.VALUE_TRUE.equals(dummyMessageString))
dummyMessage = true;
+ RMSBean rmsBean = SandeshaUtil.getRMSBeanFromInternalSequenceId(storageManager, internalSequenceId);
+
//see if the sequence is closed
- SequencePropertyBean sequenceClosed = seqPropMgr.retrieve(sequencePropertyKey, Sandesha2Constants.SequenceProperties.SEQUENCE_CLOSED_CLIENT);
- if(sequenceClosed!=null){
+ if(rmsBean != null && rmsBean.isSequenceClosedClient()){
throw new SandeshaException(SandeshaMessageHelper.getMessage(SandeshaMessageKeys.cannotSendMsgAsSequenceClosed, internalSequenceId));
}
- RMSBean rmsBean = SandeshaUtil.getRMSBeanFromInternalSequenceId(storageManager, internalSequenceId);
//see if the sequence is terminated
if(rmsBean != null && rmsBean.isTerminateAdded()) {
throw new SandeshaException(SandeshaMessageHelper.getMessage(SandeshaMessageKeys.cannotSendMsgAsSequenceTerminated, internalSequenceId));
Modified: webservices/sandesha/trunk/java/src/org/apache/sandesha2/msgprocessors/CloseSequenceProcessor.java
URL: http://svn.apache.org/viewvc/webservices/sandesha/trunk/java/src/org/apache/sandesha2/msgprocessors/CloseSequenceProcessor.java?view=diff&rev=495613&r1=495612&r2=495613
==============================================================================
--- webservices/sandesha/trunk/java/src/org/apache/sandesha2/msgprocessors/CloseSequenceProcessor.java (original)
+++ webservices/sandesha/trunk/java/src/org/apache/sandesha2/msgprocessors/CloseSequenceProcessor.java Fri Jan 12 07:45:34 2007
@@ -159,11 +159,8 @@
setupOutMessage(rmMsgCtx);
//write into the sequence proeprties that the client is now closed
- SequencePropertyBean sequenceClosedBean = new SequencePropertyBean();
- sequenceClosedBean.setSequencePropertyKey(getInternalSequenceID());
- sequenceClosedBean.setName(Sandesha2Constants.SequenceProperties.SEQUENCE_CLOSED_CLIENT);
- sequenceClosedBean.setValue(Sandesha2Constants.VALUE_TRUE);
- getStorageManager().getSequencePropertyBeanMgr().insert(sequenceClosedBean);
+ getRMSBean().setSequenceClosedClient(true);
+ getStorageManager().getRMSBeanMgr().update(getRMSBean());
AxisOperation closeOperation = SpecSpecificConstants.getWSRMOperation(
Sandesha2Constants.MessageTypes.CLOSE_SEQUENCE,
Modified: webservices/sandesha/trunk/java/src/org/apache/sandesha2/msgprocessors/CreateSeqMsgProcessor.java
URL: http://svn.apache.org/viewvc/webservices/sandesha/trunk/java/src/org/apache/sandesha2/msgprocessors/CreateSeqMsgProcessor.java?view=diff&rev=495613&r1=495612&r2=495613
==============================================================================
--- webservices/sandesha/trunk/java/src/org/apache/sandesha2/msgprocessors/CreateSeqMsgProcessor.java (original)
+++ webservices/sandesha/trunk/java/src/org/apache/sandesha2/msgprocessors/CreateSeqMsgProcessor.java Fri Jan 12 07:45:34 2007
@@ -99,158 +99,145 @@
MessageContext outMessage = null;
SequencePropertyBeanMgr seqPropMgr = storageManager.getSequencePropertyBeanMgr();
- try {
- // Create the new sequence id, as well as establishing the beans that handle the
- // sequence state.
- RMDBean rmdBean = SequenceManager.setupNewSequence(createSeqRMMsg, storageManager, secManager, token);
-
- RMMsgContext createSeqResponse = RMMsgCreator.createCreateSeqResponseMsg(createSeqRMMsg, rmdBean.getSequenceID());
- outMessage = createSeqResponse.getMessageContext();
-
- createSeqResponse.setFlow(MessageContext.OUT_FLOW);
-
- createSeqResponse.setProperty(Sandesha2Constants.APPLICATION_PROCESSING_DONE, "true"); // for
- // making
- // sure
- // that
- // this
- // wont
- // be
- // processed
- // again.
- CreateSequenceResponse createSeqResPart = (CreateSequenceResponse) createSeqResponse
- .getMessagePart(Sandesha2Constants.MessageParts.CREATE_SEQ_RESPONSE);
+ // Create the new sequence id, as well as establishing the beans that handle the
+ // sequence state.
+ RMDBean rmdBean = SequenceManager.setupNewSequence(createSeqRMMsg, storageManager, secManager, token);
+
+ RMMsgContext createSeqResponse = RMMsgCreator.createCreateSeqResponseMsg(createSeqRMMsg, rmdBean.getSequenceID());
+ outMessage = createSeqResponse.getMessageContext();
+
+ createSeqResponse.setFlow(MessageContext.OUT_FLOW);
- // OFFER PROCESSING
- SequenceOffer offer = createSeqPart.getSequenceOffer();
- if (offer != null) {
- Accept accept = createSeqResPart.getAccept();
- if (accept == null) {
- String message = SandeshaMessageHelper.getMessage(SandeshaMessageKeys.noAcceptPart);
- log.debug(message);
- throw new SandeshaException(message);
- }
+ // for making sure that this won't be processed again
+ createSeqResponse.setProperty(Sandesha2Constants.APPLICATION_PROCESSING_DONE, "true");
+
+ CreateSequenceResponse createSeqResPart = (CreateSequenceResponse) createSeqResponse
+ .getMessagePart(Sandesha2Constants.MessageParts.CREATE_SEQ_RESPONSE);
- String offeredSequenceID = offer.getIdentifer().getIdentifier(); // offered
- // seq.
- // id.
-
- boolean offerEcepted = offerAccepted(offeredSequenceID, context, createSeqRMMsg, storageManager);
-
- if (offerEcepted) {
- // Setting the CreateSequence table entry for the outgoing
- // side.
- RMSBean rMSBean = new RMSBean();
- rMSBean.setSequenceID(offeredSequenceID);
- String outgoingSideInternalSequenceId = SandeshaUtil
- .getOutgoingSideInternalSequenceID(rmdBean.getSequenceID());
- rMSBean.setInternalSequenceID(outgoingSideInternalSequenceId);
- // this is a dummy value
- rMSBean.setCreateSeqMsgID(SandeshaUtil.getUUID());
-
- rMSBean.setToEPR(rmdBean.getToEPR());
- rMSBean.setAcksToEPR(rmdBean.getAcksToEPR());
- rMSBean.setReplyToEPR(rmdBean.getReplyToEPR());
-
- String outgoingSideSequencePropertyKey = outgoingSideInternalSequenceId;
+ // OFFER PROCESSING
+ SequenceOffer offer = createSeqPart.getSequenceOffer();
+ if (offer != null) {
+ Accept accept = createSeqResPart.getAccept();
+ if (accept == null) {
+ String message = SandeshaMessageHelper.getMessage(SandeshaMessageKeys.noAcceptPart);
+ log.debug(message);
+ throw new SandeshaException(message);
+ }
- RMSBeanMgr createSeqMgr = storageManager.getRMSBeanMgr();
- createSeqMgr.insert(rMSBean);
+ // offered seq id
+ String offeredSequenceID = offer.getIdentifer().getIdentifier();
+
+ boolean offerEcepted = offerAccepted(offeredSequenceID, context, createSeqRMMsg, storageManager);
- // Setting sequence properties for the outgoing sequence.
- // Only will be used by the server side response path. Will
- // be wasted properties for the client side.
-
- // setting the internal_sequence_id
- SequencePropertyBean internalSequenceBean = new SequencePropertyBean();
- internalSequenceBean.setName(Sandesha2Constants.SequenceProperties.INTERNAL_SEQUENCE_ID);
- internalSequenceBean.setSequencePropertyKey(offeredSequenceID);
- internalSequenceBean.setValue(outgoingSideInternalSequenceId);
- seqPropMgr.insert(internalSequenceBean);
+ if (offerEcepted) {
+ // Setting the CreateSequence table entry for the outgoing
+ // side.
+ RMSBean rMSBean = new RMSBean();
+ rMSBean.setSequenceID(offeredSequenceID);
+ String outgoingSideInternalSequenceId = SandeshaUtil
+ .getOutgoingSideInternalSequenceID(rmdBean.getSequenceID());
+ rMSBean.setInternalSequenceID(outgoingSideInternalSequenceId);
+ // this is a dummy value
+ rMSBean.setCreateSeqMsgID(SandeshaUtil.getUUID());
- Endpoint endpoint = offer.getEndpoint();
- if (endpoint!=null) {
- // setting the OfferedEndpoint
- SequencePropertyBean offeredEndpointBean = new SequencePropertyBean();
- offeredEndpointBean.setName(Sandesha2Constants.SequenceProperties.OFFERED_ENDPOINT);
+ rMSBean.setToEPR(rmdBean.getToEPR());
+ rMSBean.setAcksToEPR(rmdBean.getAcksToEPR());
+ rMSBean.setReplyToEPR(rmdBean.getReplyToEPR());
+ rMSBean.setLastActivatedTime(System.currentTimeMillis());
+
+ String outgoingSideSequencePropertyKey = outgoingSideInternalSequenceId;
+
+ RMSBeanMgr rmsBeanMgr = storageManager.getRMSBeanMgr();
+ rmsBeanMgr.insert(rMSBean);
+
+ // Setting sequence properties for the outgoing sequence.
+ // Only will be used by the server side response path. Will
+ // be wasted properties for the client side.
+
+ // setting the internal_sequence_id
+ SequencePropertyBean internalSequenceBean = new SequencePropertyBean();
+ internalSequenceBean.setName(Sandesha2Constants.SequenceProperties.INTERNAL_SEQUENCE_ID);
+ internalSequenceBean.setSequencePropertyKey(offeredSequenceID);
+ internalSequenceBean.setValue(outgoingSideInternalSequenceId);
+ seqPropMgr.insert(internalSequenceBean);
- //currently we can only serialize the Address part of the Endpoint.
- //TODO correct this to serialize the whole EPR.
- offeredEndpointBean.setValue(endpoint.getEPR().getAddress());
- offeredEndpointBean.setSequencePropertyKey(outgoingSideSequencePropertyKey);
- seqPropMgr.insert(offeredEndpointBean);
- }
+ Endpoint endpoint = offer.getEndpoint();
+ if (endpoint!=null) {
+ // setting the OfferedEndpoint
+ SequencePropertyBean offeredEndpointBean = new SequencePropertyBean();
+ offeredEndpointBean.setName(Sandesha2Constants.SequenceProperties.OFFERED_ENDPOINT);
+
+ //currently we can only serialize the Address part of the Endpoint.
+ //TODO correct this to serialize the whole EPR.
+ offeredEndpointBean.setValue(endpoint.getEPR().getAddress());
+ offeredEndpointBean.setSequencePropertyKey(outgoingSideSequencePropertyKey);
+ seqPropMgr.insert(offeredEndpointBean);
+ }
- // Store the inbound token (if any) with the new sequence
- if(token != null) {
- String tokenData = secManager.getTokenRecoveryData(token);
- SequencePropertyBean tokenBean = new SequencePropertyBean(
- outgoingSideSequencePropertyKey,
- Sandesha2Constants.SequenceProperties.SECURITY_TOKEN,
- tokenData);
- seqPropMgr.insert(tokenBean);
- }
- } else {
- // removing the accept part.
- createSeqResPart.setAccept(null);
- createSeqResponse.addSOAPEnvelope();
+ // Store the inbound token (if any) with the new sequence
+ if(token != null) {
+ String tokenData = secManager.getTokenRecoveryData(token);
+ SequencePropertyBean tokenBean = new SequencePropertyBean(
+ outgoingSideSequencePropertyKey,
+ Sandesha2Constants.SequenceProperties.SECURITY_TOKEN,
+ tokenData);
+ seqPropMgr.insert(tokenBean);
}
+ } else {
+ // removing the accept part.
+ createSeqResPart.setAccept(null);
+ createSeqResponse.addSOAPEnvelope();
}
+ }
// Add this sequence to the list of inbound sequences
- SequencePropertyBean incomingSequenceListBean = seqPropMgr.retrieve(
- Sandesha2Constants.SequenceProperties.ALL_SEQUENCES,
- Sandesha2Constants.SequenceProperties.INCOMING_SEQUENCE_LIST);
-
- if (incomingSequenceListBean == null) {
- incomingSequenceListBean = new SequencePropertyBean();
- incomingSequenceListBean.setSequencePropertyKey(Sandesha2Constants.SequenceProperties.ALL_SEQUENCES);
- incomingSequenceListBean.setName(Sandesha2Constants.SequenceProperties.INCOMING_SEQUENCE_LIST);
- incomingSequenceListBean.setValue(null);
-
- // this get inserted before
- seqPropMgr.insert(incomingSequenceListBean);
- }
+ SequencePropertyBean incomingSequenceListBean = seqPropMgr.retrieve(
+ Sandesha2Constants.SequenceProperties.ALL_SEQUENCES,
+ Sandesha2Constants.SequenceProperties.INCOMING_SEQUENCE_LIST);
+
+ if (incomingSequenceListBean == null) {
+ incomingSequenceListBean = new SequencePropertyBean();
+ incomingSequenceListBean.setSequencePropertyKey(Sandesha2Constants.SequenceProperties.ALL_SEQUENCES);
+ incomingSequenceListBean.setName(Sandesha2Constants.SequenceProperties.INCOMING_SEQUENCE_LIST);
+ incomingSequenceListBean.setValue(null);
- ArrayList incomingSequenceList = SandeshaUtil.getArrayListFromString(incomingSequenceListBean.getValue());
- incomingSequenceList.add(rmdBean.getSequenceID());
- incomingSequenceListBean.setValue(incomingSequenceList.toString());
- seqPropMgr.update(incomingSequenceListBean);
+ // this get inserted before
+ seqPropMgr.insert(incomingSequenceListBean);
+ }
+ ArrayList incomingSequenceList = SandeshaUtil.getArrayListFromString(incomingSequenceListBean.getValue());
+ incomingSequenceList.add(rmdBean.getSequenceID());
+ incomingSequenceListBean.setValue(incomingSequenceList.toString());
+ seqPropMgr.update(incomingSequenceListBean);
- //TODO add createSequenceResponse message as the referenceMessage to the RMDBean.
-
- outMessage.setResponseWritten(true);
+ //TODO add createSequenceResponse message as the referenceMessage to the RMDBean.
- // commiting tr. before sending the response msg.
+ outMessage.setResponseWritten(true);
- SequenceManager.updateLastActivatedTime(rmdBean.getSequenceID(), storageManager);
+ rmdBean.setLastActivatedTime(System.currentTimeMillis());
+ storageManager.getRMDBeanMgr().update(rmdBean);
- AxisEngine engine = new AxisEngine(context);
- try{
- engine.send(outMessage);
- }
- catch(AxisFault e){
- throw new SandeshaException(
- SandeshaMessageHelper.getMessage(SandeshaMessageKeys.couldNotSendCreateSeqResponse, e.toString()),
- e);
- }
+ AxisEngine engine = new AxisEngine(context);
+ try{
+ engine.send(outMessage);
+ }
+ catch(AxisFault e){
+ throw new SandeshaException(
+ SandeshaMessageHelper.getMessage(SandeshaMessageKeys.couldNotSendCreateSeqResponse, e.toString()),
+ e);
+ }
- boolean anon = true;
- if (rmdBean.getToEPR() != null) {
- EndpointReference toEPR = new EndpointReference(rmdBean.getToEPR());
- if (!toEPR.hasAnonymousAddress()) anon = false;
- }
- if(anon) {
- createSeqMsg.getOperationContext().setProperty(org.apache.axis2.Constants.RESPONSE_WRITTEN, "true");
- } else {
+ boolean anon = true;
+ if (rmdBean.getToEPR() != null) {
+ EndpointReference toEPR = new EndpointReference(rmdBean.getToEPR());
+ if (!toEPR.hasAnonymousAddress()) anon = false;
+ }
+ if(anon) {
+ createSeqMsg.getOperationContext().setProperty(org.apache.axis2.Constants.RESPONSE_WRITTEN, "true");
+ } else {
createSeqMsg.getOperationContext().setProperty(org.apache.axis2.Constants.RESPONSE_WRITTEN, "false");
- }
-
- } catch (AxisFault e1) {
- throw new SandeshaException(e1);
}
-
+
createSeqRMMsg.pause();
if (log.isDebugEnabled())
Modified: webservices/sandesha/trunk/java/src/org/apache/sandesha2/msgprocessors/CreateSeqResponseMsgProcessor.java
URL: http://svn.apache.org/viewvc/webservices/sandesha/trunk/java/src/org/apache/sandesha2/msgprocessors/CreateSeqResponseMsgProcessor.java?view=diff&rev=495613&r1=495612&r2=495613
==============================================================================
--- webservices/sandesha/trunk/java/src/org/apache/sandesha2/msgprocessors/CreateSeqResponseMsgProcessor.java (original)
+++ webservices/sandesha/trunk/java/src/org/apache/sandesha2/msgprocessors/CreateSeqResponseMsgProcessor.java Fri Jan 12 07:45:34 2007
@@ -48,7 +48,6 @@
import org.apache.sandesha2.storage.beans.SequencePropertyBean;
import org.apache.sandesha2.util.MsgInitializer;
import org.apache.sandesha2.util.SandeshaUtil;
-import org.apache.sandesha2.util.SequenceManager;
import org.apache.sandesha2.util.SpecSpecificConstants;
import org.apache.sandesha2.wsrm.Accept;
import org.apache.sandesha2.wsrm.AckRequested;
@@ -110,9 +109,9 @@
}
SenderBeanMgr retransmitterMgr = storageManager.getSenderBeanMgr();
- RMSBeanMgr createSeqMgr = storageManager.getRMSBeanMgr();
+ RMSBeanMgr rmsBeanMgr = storageManager.getRMSBeanMgr();
- RMSBean rmsBean = createSeqMgr.retrieve(createSeqMsgId);
+ RMSBean rmsBean = rmsBeanMgr.retrieve(createSeqMsgId);
if (rmsBean == null) {
String message = SandeshaMessageHelper.getMessage(SandeshaMessageKeys.createSeqEntryNotFound);
log.debug(message);
@@ -231,7 +230,8 @@
RMDBeanMgr rmdBeanMgr = storageManager.getRMDBeanMgr();
rmdBeanMgr.insert(rMDBean);
- createSeqMgr.update(rmsBean);
+ rmsBean.setLastActivatedTime(System.currentTimeMillis());
+ rmsBeanMgr.update(rmsBean);
// Store the security token for the offered sequence
if(tokenData != null) {
@@ -365,8 +365,6 @@
// updating the message. this will correct the SOAP envelope string.
storageManager.updateMessageContext(key, applicationMsg);
}
-
- SequenceManager.updateLastActivatedTime(sequencePropertyKey, storageManager);
createSeqResponseRMMsgCtx.getMessageContext().getOperationContext().setProperty(
org.apache.axis2.Constants.RESPONSE_WRITTEN, "false");
Modified: webservices/sandesha/trunk/java/src/org/apache/sandesha2/msgprocessors/SequenceProcessor.java
URL: http://svn.apache.org/viewvc/webservices/sandesha/trunk/java/src/org/apache/sandesha2/msgprocessors/SequenceProcessor.java?view=diff&rev=495613&r1=495612&r2=495613
==============================================================================
--- webservices/sandesha/trunk/java/src/org/apache/sandesha2/msgprocessors/SequenceProcessor.java (original)
+++ webservices/sandesha/trunk/java/src/org/apache/sandesha2/msgprocessors/SequenceProcessor.java Fri Jan 12 07:45:34 2007
@@ -55,7 +55,6 @@
import org.apache.sandesha2.util.FaultManager;
import org.apache.sandesha2.util.MsgInitializer;
import org.apache.sandesha2.util.SandeshaUtil;
-import org.apache.sandesha2.util.SequenceManager;
import org.apache.sandesha2.wsrm.Sequence;
/**
@@ -138,9 +137,6 @@
FaultManager.checkForSequenceClosed(rmMsgCtx, sequenceId, storageManager);
FaultManager.checkForLastMsgNumberExceeded(rmMsgCtx, storageManager);
- // updating the last activated time of the sequence.
- SequenceManager.updateLastActivatedTime(propertyKey, storageManager);
-
long msgNo = sequence.getMessageNumber().getMessageNumber();
if (msgNo == 0) {
String message = SandeshaMessageHelper.getMessage(SandeshaMessageKeys.invalidMsgNumber, Long
@@ -157,6 +153,9 @@
throw new SandeshaException(SandeshaMessageHelper.getMessage(SandeshaMessageKeys.cannotFindSequence,
sequenceId));
}
+
+ // updating the last activated time of the sequence.
+ bean.setLastActivatedTime(System.currentTimeMillis());
String key = SandeshaUtil.getUUID(); // key to store the message.
// updating the Highest_In_Msg_No property which gives the highest
Modified: webservices/sandesha/trunk/java/src/org/apache/sandesha2/msgprocessors/TerminateSeqMsgProcessor.java
URL: http://svn.apache.org/viewvc/webservices/sandesha/trunk/java/src/org/apache/sandesha2/msgprocessors/TerminateSeqMsgProcessor.java?view=diff&rev=495613&r1=495612&r2=495613
==============================================================================
--- webservices/sandesha/trunk/java/src/org/apache/sandesha2/msgprocessors/TerminateSeqMsgProcessor.java (original)
+++ webservices/sandesha/trunk/java/src/org/apache/sandesha2/msgprocessors/TerminateSeqMsgProcessor.java Fri Jan 12 07:45:34 2007
@@ -48,7 +48,6 @@
import org.apache.sandesha2.util.FaultManager;
import org.apache.sandesha2.util.RMMsgCreator;
import org.apache.sandesha2.util.SandeshaUtil;
-import org.apache.sandesha2.util.SequenceManager;
import org.apache.sandesha2.util.SpecSpecificConstants;
import org.apache.sandesha2.util.TerminateManager;
import org.apache.sandesha2.util.WSRMMessageSender;
@@ -104,13 +103,6 @@
FaultManager.checkForUnknownSequence(terminateSeqRMMsg, sequenceId, storageManager);
- SequencePropertyBean terminateReceivedBean = new SequencePropertyBean();
- terminateReceivedBean.setSequencePropertyKey(sequencePropertyKey);
- terminateReceivedBean.setName(Sandesha2Constants.SequenceProperties.TERMINATE_RECEIVED);
- terminateReceivedBean.setValue("true");
-
- sequencePropertyBeanMgr.insert(terminateReceivedBean);
-
// add the terminate sequence response if required.
RMMsgContext terminateSequenceResponse = null;
if (SpecSpecificConstants.isTerminateSequenceResponseRequired(terminateSeqRMMsg.getRMSpecVersion()))
@@ -120,12 +112,11 @@
RMDBean rmdBean = SandeshaUtil.getRMDBeanFromSequenceId(storageManager, sequenceId);
rmdBean.setTerminated(true);
+ rmdBean.setLastActivatedTime(System.currentTimeMillis());
storageManager.getRMDBeanMgr().update(rmdBean);
TerminateManager.cleanReceivingSideOnTerminateMessage(context, sequencePropertyKey, sequenceId, storageManager);
- SequenceManager.updateLastActivatedTime(sequencePropertyKey, storageManager);
-
//sending the terminate sequence response
if (terminateSequenceResponse != null) {
@@ -134,11 +125,9 @@
AxisEngine engine = new AxisEngine(terminateSeqMsg
.getConfigurationContext());
-
-
+
outMessage.setServerSide(true);
-
-
+
engine.send(outMessage);
if (toEPR.hasAnonymousAddress()) {
@@ -164,8 +153,6 @@
if (log.isDebugEnabled())
log.debug("Enter: TerminateSeqMsgProcessor::setUpHighestMsgNumbers, " + sequenceId);
- SequencePropertyBeanMgr seqPropMgr = storageManager.getSequencePropertyBeanMgr();
-
RMDBeanMgr mgr = storageManager.getRMDBeanMgr();
RMDBean bean = mgr.retrieve(sequenceId);
@@ -187,9 +174,10 @@
// Mark up the highest inbound message as if it had the last message flag on it.
//
String inMsgId = bean.getHighestInMessageId();
- SequencePropertyBean lastInMsgBean = new SequencePropertyBean(requestSidesequencePropertyKey,
- Sandesha2Constants.SequenceProperties.LAST_IN_MSG_ID, bean.getHighestInMessageId());
- seqPropMgr.insert(lastInMsgBean);
+ bean.setLastInMessageId(inMsgId);
+
+ // Update the RMDBean
+ storageManager.getRMDBeanMgr().update(bean);
// If an outbound message has already gone out with that relatesTo, then we can terminate
// right away.
Modified: webservices/sandesha/trunk/java/src/org/apache/sandesha2/storage/beans/RMDBean.java
URL: http://svn.apache.org/viewvc/webservices/sandesha/trunk/java/src/org/apache/sandesha2/storage/beans/RMDBean.java?view=diff&rev=495613&r1=495612&r2=495613
==============================================================================
--- webservices/sandesha/trunk/java/src/org/apache/sandesha2/storage/beans/RMDBean.java (original)
+++ webservices/sandesha/trunk/java/src/org/apache/sandesha2/storage/beans/RMDBean.java Fri Jan 12 07:45:34 2007
@@ -42,11 +42,26 @@
private long highestInMessageNumber = 0;
private String highestInMessageId;
-
- /** For incoming sequences this gives the msg no's of the messages that were
+
+ /**
+ * Once an inbound sequence is closed, or we receive a message with the
+ * 'LastMessage' marker, we record the message id of the highest message
+ * in the sequence.
+ */
+ private String lastInMessageId;
+
+ /**
+ * For incoming sequences this gives the msg no's of the messages that were
* received (may be an ack was sent - depending on the policy)
*/
private List serverCompletedMessages = null;
+
+ /**
+ * For IN_ORDER sequences, we can have finite ranges of messages that can be
+ * delivered out of order. These are maintained as a String that is consistent
+ * with the form described in org.apache.sandesha2.util.RangeString
+ */
+ private String outOfOrderRanges = null;
public RMDBean() {
@@ -95,6 +110,30 @@
public void setHighestInMessageNumber(long highestInMessageNumber) {
this.highestInMessageNumber = highestInMessageNumber;
}
+
+ public List getServerCompletedMessages() {
+ return serverCompletedMessages;
+ }
+
+ public void setServerCompletedMessages(List serverCompletedMessages) {
+ this.serverCompletedMessages = serverCompletedMessages;
+ }
+
+ public String getLastInMessageId() {
+ return lastInMessageId;
+ }
+
+ public void setLastInMessageId(String lastInMessageId) {
+ this.lastInMessageId = lastInMessageId;
+ }
+
+ public String getOutOfOrderRanges() {
+ return outOfOrderRanges;
+ }
+
+ public void setOutOfOrderRanges(String outOfOrderRanges) {
+ this.outOfOrderRanges = outOfOrderRanges;
+ }
public String toString() {
StringBuffer result = new StringBuffer();
@@ -104,15 +143,8 @@
result.append("\nRef Msg Key: "); result.append(referenceMessageKey);
result.append("\nHishestInMessageNumber: "); result.append(highestInMessageNumber);
result.append("\nHishestInMessageKey: "); result.append(highestInMessageId);
+ result.append("\nLastInMessageId: "); result.append(lastInMessageId);
+ result.append("\nOutOfOrderRanges :"); result.append(outOfOrderRanges);
return result.toString();
}
-
- public List getServerCompletedMessages() {
- return serverCompletedMessages;
- }
-
- public void setServerCompletedMessages(List serverCompletedMessages) {
- this.serverCompletedMessages = serverCompletedMessages;
- }
-
}
Modified: webservices/sandesha/trunk/java/src/org/apache/sandesha2/storage/beans/RMSBean.java
URL: http://svn.apache.org/viewvc/webservices/sandesha/trunk/java/src/org/apache/sandesha2/storage/beans/RMSBean.java?view=diff&rev=495613&r1=495612&r2=495613
==============================================================================
--- webservices/sandesha/trunk/java/src/org/apache/sandesha2/storage/beans/RMSBean.java (original)
+++ webservices/sandesha/trunk/java/src/org/apache/sandesha2/storage/beans/RMSBean.java Fri Jan 12 07:45:34 2007
@@ -105,6 +105,18 @@
* Indicates that a terminate sequence message was added.
*/
private boolean terminateAdded = false;
+
+ /**
+ * The number of messages that were acked
+ */
+ private long numberOfMessagesAcked = 0;
+
+ /**
+ * Indicates the client has sent a close sequence
+ */
+ private boolean sequenceClosedClient = false;
+
+ private String transportTo;
public RMSBean() {
}
@@ -142,7 +154,6 @@
this.createSequenceMsgStoreKey = createSequenceMsgStoreKey;
}
-
public String getReferenceMessageStoreKey() {
return referenceMessageStoreKey;
}
@@ -159,16 +170,13 @@
this.lastSendError = lastSendError;
}
-
public long getLastSendErrorTimestamp() {
return lastSendErrorTimestamp;
}
-
public void setLastSendErrorTimestamp(long lastSendErrorTimestamp) {
this.lastSendErrorTimestamp = lastSendErrorTimestamp;
}
-
public long getLastOutMessage() {
return lastOutMessage;
@@ -182,17 +190,14 @@
return highestOutMessageNumber;
}
-
public void setHighestOutMessageNumber(long highestOutMessageNumber) {
this.highestOutMessageNumber = highestOutMessageNumber;
}
-
public String getHighestOutRelatesTo() {
return highestOutRelatesTo;
}
-
public void setHighestOutRelatesTo(String highestOutRelatesTo) {
this.highestOutRelatesTo = highestOutRelatesTo;
}
@@ -201,7 +206,6 @@
return nextMessageNumber;
}
-
public void setNextMessageNumber(long nextMessageNumber) {
this.nextMessageNumber = nextMessageNumber;
}
@@ -210,21 +214,42 @@
return clientCompletedMessages;
}
-
public void setClientCompletedMessages(List clientCompletedMessages) {
this.clientCompletedMessages = clientCompletedMessages;
}
-
public boolean isTerminateAdded() {
return terminateAdded;
}
-
public void setTerminateAdded(boolean terminateAdded) {
this.terminateAdded = terminateAdded;
}
+
+ public boolean isSequenceClosedClient() {
+ return sequenceClosedClient;
+ }
+
+ public void setSequenceClosedClient(boolean sequenceClosedClient) {
+ this.sequenceClosedClient = sequenceClosedClient;
+ }
+
+ public long getNumberOfMessagesAcked() {
+ return numberOfMessagesAcked;
+ }
+
+ public void setNumberOfMessagesAcked(long numberOfMessagesAcked) {
+ this.numberOfMessagesAcked = numberOfMessagesAcked;
+ }
+ public String getTransportTo() {
+ return transportTo;
+ }
+
+ public void setTransportTo(String transportTo) {
+ this.transportTo = transportTo;
+ }
+
public String toString() {
StringBuffer result = new StringBuffer();
result.append(this.getClass().getName());
@@ -239,7 +264,9 @@
result.append("\nHighestOutRelatesTo: ");result.append(highestOutRelatesTo);
result.append("\nNextMessageNumber: "); result.append(nextMessageNumber);
result.append("\nTerminateAdded : "); result.append(terminateAdded);
+ result.append("\nClosedClient : "); result.append(sequenceClosedClient);
+ result.append("\nNumAckedMsgs : "); result.append(numberOfMessagesAcked);
+ result.append("\nTransportTo : "); result.append(transportTo);
return result.toString();
}
-
}
Modified: webservices/sandesha/trunk/java/src/org/apache/sandesha2/storage/beans/RMSequenceBean.java
URL: http://svn.apache.org/viewvc/webservices/sandesha/trunk/java/src/org/apache/sandesha2/storage/beans/RMSequenceBean.java?view=diff&rev=495613&r1=495612&r2=495613
==============================================================================
--- webservices/sandesha/trunk/java/src/org/apache/sandesha2/storage/beans/RMSequenceBean.java (original)
+++ webservices/sandesha/trunk/java/src/org/apache/sandesha2/storage/beans/RMSequenceBean.java Fri Jan 12 07:45:34 2007
@@ -36,6 +36,8 @@
private String replyToEPR;
private String acksToEPR;
+
+ private long lastActivatedTime;
/**
* Indicates that a sequence is terminated
@@ -108,11 +110,18 @@
return terminated;
}
-
public void setTerminated(boolean terminated) {
this.terminated = terminated;
}
+ public long getLastActivatedTime() {
+ return lastActivatedTime;
+ }
+
+ public void setLastActivatedTime(long lastActivatedTime) {
+ this.lastActivatedTime = lastActivatedTime;
+ }
+
public String toString() {
StringBuffer result = new StringBuffer();
result.append("\nSequence Id: "); result.append(sequenceID);
@@ -121,6 +130,7 @@
result.append("\nacksToEPR : "); result.append(acksToEPR);
result.append("\nPolling : "); result.append(pollingMode);
result.append("\nTerminated : "); result.append(terminated);
+ result.append("\nLastActivatedTime: "); result.append(lastActivatedTime);
return result.toString();
}
}
Modified: webservices/sandesha/trunk/java/src/org/apache/sandesha2/util/SequenceManager.java
URL: http://svn.apache.org/viewvc/webservices/sandesha/trunk/java/src/org/apache/sandesha2/util/SequenceManager.java?view=diff&rev=495613&r1=495612&r2=495613
==============================================================================
--- webservices/sandesha/trunk/java/src/org/apache/sandesha2/util/SequenceManager.java (original)
+++ webservices/sandesha/trunk/java/src/org/apache/sandesha2/util/SequenceManager.java Fri Jan 12 07:45:34 2007
@@ -269,12 +269,7 @@
// saving transportTo value;
String transportTo = (String) firstAplicationMsgCtx.getProperty(Constants.Configuration.TRANSPORT_URL);
if (transportTo != null) {
- SequencePropertyBean transportToBean = new SequencePropertyBean();
- transportToBean.setSequencePropertyKey(sequencePropertyKey);
- transportToBean.setName(Sandesha2Constants.SequenceProperties.TRANSPORT_TO);
- transportToBean.setValue(transportTo);
-
- seqPropMgr.insert(transportToBean);
+ rmsBean.setTransportTo(transportTo);
}
// setting the spec version for the client side.
@@ -285,7 +280,7 @@
seqPropMgr.insert(specVerionBean);
// updating the last activated time.
- updateLastActivatedTime(sequencePropertyKey, storageManager);
+ rmsBean.setLastActivatedTime(System.currentTimeMillis());
SandeshaUtil.startSenderForTheSequence(configurationContext, sequencePropertyKey);
@@ -363,58 +358,7 @@
}
- /**
- * Takes the internalSeqID as the param. Not the sequenceID.
- *
- * @param internalSequenceID
- * @param configContext
- * @throws SandeshaException
- */
- public static void updateLastActivatedTime(String sequencePropertyKey, StorageManager storageManager)
- throws SandeshaException {
- // Transaction lastActivatedTransaction =
- // storageManager.getTransaction();
- SequencePropertyBeanMgr sequencePropertyBeanMgr = storageManager.getSequencePropertyBeanMgr();
-
- SequencePropertyBean lastActivatedBean = sequencePropertyBeanMgr.retrieve(sequencePropertyKey,
- Sandesha2Constants.SequenceProperties.LAST_ACTIVATED_TIME);
-
- boolean added = false;
-
- if (lastActivatedBean == null) {
- added = true;
- lastActivatedBean = new SequencePropertyBean();
- lastActivatedBean.setSequencePropertyKey(sequencePropertyKey);
- lastActivatedBean.setName(Sandesha2Constants.SequenceProperties.LAST_ACTIVATED_TIME);
- }
-
- long currentTime = System.currentTimeMillis();
- lastActivatedBean.setValue(Long.toString(currentTime));
-
- if (added)
- sequencePropertyBeanMgr.insert(lastActivatedBean);
- else
- sequencePropertyBeanMgr.update(lastActivatedBean);
-
- }
-
- public static long getLastActivatedTime(String propertyKey, StorageManager storageManager) throws SandeshaException {
-
- SequencePropertyBeanMgr seqPropBeanMgr = storageManager.getSequencePropertyBeanMgr();
-
- SequencePropertyBean lastActivatedBean = seqPropBeanMgr.retrieve(propertyKey,
- Sandesha2Constants.SequenceProperties.LAST_ACTIVATED_TIME);
-
- long lastActivatedTime = -1;
-
- if (lastActivatedBean != null) {
- lastActivatedTime = Long.parseLong(lastActivatedBean.getValue());
- }
-
- return lastActivatedTime;
- }
-
- public static boolean hasSequenceTimedOut(String propertyKey, RMMsgContext rmMsgCtx, StorageManager storageManager)
+ public static boolean hasSequenceTimedOut(String internalSequenceId, RMMsgContext rmMsgCtx, StorageManager storageManager)
throws SandeshaException {
// operation is the lowest level, Sandesha2 could be engaged.
@@ -426,11 +370,14 @@
boolean sequenceTimedOut = false;
- long lastActivatedTime = getLastActivatedTime(propertyKey, storageManager);
- long timeNow = System.currentTimeMillis();
- if (lastActivatedTime > 0 && (lastActivatedTime + propertyBean.getInactivityTimeoutInterval() < timeNow))
- sequenceTimedOut = true;
-
+ RMSBean rmsBean = SandeshaUtil.getRMSBeanFromInternalSequenceId(storageManager, internalSequenceId);
+
+ if (rmsBean != null) {
+ long lastActivatedTime = rmsBean.getLastActivatedTime();
+ long timeNow = System.currentTimeMillis();
+ if (lastActivatedTime > 0 && (lastActivatedTime + propertyBean.getInactivityTimeoutInterval() < timeNow))
+ sequenceTimedOut = true;
+ }
return sequenceTimedOut;
}
}
Modified: webservices/sandesha/trunk/java/src/org/apache/sandesha2/util/TerminateManager.java
URL: http://svn.apache.org/viewvc/webservices/sandesha/trunk/java/src/org/apache/sandesha2/util/TerminateManager.java?view=diff&rev=495613&r1=495612&r2=495613
==============================================================================
--- webservices/sandesha/trunk/java/src/org/apache/sandesha2/util/TerminateManager.java (original)
+++ webservices/sandesha/trunk/java/src/org/apache/sandesha2/util/TerminateManager.java Fri Jan 12 07:45:34 2007
@@ -229,9 +229,6 @@
private static boolean isPropertyDeletable(String name) {
boolean deleatable = true;
- if (Sandesha2Constants.SequenceProperties.NO_OF_OUTGOING_MSGS_ACKED.equals(name))
- deleatable = false;
-
if (Sandesha2Constants.SequenceProperties.INTERNAL_SEQUENCE_ID.equals(name))
deleatable = false;
@@ -355,10 +352,8 @@
terminateRMMessage.setWSAAction(SpecSpecificConstants.getTerminateSequenceAction(rmVersion));
terminateRMMessage.setSOAPAction(SpecSpecificConstants.getTerminateSequenceSOAPAction(rmVersion));
- SequencePropertyBean transportToBean = seqPropMgr.retrieve(sequencePropertyKey,
- Sandesha2Constants.SequenceProperties.TRANSPORT_TO);
- if (transportToBean != null) {
- terminateRMMessage.setProperty(Constants.Configuration.TRANSPORT_URL, transportToBean.getValue());
+ if (rmsBean.getTransportTo() != null) {
+ terminateRMMessage.setProperty(Constants.Configuration.TRANSPORT_URL, rmsBean.getTransportTo());
}
terminateRMMessage.addSOAPEnvelope();
@@ -366,6 +361,8 @@
String key = SandeshaUtil.getUUID();
SenderBean terminateBean = new SenderBean();
+ terminateBean.setInternalSequenceID(internalSequenceID);
+ terminateBean.setSequenceID(outSequenceId);
terminateBean.setMessageContextRefKey(key);
// Set a retransmitter lastSentTime so that terminate will be send with
Modified: webservices/sandesha/trunk/java/src/org/apache/sandesha2/util/WSRMMessageSender.java
URL: http://svn.apache.org/viewvc/webservices/sandesha/trunk/java/src/org/apache/sandesha2/util/WSRMMessageSender.java?view=diff&rev=495613&r1=495612&r2=495613
==============================================================================
--- webservices/sandesha/trunk/java/src/org/apache/sandesha2/util/WSRMMessageSender.java (original)
+++ webservices/sandesha/trunk/java/src/org/apache/sandesha2/util/WSRMMessageSender.java Fri Jan 12 07:45:34 2007
@@ -50,6 +50,7 @@
private boolean sequenceExists;
private String outSequenceID;
private String rmVersion;
+ private RMSBean rmsBean;
/**
* Extracts information from the rmMsgCtx specific for processing out messages
@@ -76,13 +77,10 @@
sequenceExists = false;
outSequenceID = null;
- // Get the Create sequence bean with the matching internal sequenceid
- RMSBean createSeqFindBean = new RMSBean();
- createSeqFindBean.setInternalSequenceID(internalSequenceID);
-
- RMSBean rMSBean = storageManager.getRMSBeanMgr().findUnique(createSeqFindBean);
+ // Get the RMSBean with the matching internal sequenceid
+ rmsBean = SandeshaUtil.getRMSBeanFromInternalSequenceId(storageManager, internalSequenceID);
- if (rMSBean == null)
+ if (rmsBean == null)
{
if (log.isDebugEnabled())
log.debug("Exit: WSRMParentProcessor::setupOutMessage Sequence doesn't exist");
@@ -91,10 +89,10 @@
SandeshaMessageKeys.couldNotSendTerminateSeqNotFound, internalSequenceID));
}
- if (rMSBean.getSequenceID() != null)
+ if (rmsBean.getSequenceID() != null)
{
sequenceExists = true;
- outSequenceID = rMSBean.getSequenceID();
+ outSequenceID = rmsBean.getSequenceID();
}
else
outSequenceID = Sandesha2Constants.TEMP_SEQUENCE_ID;
@@ -117,8 +115,7 @@
rmMsgCtx.setTo(new EndpointReference(toAddress));
- String transportTo = SandeshaUtil.getSequenceProperty(internalSequenceID,
- Sandesha2Constants.SequenceProperties.TRANSPORT_TO, storageManager);
+ String transportTo = rmsBean.getTransportTo();
if (transportTo != null) {
rmMsgCtx.setProperty(Constants.Configuration.TRANSPORT_URL, transportTo);
}
@@ -213,5 +210,9 @@
public final String getRMVersion() {
return rmVersion;
}
+
+ public final RMSBean getRMSBean() {
+ return rmsBean;
+ }
}
Modified: webservices/sandesha/trunk/java/src/org/apache/sandesha2/workers/Invoker.java
URL: http://svn.apache.org/viewvc/webservices/sandesha/trunk/java/src/org/apache/sandesha2/workers/Invoker.java?view=diff&rev=495613&r1=495612&r2=495613
==============================================================================
--- webservices/sandesha/trunk/java/src/org/apache/sandesha2/workers/Invoker.java (original)
+++ webservices/sandesha/trunk/java/src/org/apache/sandesha2/workers/Invoker.java Fri Jan 12 07:45:34 2007
@@ -76,8 +76,8 @@
InvokerBeanMgr storageMapMgr = storageManager
.getInvokerBeanMgr();
- RMDBeanMgr nextMsgMgr = storageManager.getRMDBeanMgr();
- RMDBean rMDBean = nextMsgMgr.retrieve(sequenceID);
+ RMDBeanMgr rmdBeanMgr = storageManager.getRMDBeanMgr();
+ RMDBean rMDBean = rmdBeanMgr.retrieve(sequenceID);
if (rMDBean != null) {
@@ -124,7 +124,6 @@
if(msgNumber>highestMsgNumberInvoked){
highestMsgNumberInvoked = invoker.getMsgNo();
rMDBean.setNextMsgNoToProcess(highestMsgNumberInvoked+1);
- nextMsgMgr.update(rMDBean);
if(allowLaterDeliveryOfMissingMessages){
//we also need to update the sequence OUT_OF_ORDER_RANGES property
@@ -134,26 +133,19 @@
Range r = new Range(firstMessageInOutOfOrderWindow,highestMsgNumberInvoked);
RangeString rangeString = null;
- SequencePropertyBeanMgr seqPropertyManager = storageManager.getSequencePropertyBeanMgr();
- SequencePropertyBean outOfOrderRanges =
- seqPropertyManager.retrieve(sequenceID, Sandesha2Constants.SequenceProperties.OUT_OF_ORDER_RANGES);
- if(outOfOrderRanges==null){
+ if(rMDBean.getOutOfOrderRanges()==null){
//insert a new blank one one
- outOfOrderRanges = new SequencePropertyBean(sequenceID,
- Sandesha2Constants.SequenceProperties.OUT_OF_ORDER_RANGES,
- "");
-
- seqPropertyManager.insert(outOfOrderRanges);
rangeString = new RangeString("");
}
else{
- rangeString = new RangeString(outOfOrderRanges.getValue());
+ rangeString = new RangeString(rMDBean.getOutOfOrderRanges());
}
//update the range String with the new value
rangeString.addRange(r);
- outOfOrderRanges.setValue(rangeString.toString());
- seqPropertyManager.update(outOfOrderRanges);
+ rMDBean.setOutOfOrderRanges(rangeString.toString());
}
+
+ rmdBeanMgr.update(rMDBean);
}
}
@@ -179,21 +171,19 @@
}
private void addOutOfOrderInvokerBeansToList(String sequenceID,
- StorageManager strMgr, List list)throws SandeshaException{
+ StorageManager storageManager, List list)throws SandeshaException{
if (log.isDebugEnabled())
- log.debug("Enter: InOrderInvoker::addOutOfOrderInvokerBeansToList");
+ log.debug("Enter: InOrderInvoker::addOutOfOrderInvokerBeansToList " + sequenceID + ", " + list);
- SequencePropertyBeanMgr seqPropertyManager = strMgr.getSequencePropertyBeanMgr();
+ RMDBean rmdBean = SandeshaUtil.getRMDBeanFromSequenceId(storageManager, sequenceID);
- SequencePropertyBean outOfOrderRanges =
- seqPropertyManager.retrieve(sequenceID, Sandesha2Constants.SequenceProperties.OUT_OF_ORDER_RANGES);
- if(outOfOrderRanges!=null){
- String sequenceRanges = outOfOrderRanges.getValue();
+ if(rmdBean != null && rmdBean.getOutOfOrderRanges() != null){
+ String sequenceRanges = rmdBean.getOutOfOrderRanges();
RangeString rangeString = new RangeString(sequenceRanges);
//we now have the set of ranges that can be delivered out of order.
//Look for any invokable message that lies in one of those ranges
Iterator invokerBeansIterator =
- strMgr.getInvokerBeanMgr().find(
+ storageManager.getInvokerBeanMgr().find(
new InvokerBean(null,
0, //finds all invoker beans
sequenceID)).iterator();
Modified: webservices/sandesha/trunk/java/test/src/org/apache/sandesha2/workers/ForceInboundDispatchTest.java
URL: http://svn.apache.org/viewvc/webservices/sandesha/trunk/java/test/src/org/apache/sandesha2/workers/ForceInboundDispatchTest.java?view=diff&rev=495613&r1=495612&r2=495613
==============================================================================
--- webservices/sandesha/trunk/java/test/src/org/apache/sandesha2/workers/ForceInboundDispatchTest.java (original)
+++ webservices/sandesha/trunk/java/test/src/org/apache/sandesha2/workers/ForceInboundDispatchTest.java Fri Jan 12 07:45:34 2007
@@ -9,7 +9,6 @@
import org.apache.axis2.client.ServiceClient;
import org.apache.axis2.context.ConfigurationContext;
import org.apache.axis2.context.ConfigurationContextFactory;
-import org.apache.sandesha2.Sandesha2Constants;
import org.apache.sandesha2.SandeshaException;
import org.apache.sandesha2.SandeshaTestCase;
import org.apache.sandesha2.client.SandeshaClient;
@@ -19,7 +18,6 @@
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.SequencePropertyBean;
import org.apache.sandesha2.util.RangeString;
import org.apache.sandesha2.util.SandeshaUtil;
@@ -84,14 +82,9 @@
assertNotNull(rMDBean);
assertEquals(rMDBean.getNextMsgNoToProcess(), 4);
- //also check that the sequence has an out of order gap that contains msg 2
- SequencePropertyBean outOfOrderRanges =
- serverStore.getSequencePropertyBeanMgr().retrieve(
- inboundSequenceID,
- Sandesha2Constants.SequenceProperties.OUT_OF_ORDER_RANGES);
-
- assertNotNull(outOfOrderRanges);
- RangeString rangeString = new RangeString(outOfOrderRanges.getValue());
+ //also check that the sequence has an out of order gap that contains msg 2
+ assertNotNull(rMDBean.getOutOfOrderRanges());
+ RangeString rangeString = new RangeString(rMDBean.getOutOfOrderRanges());
assertTrue(rangeString.isMessageNumberInRanges(2));
t.commit();
---------------------------------------------------------------------
To unsubscribe, e-mail: sandesha-dev-unsubscribe@ws.apache.org
For additional commands, e-mail: sandesha-dev-help@ws.apache.org