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 ch...@apache.org on 2005/02/20 18:42:39 UTC
cvs commit: ws-fx/sandesha/src/org/apache/sandesha/storage/queue IncomingSequence.java OutgoingSequence.java SandeshaQueue.java
chamikara 2005/02/20 09:42:38
Modified: sandesha/src/org/apache/sandesha Constants.java
IStorageManager.java
sandesha/src/org/apache/sandesha/client
ClientStorageManager.java
sandesha/src/org/apache/sandesha/server
MessageValidator.java ServerStorageManager.java
sandesha/src/org/apache/sandesha/storage/dao
ISandeshaDAO.java SandeshaDAOFactory.java
SandeshaDatabaseDAO.java SandeshaQueueDAO.java
sandesha/src/org/apache/sandesha/storage/queue
IncomingSequence.java OutgoingSequence.java
SandeshaQueue.java
Log:
Following methods were added to IStorageManager
hasLastOutgoingMsgReceived(String seqId);
getLastOutgoingMsgNo(String seqId);
hasLastIncomingMsgReceived(String seqId);
getLastIncomingMsgNo(String seqId);
Following methods were added to ISandeshaDAO
hasLastOutgoingMsgReceived(String seqId);
getLastOutgoingMsgNo(String seqId);
hasLastIncomingMsgReceived(String seqId);
getLastIncomingMsgNo(String seqId);
addRequestedSequence(String seqId);
isRequestedSeqPresent(String seqId);
getKeyFromIncomingSequenceId(String incomingSeqID);
getKeyFromOutgoingSequenceId(String outgoingSeqID);
Also code in the Queue classes were formatted.
Revision Changes Path
1.28 +12 -0 ws-fx/sandesha/src/org/apache/sandesha/Constants.java
Index: Constants.java
===================================================================
RCS file: /home/cvs/ws-fx/sandesha/src/org/apache/sandesha/Constants.java,v
retrieving revision 1.27
retrieving revision 1.28
diff -u -r1.27 -r1.28
--- Constants.java 20 Feb 2005 08:49:45 -0000 1.27
+++ Constants.java 20 Feb 2005 17:42:38 -0000 1.28
@@ -303,6 +303,18 @@
public static final double MAX_MSG_NO=18446744073709551615d;
public static final String UUID="uuid:";
+
+ //Constants related to the queue.
+ public interface Queue{
+ public static final String ADD_ERROR = "Error in adding message";
+ public static final String QUEUE_INCONSIS= "Inconsistent queue";
+ public static final String MESSAGE_EXISTS= "Message already exists";
+ public static final String SEQUENCE_NOTPRESENT= "Sequence not present";
+ public static final String SEQUENCE_ABSENT = "Sequence id does not exist";
+ public static final String RESPONSE_SEQ_NULL = "ERROR: RESPONSE SEQ IS NULL";
+ public static final String SEQUENCE_ID_NULL = "Sequence Id is null";
+ public static final String MESSAGE_ID_NULL = "Message is null";
+ }
}
1.16 +6 -3 ws-fx/sandesha/src/org/apache/sandesha/IStorageManager.java
Index: IStorageManager.java
===================================================================
RCS file: /home/cvs/ws-fx/sandesha/src/org/apache/sandesha/IStorageManager.java,v
retrieving revision 1.15
retrieving revision 1.16
diff -u -r1.15 -r1.16
--- IStorageManager.java 18 Feb 2005 13:00:33 -0000 1.15
+++ IStorageManager.java 20 Feb 2005 17:42:38 -0000 1.16
@@ -119,7 +119,6 @@
public boolean isAllSequenceComplete();
-
public boolean isResponseComplete(String sequenceID);
public void terminateSequence(String sequenceID);
@@ -132,9 +131,13 @@
public boolean isSentMsg(String seqId,long msgNo);
- public boolean hasLastMsgReceived(String seqId);
+ public boolean hasLastOutgoingMsgReceived(String seqId);
+
+ public long getLastOutgoingMsgNo(String seqId);
+
+ public boolean hasLastIncomingMsgReceived(String seqId);
- public long getLastMsgNo(String seqId);
+ public long getLastIncomingMsgNo(String seqId);
public void addOutgoingSequence(String seqId);
1.20 +23 -109 ws-fx/sandesha/src/org/apache/sandesha/client/ClientStorageManager.java
Index: ClientStorageManager.java
===================================================================
RCS file: /home/cvs/ws-fx/sandesha/src/org/apache/sandesha/client/ClientStorageManager.java,v
retrieving revision 1.19
retrieving revision 1.20
diff -u -r1.19 -r1.20
--- ClientStorageManager.java 20 Feb 2005 08:49:45 -0000 1.19
+++ ClientStorageManager.java 20 Feb 2005 17:42:38 -0000 1.20
@@ -49,7 +49,6 @@
*/
public void init() {
// TODO Auto-generated method stub
-
}
public ClientStorageManager() {
@@ -79,21 +78,15 @@
public void setAcknowledged(String seqID, long msgNumber) {
//seqId is just a dummy since the client will hv only a one seq.
//No hard checking. User may insert the real sequence.
-
String sequenceId = seqID;
- //accessor.moveOutgoingMessageToBin(sequenceId, new Long(msgNumber));
-
accessor.markOutgoingMessageToDelete(sequenceId, new Long(msgNumber));
}
public void addSequence(String sequenceID) {
- //boolean result = accessor.addIncomingSequence(sequenceID);
boolean result = accessor.addOutgoingSequence(sequenceID);
-
if (!result)
log.error("Sequence was not created correcly in the in queue");
-
}
/**
@@ -102,7 +95,6 @@
*/
public void addCreateSequenceResponse(RMMessageContext rmMessageContext) {
addPriorityMessage(rmMessageContext);
-
}
/**
@@ -110,7 +102,6 @@
*/
public void addCreateSequenceRequest(RMMessageContext rmMessageContext) {
addPriorityMessage(rmMessageContext);
-
}
/**
@@ -119,7 +110,6 @@
*/
public void addAcknowledgement(RMMessageContext rmMessageContext) {
addPriorityMessage(rmMessageContext);
-
}
//private method
@@ -139,37 +129,14 @@
*/
public Map getListOfMessageNumbers(String sequenceID) {
return null;
- /*
- * ISandeshaDAO accessor = SandeshaDAOFactory.getStorageAccessor(
- * Constants.SERVER_QUEUE_ACCESSOR);
- *
- * Set st = accessor.getAllReceivedMsgNumsOfIncomingSeq(sequenceID);
- *
- * Iterator it = st.iterator();
- *
- * //To find the largest id present long largest=0; while(it.hasNext()){
- * Long key = (Long) it.next(); if(key==null) continue;
- *
- * long l = key.longValue(); if(l>largest) largest = l; }
- *
- *
- * HashMap results = new HashMap(); //Add Keys to the results in order.
- * long currentPosition=1; for(long l=1;l <=largest;l++){ boolean
- * present = st.contains(new Long(l)); if(present){ results.put(new
- * Long(currentPosition),new Long(l)); currentPosition++; } } return
- * results;
- */
}
/**
* This will be used by the sender.
*/
public RMMessageContext getNextMessageToSend() {
- //System.out.println("getNextMessageToSend() is called");
RMMessageContext msg;
-
msg = accessor.getNextPriorityMessageContextToSend();
-
if (msg == null)
msg = accessor.getNextOutgoingMsgContextToSend();
@@ -185,15 +152,8 @@
* queue using this temporary ID as the messageID.
*/
public void setTemporaryOutSequence(String sequenceId, String outSequenceId) {
-
- /*if (!sequenceId.equals(Constants.CLIENT_DEFAULD_SEQUENCE_ID)) {
- System.out.println("Error: Wrong sequence id for client");
- return;
- }*/
-
accessor.setOutSequence(sequenceId, outSequenceId);
accessor.setOutSequenceApproved(sequenceId, false);
-
}
/**
@@ -207,22 +167,12 @@
String sequenceID = accessor.getSequenceOfOutSequence(oldOutsequenceId);
if (sequenceID == null) {
- System.out.println("ERROR: setApprovedOutSequence()");
+ log.error("ERROR: setApprovedOutSequence()");
return false;
}
- //sequenceid should be the default one.
- /* if (!sequenceID.equals(Constants.CLIENT_DEFAULD_SEQUENCE_ID)) {
- System.out.println("Error: Wrong sequence id for client");
- return false;
- }
- */
accessor.setOutSequence(sequenceID, newOutSequenceId);
accessor.setOutSequenceApproved(sequenceID, true);
-
- //Deleting create sequence message from the priority queue.
- //System.out.println("OLD OUT SEQ IS "+oldOutsequenceId);
accessor.removeCreateSequenceMsg(oldOutsequenceId);
-
return true;
}
@@ -233,49 +183,16 @@
public long getNextMessageNumber(String sequenceID) {
long msgNo = accessor.getNextOutgoingMessageNumber(sequenceID);
return msgNo;
-
}
- /**
- * This sholud be called by the RMSender when adding request messages.
- */
- /*private void insertClientRequestMessage(RMMessageContext msg) {
- ISandeshaDAO accessor = SandeshaDAOFactory
- .getStorageAccessor(Constants.SERVER_QUEUE_ACCESSOR);
-
- //This is the seuqnceid used to create the map entry.
- // (not the actual seq id of the msg).
- String sequenceId = Constants.CLIENT_DEFAULD_SEQUENCE_ID;
-
- boolean exists = accessor.isOutgoingSequenceExists(sequenceId);
- if (!exists)
- accessor.addOutgoingSequence(sequenceId);
-
- accessor.addMessageToOutgoingSequence(sequenceId, msg);
- }*/
-
/*
* (non-Javadoc)
*
* @see org.apache.sandesha.IStorageManager#insertOutgoingMessage(org.apache.sandesha.RMMessageContext)
*/
public void insertOutgoingMessage(RMMessageContext msg) {
- //System.out.println("RESPONSE MESSAGE IS RECEIVED..");
-
- //System.out.println("Client StorageManager is called");
-
- //This is the seuqnceid used to create the map entry.
- // (not the actual seq id of the msg).
-
- //if(sequenceId==null)
- // sequenceId = Constants.CLIENT_DEFAULD_SEQUENCE_ID;
-
- // boolean exists = accessor.isOutgoingSequenceExists(sequenceId);
- //if (!exists)
- // accessor.addOutgoingSequence(sequenceId);
String sequenceId = msg.getSequenceID();
accessor.addMessageToOutgoingSequence(sequenceId, msg);
-
}
/*
@@ -286,21 +203,14 @@
//IN THE CLIENT RESPONSE HASH HAS THE ID OF RESPONSE MESSAGES.
public void insertIncomingMessage(RMMessageContext rmMessageContext) {
-
RMHeaders rmHeaders = rmMessageContext.getRMHeaders();
-
RelatesTo relatesTo = (RelatesTo) rmMessageContext
.getAddressingHeaders().getRelatesTo().get(0);
String messageId = relatesTo.getURI().toString();
- //CHANGE THIS. SEARCH FOR THE SEQ USING MESID
- //String sequenceId = rmMessageContext.getSequenceID();
-
String sequenceId = null;
sequenceId = accessor.searchForSequenceId(messageId);
-
SandeshaQueue sq = SandeshaQueue.getInstance();
sq.displayOutgoingMap();
- System.out.println("************************************************ "+sequenceId);
boolean exists = accessor.isIncomingSequenceExists(sequenceId);
if (!exists) {
@@ -310,14 +220,11 @@
//TODO: add getRmHeaders method to MessageContext
long messageNumber = rmHeaders.getSequence().getMessageNumber()
.getMessageNumber();
-
if (messageNumber <= 0)
return; //TODO: throw some exception
-
Long msgNo = new Long(messageNumber);
accessor.addMessageToIncomingSequence(sequenceId, msgNo,
rmMessageContext);
-
}
@@ -382,12 +289,10 @@
*/
public void terminateSequence(String sequenceID) {
// TODO Auto-generated method stub
-
}
public void setAckReceived(String seqId, long msgNo) {
accessor.setAckReceived(seqId, msgNo);
-
}
public void insertFault(RMMessageContext rmMsgCtx) {
@@ -403,16 +308,8 @@
* @see org.apache.sandesha.IStorageManager#addSendMsgNo(java.lang.String, long)
*/
public void addSendMsgNo(String seqId, long msgNo) {
- System.out.println("2222222222222222222222222222222222222222222222222222");
System.out.println( accessor.getSequenceOfOutSequence(seqId));
accessor.addSendMsgNo(accessor.getSequenceOfOutSequence(seqId),msgNo);
-
- }
- /* (non-Javadoc)
- * @see org.apache.sandesha.IStorageManager#getLastMsgNo(java.lang.String)
- */
- public long getLastMsgNo(String seqId) {
- return accessor.getLastMsgNo(seqId);
}
public void addOutgoingSequence(String sequenceId) {
@@ -426,10 +323,29 @@
/* (non-Javadoc)
* @see org.apache.sandesha.IStorageManager#hasLastMsgReceived(java.lang.String)
*/
- public boolean hasLastMsgReceived(String seqId) {
- return accessor.hasLastMsgReceived(seqId);
+ public boolean hasLastOutgoingMsgReceived(String seqId) {
+ String key = accessor.getKeyFromOutgoingSequenceId(seqId);
+ return accessor.hasLastOutgoingMsgReceived(key);
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.sandesha.IStorageManager#getLastMsgNo(java.lang.String)
+ */
+ public long getLastOutgoingMsgNo(String seqId) {
+ String key = accessor.getKeyFromOutgoingSequenceId(seqId);
+ return accessor.getLastOutgoingMsgNo(key);
}
+
+ public long getLastIncomingMsgNo(String seqId) {
+ String key = accessor.getKeyFromIncomingSequenceId(seqId);
+ return accessor.getLastIncomingMsgNo(key);
+ }
+ public boolean hasLastIncomingMsgReceived(String seqId) {
+ String key = accessor.getKeyFromIncomingSequenceId(seqId);
+ return accessor.hasLastIncomingMsgReceived(key);
+ }
+
public void addRequestedSequence(String seqId) {
accessor.addRequestedSequence(seqId);
}
@@ -438,9 +354,6 @@
return accessor.isRequestedSeqPresent(seqId);
}
- /* (non-Javadoc)
- * @see org.apache.sandesha.IStorageManager#isSentMsg(java.lang.String, long)
- */
public boolean isSentMsg(String seqId, long msgNo) {
return accessor.isSentMsg(accessor.getSequenceOfOutSequence(seqId) ,msgNo);
}
@@ -454,4 +367,5 @@
return accessor.searchForSequenceId(msgId);
}
+
}
\ No newline at end of file
1.7 +2 -2 ws-fx/sandesha/src/org/apache/sandesha/server/MessageValidator.java
Index: MessageValidator.java
===================================================================
RCS file: /home/cvs/ws-fx/sandesha/src/org/apache/sandesha/server/MessageValidator.java,v
retrieving revision 1.6
retrieving revision 1.7
diff -u -r1.6 -r1.7
--- MessageValidator.java 20 Feb 2005 08:49:45 -0000 1.6
+++ MessageValidator.java 20 Feb 2005 17:42:38 -0000 1.7
@@ -73,8 +73,8 @@
}
if (sequence.getMessageNumber() != null) {
long msgNo = sequence.getMessageNumber().getMessageNumber();
- if (storageMgr.hasLastMsgReceived(sequence.getIdentifier().getIdentifier())) {
- long lastMsg = storageMgr.getLastMsgNo(seqId);
+ if (storageMgr.hasLastIncomingMsgReceived(sequence.getIdentifier().getIdentifier())) {
+ long lastMsg = storageMgr.getLastIncomingMsgNo(seqId);
if (msgNo > lastMsg)
throw new AxisFault(new QName(Constants.FaultCodes.WSRM_FAULR_LAST_MSG_NO_EXCEEDED), Constants.FaultMessages.LAST_MSG_NO_EXCEEDED, null, null);
}
1.20 +18 -56 ws-fx/sandesha/src/org/apache/sandesha/server/ServerStorageManager.java
Index: ServerStorageManager.java
===================================================================
RCS file: /home/cvs/ws-fx/sandesha/src/org/apache/sandesha/server/ServerStorageManager.java,v
retrieving revision 1.19
retrieving revision 1.20
diff -u -r1.19 -r1.20
--- ServerStorageManager.java 20 Feb 2005 08:49:45 -0000 1.19
+++ ServerStorageManager.java 20 Feb 2005 17:42:38 -0000 1.20
@@ -39,9 +39,7 @@
protected static Log log = LogFactory.getLog(ServerStorageManager.class
.getName());
-
private String tempSeqId = null; // used by getNextMessageToProcess();
-
ISandeshaDAO accessor;
public ServerStorageManager() {
@@ -76,7 +74,6 @@
public void setAcknowledged(String seqID, long msgNumber) {
//TODO decide this in implementing the ServerSender.
-
accessor.moveOutgoingMessageToBin(seqID, new Long(msgNumber));
}
@@ -103,39 +100,18 @@
*/
public RMMessageContext getNextMessageToSend() {
RMMessageContext msg;
-
msg = accessor.getNextPriorityMessageContextToSend();
-
if (msg == null)
msg = accessor.getNextOutgoingMsgContextToSend();
-
- //if(msg==null){
- // msg = accessor.getNextLowPriorityMessageContextToSend(); // checks whether all the request messages hv been acked
- //}
-
return msg;
}
/**
- * This is used to add a new message to the out queue. Will be used by
- * various processors.
- */
- /*
- * public void addMessageToOutQueue(RMMessageContext rmMessageContext) {
- * ISandeshaDAO accessor = SandeshaDAOFactory.getStorageAccessor(
- * Constants.SERVER_QUEUE_ACCESSOR); boolean result =
- * accessor.addOutQueueMessage(rmMessageContext);
- *
- * if(!result) log.error("Message was not added to the out queue"); }
- */
-
- /**
* Will be used to add a new Sequence Hash to the In Queue.
*/
public void addSequence(String sequenceId) {
boolean result = accessor.addIncomingSequence(sequenceId);
-
if (!result)
log.error("Sequence was not created correcly in the in queue");
}
@@ -146,9 +122,7 @@
*/
public Map getListOfMessageNumbers(String sequenceID) {
Set st = accessor.getAllReceivedMsgNumsOfIncomingSeq(sequenceID);
-
Iterator it = st.iterator();
-
//To find the largest id present
long largest = 0;
while (it.hasNext()) {
@@ -251,24 +225,15 @@
boolean exists = accessor.isOutgoingSequenceExists(sequenceId);
if (!exists)
accessor.addOutgoingSequence(sequenceId);
-
accessor.addMessageToOutgoingSequence(sequenceId, msg);
}
public void insertIncomingMessage(RMMessageContext rmMessageContext) {
- //No need to use this property
- //RMHeaders rmHeaders =
- // (RMHeaders) rmMessageContext.getMsgContext().getProperty(
- // org.apache.sandesha.Constants.ENV_RM_REQUEST_HEADERS);
-
RMHeaders rmHeaders = rmMessageContext.getRMHeaders();
-
String sequenceId = rmHeaders.getSequence().getIdentifier()
.getIdentifier();
-
boolean exists = accessor.isIncomingSequenceExists(sequenceId);
-
if (!exists)
addSequence(sequenceId); //Creating new sequence
@@ -365,32 +330,12 @@
// TODO Auto-generated method stub
return 0;
}
- /* (non-Javadoc)
- * @see org.apache.sandesha.IStorageManager#hasLastMsgNo(java.lang.String)
- */
- public boolean hasLastMsgNo(String seq) {
- // TODO Auto-generated method stub
- return false;
- }
-
- /* (non-Javadoc)
- * @see org.apache.sandesha.IStorageManager#addSendMsgNo(java.lang.String, long)
- */
public void addSendMsgNo(String seqId, long msgNo) {
accessor.addSendMsgNo(accessor.getSequenceOfOutSequence(seqId),msgNo);
}
- /* (non-Javadoc)
- * @see org.apache.sandesha.IStorageManager#hasLastMsgReceived(java.lang.String)
- */
- public boolean hasLastMsgReceived(String seqId) {
- // TODO Auto-generated method stub
- return false;
- }
- /* (non-Javadoc)
- * @see org.apache.sandesha.IStorageManager#isSentMsg(java.lang.String, long)
- */
+
public boolean isSentMsg(String seqId, long msgNo) {
return accessor.isSentMsg(accessor.getSequenceOfOutSequence(seqId) ,msgNo);
@@ -421,4 +366,21 @@
return msg.getSequenceID();
}
+
+ public boolean hasLastOutgoingMsgReceived(String seqId) {
+ return accessor.hasLastOutgoingMsgReceived(seqId);
+ }
+
+ public long getLastOutgoingMsgNo(String seqId) {
+ return accessor.getLastOutgoingMsgNo(seqId);
+ }
+
+ public long getLastIncomingMsgNo(String seqId) {
+ return accessor.getLastIncomingMsgNo(seqId);
+ }
+
+ public boolean hasLastIncomingMsgReceived(String seqId) {
+ return accessor.hasLastIncomingMsgReceived(seqId);
+ }
+
}
\ No newline at end of file
1.4 +16 -9 ws-fx/sandesha/src/org/apache/sandesha/storage/dao/ISandeshaDAO.java
Index: ISandeshaDAO.java
===================================================================
RCS file: /home/cvs/ws-fx/sandesha/src/org/apache/sandesha/storage/dao/ISandeshaDAO.java,v
retrieving revision 1.3
retrieving revision 1.4
diff -u -r1.3 -r1.4
--- ISandeshaDAO.java 18 Feb 2005 13:00:34 -0000 1.3
+++ ISandeshaDAO.java 20 Feb 2005 17:42:38 -0000 1.4
@@ -49,11 +49,7 @@
RMMessageContext getNextMsgContextToProcess(String sequenceId);
- //RMMessageContext getNextResponseMsgContextToSend(String sequenceId);
- //String getRandomResponseSeqIdToSend();
- RMMessageContext getNextOutgoingMsgContextToSend(); //Used this instead of
-
- // above 2.
+ RMMessageContext getNextOutgoingMsgContextToSend();
String getRandomSeqIdToProcess();
@@ -61,8 +57,6 @@
Set getAllReceivedMsgNumsOfOutgoingSeq(String sequenceId);
- // boolean hasNewMessages();
-
void setOutSequence(String sequenceId, String outSequenceId);
void setOutSequenceApproved(String sequenceID, boolean approved);
@@ -97,11 +91,24 @@
public boolean isSentMsg(String seqId,long msgNo);
- public boolean hasLastMsgReceived(String seqId);
+ public boolean hasLastOutgoingMsgReceived(String seqId);
+
+ public long getLastOutgoingMsgNo(String seqId);
+
+ public boolean hasLastIncomingMsgReceived(String seqId);
- public long getLastMsgNo(String seqId);
+ public long getLastIncomingMsgNo(String seqId);
public void addRequestedSequence(String seqId);
public boolean isRequestedSeqPresent(String seqId);
+
+ //The client side will not be able to have sequenceId as a key for storing request messages.
+ //Since it may be not known when the user adds first message.
+ //Two methods below will be used to get this key from the actual sequenceid.
+ public String getKeyFromIncomingSequenceId(String incomingSeqID);
+
+ public String getKeyFromOutgoingSequenceId(String outgoingSeqID);
+
+
}
\ No newline at end of file
1.2 +5 -0 ws-fx/sandesha/src/org/apache/sandesha/storage/dao/SandeshaDAOFactory.java
Index: SandeshaDAOFactory.java
===================================================================
RCS file: /home/cvs/ws-fx/sandesha/src/org/apache/sandesha/storage/dao/SandeshaDAOFactory.java,v
retrieving revision 1.1
retrieving revision 1.2
diff -u -r1.1 -r1.2
--- SandeshaDAOFactory.java 16 Feb 2005 08:03:13 -0000 1.1
+++ SandeshaDAOFactory.java 20 Feb 2005 17:42:38 -0000 1.2
@@ -16,7 +16,10 @@
*/
package org.apache.sandesha.storage.dao;
+import org.apache.axis.components.logger.LogFactory;
+import org.apache.commons.logging.Log;
import org.apache.sandesha.Constants;
+import org.apache.sandesha.storage.queue.IncomingSequence;
/**
* @author Chamikara Jayalath
@@ -25,6 +28,8 @@
public class SandeshaDAOFactory {
+ private static final Log log = LogFactory.getLog(SandeshaDAOFactory.class.getName());
+
public static ISandeshaDAO getStorageAccessor(int accessor) {
if (accessor == Constants.SERVER_QUEUE_ACCESSOR)
1.4 +54 -17 ws-fx/sandesha/src/org/apache/sandesha/storage/dao/SandeshaDatabaseDAO.java
Index: SandeshaDatabaseDAO.java
===================================================================
RCS file: /home/cvs/ws-fx/sandesha/src/org/apache/sandesha/storage/dao/SandeshaDatabaseDAO.java,v
retrieving revision 1.3
retrieving revision 1.4
diff -u -r1.3 -r1.4
--- SandeshaDatabaseDAO.java 18 Feb 2005 13:00:34 -0000 1.3
+++ SandeshaDatabaseDAO.java 20 Feb 2005 17:42:38 -0000 1.4
@@ -16,7 +16,10 @@
*/
package org.apache.sandesha.storage.dao;
+import org.apache.axis.components.logger.LogFactory;
+import org.apache.commons.logging.Log;
import org.apache.sandesha.RMMessageContext;
+import org.apache.sandesha.storage.queue.IncomingSequence;
import java.util.Set;
@@ -27,9 +30,10 @@
public class SandeshaDatabaseDAO implements ISandeshaDAO {
+ private static final Log log = LogFactory.getLog(SandeshaDatabaseDAO.class.getName());
+
public void setAckReceived(String seqId, long msgNo) {
// TODO Auto-generated method stub
-
}
/* (non-Javadoc)
@@ -37,7 +41,6 @@
*/
public void setResponseReceived(RMMessageContext msg) {
// TODO Auto-generated method stub
-
}
public void addRequestedSequence(String seqId) {
@@ -53,7 +56,6 @@
*/
public void markOutgoingMessageToDelete(String seqId, Long msgNo) {
// TODO Auto-generated method stub
-
}
public boolean addIncomingSequence(String sequenceId) {
@@ -328,20 +330,37 @@
// TODO Auto-generated method stub
}
- /* (non-Javadoc)
- * @see org.apache.sandesha.storage.dao.ISandeshaDAO#getLastMsgNo(java.lang.String)
- */
- public long getLastMsgNo(String seqId) {
- // TODO Auto-generated method stub
- return 0;
- }
- /* (non-Javadoc)
- * @see org.apache.sandesha.storage.dao.ISandeshaDAO#hasLastMsgReceived(java.lang.String)
- */
- public boolean hasLastMsgReceived(String seqId) {
- // TODO Auto-generated method stub
- return false;
- }
+
+
+
+ /* (non-Javadoc)
+ * @see org.apache.sandesha.storage.dao.ISandeshaDAO#getLastIncomingMsgNo(java.lang.String)
+ */
+ public long getLastIncomingMsgNo(String seqId) {
+ // TODO Auto-generated method stub
+ return 0;
+ }
+ /* (non-Javadoc)
+ * @see org.apache.sandesha.storage.dao.ISandeshaDAO#getLastOutgoingMsgNo(java.lang.String)
+ */
+ public long getLastOutgoingMsgNo(String seqId) {
+ // TODO Auto-generated method stub
+ return 0;
+ }
+ /* (non-Javadoc)
+ * @see org.apache.sandesha.storage.dao.ISandeshaDAO#hasLastIncomingMsgReceived(java.lang.String)
+ */
+ public boolean hasLastIncomingMsgReceived(String seqId) {
+ // TODO Auto-generated method stub
+ return false;
+ }
+ /* (non-Javadoc)
+ * @see org.apache.sandesha.storage.dao.ISandeshaDAO#hasLastOutgoingMsgReceived(java.lang.String)
+ */
+ public boolean hasLastOutgoingMsgReceived(String seqId) {
+ // TODO Auto-generated method stub
+ return false;
+ }
/* (non-Javadoc)
* @see org.apache.sandesha.storage.dao.ISandeshaDAO#isSentMsg(java.lang.String, long)
*/
@@ -349,4 +368,22 @@
// TODO Auto-generated method stub
return false;
}
+
+
+ /* (non-Javadoc)
+ * @see org.apache.sandesha.storage.dao.ISandeshaDAO#getKeyFromIncomingSequenceId(java.lang.String)
+ */
+ public String getKeyFromIncomingSequenceId(String seqID) {
+ // TODO Auto-generated method stub
+ return null;
+ }
+
+
+ /* (non-Javadoc)
+ * @see org.apache.sandesha.storage.dao.ISandeshaDAO#getKeyFromOutgoingSequenceId(java.lang.String)
+ */
+ public String getKeyFromOutgoingSequenceId(String seqID) {
+ // TODO Auto-generated method stub
+ return null;
+ }
}
\ No newline at end of file
1.5 +29 -72 ws-fx/sandesha/src/org/apache/sandesha/storage/dao/SandeshaQueueDAO.java
Index: SandeshaQueueDAO.java
===================================================================
RCS file: /home/cvs/ws-fx/sandesha/src/org/apache/sandesha/storage/dao/SandeshaQueueDAO.java,v
retrieving revision 1.4
retrieving revision 1.5
diff -u -r1.4 -r1.5
--- SandeshaQueueDAO.java 20 Feb 2005 08:49:45 -0000 1.4
+++ SandeshaQueueDAO.java 20 Feb 2005 17:42:38 -0000 1.5
@@ -34,8 +34,7 @@
public class SandeshaQueueDAO implements ISandeshaDAO {
- protected static Log log = LogFactory
- .getLog(SandeshaQueueDAO.class.getName());
+ private static final Log log = LogFactory.getLog(SandeshaQueueDAO.class.getName());
public boolean addIncomingSequence(String sequenceId) {
boolean result = false;
@@ -46,7 +45,6 @@
} catch (QueueException e) {
log.error(e);
}
-
return result;
}
@@ -58,7 +56,6 @@
} catch (QueueException e) {
log.error(e);
}
-
return result;
}
@@ -71,7 +68,6 @@
log.error(e);
e.printStackTrace();
}
-
return msg;
}
@@ -83,7 +79,6 @@
*/
public boolean addMessageToIncomingSequence(String sequenceId, Long msgNo,
RMMessageContext rmMessageContext) {
-
boolean result = false;
try {
SandeshaQueue sq = SandeshaQueue.getInstance();
@@ -103,10 +98,8 @@
* @see org.apache.sandesha.storage.dao.ISandeshaDAO#isSequenceExists(java.lang.String)
*/
public boolean isIncomingSequenceExists(String sequenceId) {
-
SandeshaQueue sq = SandeshaQueue.getInstance();
boolean exists = sq.isIncomingSequenceExists(sequenceId);
-
return exists;
}
@@ -117,7 +110,6 @@
*/
public RMMessageContext getNextMsgContextToProcess(String sequenceId) {
SandeshaQueue sq = SandeshaQueue.getInstance();
-
RMMessageContext msg = null;
try {
msg = sq.nextIncomingMessageToProcess(sequenceId);
@@ -125,7 +117,6 @@
log.error(e);
e.printStackTrace();
}
-
return msg;
}
@@ -148,16 +139,12 @@
// TODO Auto-generated method stub
SandeshaQueue sq = SandeshaQueue.getInstance();
Vector ids = sq.nextAllSeqIdsToProcess();
-
int size = ids.size();
-
if (size <= 0)
return null;
-
Random r = new Random();
int number = r.nextInt(size);
String id = (String) ids.get(number);
-
sq.setSequenceLock(id, true); //Locks the sequence temporarily to tell
// that it is locked.
return id;
@@ -169,7 +156,6 @@
* @see org.apache.sandesha.storage.dao.ISandeshaDAO#getAllReceivedMsgNumsOfSeq(java.lang.String)
*/
public Set getAllReceivedMsgNumsOfIncomingSeq(String sequenceId) {
-
SandeshaQueue sq = SandeshaQueue.getInstance();
return sq.getAllReceivedMsgNumsOfIncomingSeq(sequenceId);
}
@@ -181,7 +167,6 @@
* java.lang.String)
*/
public boolean isIncomingMessageExists(String sequenceId, Long msgNo) {
-
SandeshaQueue sq = SandeshaQueue.getInstance();
return sq.isIncomingMessageExists(sequenceId, msgNo);
}
@@ -194,7 +179,6 @@
*/
public boolean addMessageToOutgoingSequence(String sequenceId,
RMMessageContext rmMessageContext) {
-
boolean result = false;
try {
SandeshaQueue sq = SandeshaQueue.getInstance();
@@ -221,7 +205,6 @@
} catch (QueueException e) {
log.error(e);
}
-
return result;
}
@@ -231,34 +214,14 @@
* @see org.apache.sandesha.storage.dao.ISandeshaDAO#isResponseSequenceExists(java.lang.String)
*/
public boolean isOutgoingSequenceExists(String sequenceId) {
-
SandeshaQueue sq = SandeshaQueue.getInstance();
boolean exists = sq.isOutgoingSequenceExists(sequenceId);
-
return exists;
}
/*
* (non-Javadoc)
*
- * @see org.apache.sandesha.storage.dao.ISandeshaDAO#getNextResponseMsgContextToProcess(java.lang.String)
- */
- /*
- * public RMMessageContext getNextResponseMsgContextToSend(String
- * sequenceId) {
- *
- * SandeshaQueue sq = SandeshaQueue.getInstance();
- *
- * RMMessageContext msg = null; try{ msg =
- * sq.nextResponseMessageToSend(sequenceId); }catch(Exception e){
- * log.error(e); e.printStackTrace(); }
- *
- * return msg; }
- */
-
- /*
- * (non-Javadoc)
- *
* @see org.apache.sandesha.storage.dao.ISandeshaDAO#getAllReceivedMsgNumsOfResponseSeq(java.lang.String)
*/
public Set getAllReceivedMsgNumsOfOutgoingSeq(String sequenceId) {
@@ -269,27 +232,6 @@
/*
* (non-Javadoc)
*
- * @see org.apache.sandesha.storage.dao.ISandeshaDAO#getRandomResponseSeqIdToSend()
- */
- /*
- * public String getRandomResponseSeqIdToSend() { // TODO Auto-generated
- * method stub SandeshaQueue sq = SandeshaQueue.getInstance(); Vector ids =
- * sq.nextAllResponseSeqIdsToSend();
- *
- * int size = ids.size();
- *
- * if(size <=0) return null;
- *
- * Random r = new Random(); int number = r.nextInt(size); String id =
- * (String) ids.get(number);
- *
- * sq.setSequenceLock(id,true); //Locks the sequence temporarily to tell
- * that it is locked. return id; }
- */
-
- /*
- * (non-Javadoc)
- *
* @see org.apache.sandesha.storage.dao.ISandeshaDAO#getNextResponseMsgContextToSend()
*/
public RMMessageContext getNextOutgoingMsgContextToSend() {
@@ -306,7 +248,6 @@
public void setOutSequence(String seqId, String outSeqId) {
SandeshaQueue sq = SandeshaQueue.getInstance();
sq.setOutSequence(seqId, outSeqId);
-
}
public void setOutSequenceApproved(String seqId, boolean approved) {
@@ -335,7 +276,6 @@
// TODO Auto-generated method stub
SandeshaQueue sq = SandeshaQueue.getInstance();
sq.moveOutgoingMsgToBin(sequenceId, msgNo);
-
}
/*
@@ -347,7 +287,6 @@
// TODO Auto-generated method stub
SandeshaQueue sq = SandeshaQueue.getInstance();
sq.movePriorityMsgToBin(messageId);
-
}
@@ -406,13 +345,10 @@
boolean result = true;
for (int i = 0; i < outGoingMsgs.size(); i++) {
if (!acks.contains(outGoingMsgs.get(i))) {
-
- //System.out.println("result false "+outGoingMsgs.get(i));
result = false;
break;
}
}
-
return result;
}
@@ -421,13 +357,11 @@
sq.setResponseReceived(msg);
}
-
/* (non-Javadoc)
* @see org.apache.sandesha.storage.dao.ISandeshaDAO#addLowPriorityMessage(org.apache.sandesha.RMMessageContext)
*/
public void addLowPriorityMessage(RMMessageContext msg) {
SandeshaQueue sq = SandeshaQueue.getInstance();
-
try {
sq.addLowPriorityMessage(msg);
} catch (QueueException e) {
@@ -441,7 +375,6 @@
* @see org.apache.sandesha.storage.dao.ISandeshaDAO#getNextLowPriorityMessageContextToSend()
*/
public RMMessageContext getNextLowPriorityMessageContextToSend() {
-
SandeshaQueue sq = SandeshaQueue.getInstance();
try {
return sq.getLowPriorityMessageIfAcked();
@@ -467,16 +400,26 @@
return sq.isSentMsg(seqId,msgNo);
}
- public boolean hasLastMsgReceived(String seqId){
+ public boolean hasLastOutgoingMsgReceived(String seqId){
SandeshaQueue sq = SandeshaQueue.getInstance();
- return sq.hasLastMsgReceived(seqId);
+ return sq.hasLastOutgoingMsgReceived(seqId);
}
- public long getLastMsgNo(String seqId){
+ public long getLastOutgoingMsgNo(String seqId){
SandeshaQueue sq = SandeshaQueue.getInstance();
- return sq.getLastMsgNo(seqId);
+ return sq.getLastOutgoingMsgNo(seqId);
}
+ public boolean hasLastIncomingMsgReceived(String seqId){
+ SandeshaQueue sq = SandeshaQueue.getInstance();
+ return sq.hasLastIncomingMsgReceived(seqId);
+ }
+
+ public long getLastIncomingMsgNo(String seqId){
+ SandeshaQueue sq = SandeshaQueue.getInstance();
+ return sq.getLastIncomingMsgNo(seqId);
+ }
+
public void addRequestedSequence(String seqId) {
SandeshaQueue sq = SandeshaQueue.getInstance();
sq.addRequestedSequence(seqId);
@@ -486,4 +429,18 @@
SandeshaQueue sq = SandeshaQueue.getInstance();
return sq.isRequestedSeqPresent(seqId);
}
+
+
+ /* (non-Javadoc)
+ * @see org.apache.sandesha.storage.dao.ISandeshaDAO#getKeyFromIncomingSequenceId(java.lang.String)
+ */
+ public String getKeyFromIncomingSequenceId(String seqID) {
+ SandeshaQueue sq = SandeshaQueue.getInstance();
+ return sq.getKeyFromIncomingSequenceId(seqID);
+ }
+
+ public String getKeyFromOutgoingSequenceId(String seqID) {
+ SandeshaQueue sq = SandeshaQueue.getInstance();
+ return sq.getKeyFromOutgoingSequenceId(seqID);
+ }
}
1.2 +26 -40 ws-fx/sandesha/src/org/apache/sandesha/storage/queue/IncomingSequence.java
Index: IncomingSequence.java
===================================================================
RCS file: /home/cvs/ws-fx/sandesha/src/org/apache/sandesha/storage/queue/IncomingSequence.java,v
retrieving revision 1.1
retrieving revision 1.2
diff -u -r1.1 -r1.2
--- IncomingSequence.java 18 Feb 2005 13:00:34 -0000 1.1
+++ IncomingSequence.java 20 Feb 2005 17:42:38 -0000 1.2
@@ -17,9 +17,11 @@
package org.apache.sandesha.storage.queue;
-
+import org.apache.axis.components.logger.LogFactory;
import org.apache.axis.message.addressing.RelatesTo;
+import org.apache.commons.logging.Log;
import org.apache.sandesha.RMMessageContext;
+import org.apache.sandesha.storage.dao.SandeshaQueueDAO;
import java.util.*;
@@ -35,22 +37,16 @@
public class IncomingSequence {
private long lastProcessed;
-
private boolean hasProcessableMessages;
-
private String sequenceId;
-
- //private String outSequenceId;
private HashMap hash;
-
private boolean beingProcessedLock = false; //When true messages are
-
-
- // currently being processed.
+ private long lastMsgNo = -1;
+
+ private static final Log log = LogFactory.getLog(IncomingSequence.class.getName());
public IncomingSequence(String sequenceId) {
lastProcessed = 0;
- //cacheBottom = 1;
hasProcessableMessages = false;
this.sequenceId = sequenceId;
hash = new HashMap();
@@ -72,30 +68,23 @@
* adds the message to map. Also adds a record to cache if needed.
*/
public Object putNewMessage(Long key, RMMessageContext value) {
-
Object obj = hash.put(key, value);
- //addToCacheIfNeeded(key);
refreshHasProcessableMessages();
return obj;
}
public boolean removeMessage(long msgId) {
- //TODO: Add messageremoving code if needed.
boolean removed = false;
-
Long key = new Long(msgId);
Object obj = hash.remove(key);
if (obj != null)
removed = true;
-
return removed;
}
public long getNextMessageIdToProcess() {
-
long id = lastProcessed + 1;
-
return id;
}
@@ -106,8 +95,7 @@
incrementProcessedCount();
refreshHasProcessableMessages();
} else {
- setProcessLock(false); // Not a must. (det done in
- // refreshHasProcessableMessages();)
+ setProcessLock(false);
}
return msg;
@@ -131,7 +119,6 @@
}
}
refreshHasProcessableMessages();
-
return messages;
}
@@ -144,20 +131,18 @@
hasProcessableMessages = hash.containsKey(nextKey);
if (!hasProcessableMessages) //Cant be being procesed if no messages to
- // process.
+ // process.
setProcessLock(false);
}
public boolean hasMessage(Long msgId) {
Object obj = hash.get(msgId);
-
return (!(obj == null));
}
public void clearSequence(boolean yes) {
if (!yes)
return;
-
hash.clear();
lastProcessed = 0;
hasProcessableMessages = false;
@@ -181,7 +166,6 @@
return null;
return msg.getMessageID();
-
}
//Only for client.
@@ -204,18 +188,6 @@
}
return msgToSend;
-
- /*Long nextKey = new Long(lastProcessed + 1);
- RMMessageContext msg = (RMMessageContext) hash.get(nextKey);
- if (msg != null) {
- incrementProcessedCount();
- refreshHasProcessableMessages();
- } else {
- setProcessLock(false); // Not a must. (det done in
- // refreshHasProcessableMessages();)
- }
-
- return msg;*/
}
public boolean isMessagePresent(String msgId) {
@@ -226,8 +198,6 @@
}
public boolean hasMessageWithId(String msgId) {
- //boolean b = false;
-
Iterator it = hash.keySet().iterator();
while (it.hasNext()) {
@@ -237,7 +207,23 @@
}
return false;
}
-
-
+ public boolean hasLastMsgReceived() {
+ if (lastMsgNo > 0)
+ return true;
+
+ return false;
+ }
+
+ public long getLastMsgNumber() {
+ if (lastMsgNo > 0)
+ return lastMsgNo;
+
+ return -1;
+ }
+
+ public void setLastMsg(long lastMsg) {
+ lastMsgNo = lastMsg;
+ }
+
}
\ No newline at end of file
1.2 +29 -115 ws-fx/sandesha/src/org/apache/sandesha/storage/queue/OutgoingSequence.java
Index: OutgoingSequence.java
===================================================================
RCS file: /home/cvs/ws-fx/sandesha/src/org/apache/sandesha/storage/queue/OutgoingSequence.java,v
retrieving revision 1.1
retrieving revision 1.2
diff -u -r1.1 -r1.2
--- OutgoingSequence.java 18 Feb 2005 13:00:34 -0000 1.1
+++ OutgoingSequence.java 20 Feb 2005 17:42:38 -0000 1.2
@@ -17,12 +17,12 @@
package org.apache.sandesha.storage.queue;
+import org.apache.axis.components.logger.LogFactory;
+import org.apache.commons.logging.Log;
import org.apache.sandesha.Constants;
import org.apache.sandesha.RMMessageContext;
-
import java.util.*;
-
/*
* Created on Aug 4, 2004 at 5:08:29 PM
*/
@@ -38,40 +38,23 @@
*/
public class OutgoingSequence {
- //following concepts was removed from highPriorityQueue.
- //This was to reduce complexity. (Since here time is also taken into
- // account)
- //private long lastProcessed;
- //private boolean hasMessagesToSend;
-
private String sequenceId;
-
private String outSequenceId;
-
private boolean outSeqApproved;
-
private HashMap hash;
-
private Vector markedAsDelete;
-
private Vector sendMsgNoList;
-
private long lastMsgNo = -1;
-
private long nextAutoNumber; // key for storing messages.
- //--> USING AUTONUMBER FOR MESSAGENUMBERS
- // private long nextMessageNumber;
-
+ private static final Log log = LogFactory.getLog(OutgoingSequence.class.getName());
+
public OutgoingSequence(String sequenceId) {
- //lastProcessed = 0;
- //hasMessagesToSend = false;
this.sequenceId = sequenceId;
hash = new HashMap();
markedAsDelete = new Vector();
nextAutoNumber = 1; //This is the key for storing messages.
outSeqApproved = false;
-
- sendMsgNoList = new Vector ();
+ sendMsgNoList = new Vector();
}
/*
@@ -109,7 +92,6 @@
Long key = new Long(nextAutoNumber);
Object obj = hash.put(key, msg);
increaseAutoNo();
- //refreshHasProcessableMessages();
return obj;
}
@@ -132,52 +114,26 @@
if (obj != null)
removed = true;
-
return removed;
}
/**
- * Returns the key of the next message to be sent.
- */
- /*
- * public long getNextMessageKeyToSend(){
- *
- * long id = lastProcessed+1;
- *
- * return id; }
- */
-
- /**
* Returns the next deliverable message if has any. Otherwise returns null.
*/
public RMMessageContext getNextMessageToSend() {
- //Long nextKey = new Long (lastProcessed+1);
- //RMMessageContext msg = (RMMessageContext) hash.get(nextKey);
-
- //RMMessageContext msg = null;
RMMessageContext minMsg = null;
Iterator keys = hash.keySet().iterator();
whileLoop: while (keys.hasNext()) {
RMMessageContext tempMsg;
tempMsg = (RMMessageContext) hash.get(keys.next());
-
- //EDITED FOR MARKASDELETE
Long msgNo = new Long(tempMsg.getMsgNumber());
if (markedAsDelete.contains(msgNo)) {
- System.out.println("mark as delete contains " + msgNo);
continue;
}
- //System.out.println("mark as delete does not contains "+msgNo);
-
- //END - EDITED FOR MARKASDELETE
-
long lastSentTime = tempMsg.getLastSentTime();
- //System.out.println("last sent time: "+lastSentTime);
Date d = new Date();
long currentTime = d.getTime();
- //System.out.println("current time: "+currentTime);
- // System.out.println("difference: "+(currentTime-lastSentTime));
if (currentTime >= lastSentTime + Constants.RETRANSMISSION_INTERVAL) {
if (minMsg == null)
minMsg = tempMsg;
@@ -200,31 +156,6 @@
return minMsg;
}
- /**
- * Gives all the deliverable messages of this sequence. Resturns a vector.
- */
- /*
- * public Vector getNextMessagesToSend(){
- *
- * boolean done = false; Vector messages = new Vector();
- *
- * while(!done){ Long nextKey = new Long(lastProcessed+1); Object obj =
- * hash.get(nextKey); if(obj!=null){ messages.add(obj);
- * incrementProcessedCount(); }else{ done=true; //To exit the loop. } }
- * refreshHasProcessableMessages();
- *
- * return messages; }
- */
-
- /*
- * private void incrementProcessedCount(){ lastProcessed++; }
- */
-
- /*
- * private void refreshHasProcessableMessages(){ Long nextKey = new
- * Long(lastProcessed+1); hasMessagesToSend = hash.containsKey(nextKey); }
- */
-
public boolean hasMessage(Long key) {
Object obj = hash.get(key);
@@ -234,10 +165,7 @@
public void clearSequence(boolean yes) {
if (!yes)
return;
-
hash.clear();
- //lastProcessed = 0;
- //hasMessagesToSend = false;
nextAutoNumber = 1;
outSeqApproved = false;
outSequenceId = null;
@@ -260,10 +188,8 @@
//Deleting returns the deleted message.
public RMMessageContext deleteMessage(Long msgId) {
RMMessageContext msg = (RMMessageContext) hash.get(msgId);
-
if (msg == null)
return null;
-
hash.remove(msgId);
return msg;
}
@@ -272,8 +198,7 @@
if (hash.containsKey(messageNo)) {
markedAsDelete.add(messageNo);
String msgId = ((RMMessageContext) hash.get(messageNo)).getMessageID();
- System.out.println("INFO: Marking outgoing message deleted : msgId "
- + msgId);
+ log.info("INFO: Marking outgoing message deleted : msgId " + msgId);
return true;
}
return false;
@@ -285,27 +210,22 @@
public boolean isMessagePresent(String msgId) {
boolean b = false;
-
b = hash.containsKey(msgId);
return b;
}
public boolean hasMessageWithId(String msgId) {
- //boolean b = false;
-
Iterator it = hash.keySet().iterator();
while (it.hasNext()) {
RMMessageContext msg = (RMMessageContext) hash.get(it.next());
if (msg.getMessageID().equals(msgId))
return true;
-
}
return false;
}
public Vector getReceivedMsgNumbers() {
-
Vector result = new Vector();
Iterator it = hash.keySet().iterator();
@@ -315,13 +235,11 @@
long l = msg.getMsgNumber();
result.add(new Long(l));
}
-
return result;
}
public void setResponseReceived(String msgID) {
Iterator it = hash.keySet().iterator();
-
while (it.hasNext()) {
RMMessageContext msg = (RMMessageContext) hash.get(it.next());
if (msg.getMessageID().equals(msgID))
@@ -331,7 +249,6 @@
public void setAckReceived(String msgID) {
Iterator it = hash.keySet().iterator();
-
while (it.hasNext()) {
RMMessageContext msg = (RMMessageContext) hash.get(it.next());
if (msg.getMessageID().equals(msgID))
@@ -340,12 +257,12 @@
}
public void setAckReceived(long msgNo) {
-
RMMessageContext msg = (RMMessageContext) hash.get(new Long(msgNo));
if (msg != null) {
msg.setAckReceived(true);
- } else
- System.out.println("ERROR: MESSAGE IS NULL IN ResponseSeqHash");
+ } else {
+ log.error("ERROR: MESSAGE IS NULL IN ResponseSeqHash");
+ }
}
@@ -369,8 +286,8 @@
return false;
}
}
-
return true;
+
} catch (Exception e) {
e.printStackTrace();
return false;
@@ -385,36 +302,33 @@
return msg.getMsgNumber();
}
}
-
return -1;
}
-
- public void addMsgToSendList(long msgNo){
- sendMsgNoList.add( new Long(msgNo));
- }
-
- public boolean isMsgInSentList(long msgNo){
- return sendMsgNoList.contains(new Long(msgNo));
+
+ public void addMsgToSendList(long msgNo) {
+ sendMsgNoList.add(new Long(msgNo));
}
-
- public boolean hasLastMsgReceived(){
- if(lastMsgNo >0)
- return true;
- return false;
+ public boolean isMsgInSentList(long msgNo) {
+ return sendMsgNoList.contains(new Long(msgNo));
}
- public long getLastMsgNumber(){
- if(lastMsgNo>0)
- return lastMsgNo;
-
- return -1;
+ public boolean hasLastMsgReceived() {
+ if (lastMsgNo > 0)
+ return true;
+
+ return false;
}
-
- public void setLastMsg(long lastMsg){
- lastMsgNo = lastMsg;
+
+ public long getLastMsgNumber() {
+ if (lastMsgNo > 0)
+ return lastMsgNo;
+
+ return -1;
}
-
+ public void setLastMsg(long lastMsg) {
+ lastMsgNo = lastMsg;
+ }
}
\ No newline at end of file
1.5 +165 -274 ws-fx/sandesha/src/org/apache/sandesha/storage/queue/SandeshaQueue.java
Index: SandeshaQueue.java
===================================================================
RCS file: /home/cvs/ws-fx/sandesha/src/org/apache/sandesha/storage/queue/SandeshaQueue.java,v
retrieving revision 1.4
retrieving revision 1.5
diff -u -r1.4 -r1.5
--- SandeshaQueue.java 20 Feb 2005 08:49:45 -0000 1.4
+++ SandeshaQueue.java 20 Feb 2005 17:42:38 -0000 1.5
@@ -17,7 +17,8 @@
package org.apache.sandesha.storage.queue;
-
+import org.apache.axis.components.logger.LogFactory;
+import org.apache.commons.logging.Log;
import org.apache.sandesha.Constants;
import org.apache.sandesha.RMMessageContext;
import org.apache.sandesha.ws.rm.AcknowledgementRange;
@@ -37,23 +38,13 @@
public class SandeshaQueue {
private static SandeshaQueue queue = null;
-
HashMap incomingMap; //In comming messages.
-
HashMap outgoingMap; //Response messages
-
ArrayList highPriorityQueue; // Acks and create seq. responses.
-
HashMap queueBin; // Messaged processed from out queue will be moved
-
ArrayList lowPriorityQueue;
-
private List requestedSequences;
-
- //private Vector requestedSequences
- // to this.
-
- // to this.
+ private static final Log log = LogFactory.getLog(SandeshaQueue.class.getName());
private SandeshaQueue() {
incomingMap = new HashMap();
@@ -61,7 +52,6 @@
highPriorityQueue = new ArrayList();
queueBin = new HashMap();
lowPriorityQueue = new ArrayList();
-
requestedSequences = new ArrayList();
}
@@ -75,67 +65,71 @@
/**
* This will not replace messages automatically.
*/
- public boolean addMessageToIncomingSequence(String seqId, Long messageNo, RMMessageContext msgCon) throws QueueException {
+ public boolean addMessageToIncomingSequence(String seqId, Long messageNo,
+ RMMessageContext msgCon) throws QueueException {
boolean successful = false;
if (seqId == null || msgCon == null)
- throw new QueueException("Error in adding message");
+ throw new QueueException(Constants.Queue.ADD_ERROR);
if (isIncomingSequenceExists(seqId)) {
IncomingSequence seqHash = (IncomingSequence) incomingMap.get(seqId);
synchronized (seqHash) {
-
if (seqHash == null)
- throw new QueueException("Inconsistent queue");
+ throw new QueueException(Constants.Queue.QUEUE_INCONSIS);
if (seqHash.hasMessage(messageNo))
- throw new QueueException("Message already exists");
+ throw new QueueException(Constants.Queue.MESSAGE_EXISTS);
//Messages will not be replaced automatically.
+ //setting last message
+ if (msgCon.isLastMessage())
+ seqHash.setLastMsg(msgCon.getMsgNumber());
+
+ //setting sequence id
+ //TODO: Do this in create seq response processor
+ seqHash.setSequenceId(msgCon.getSequenceID());
+
seqHash.putNewMessage(messageNo, msgCon);
+ successful = true;
}
}
return successful;
}
- /**
- *
- */
- public boolean addMessageToOutgoingSequence(String seqId,
- RMMessageContext msgCon) throws QueueException {
+ public boolean addMessageToOutgoingSequence(String seqId, RMMessageContext msgCon)
+ throws QueueException {
boolean successful = false;
if (seqId == null || msgCon == null)
- throw new QueueException("Error in adding message");
+ throw new QueueException(Constants.Queue.ADD_ERROR);
if (isOutgoingSequenceExists(seqId)) {
OutgoingSequence resSeqHash = (OutgoingSequence) outgoingMap.get(seqId);
synchronized (resSeqHash) {
if (resSeqHash == null)
- throw new QueueException("Inconsistent queue");
- resSeqHash.putNewMessage(msgCon);
- successful=true;
+ throw new QueueException(Constants.Queue.QUEUE_INCONSIS);
+ resSeqHash.putNewMessage(msgCon);
+ successful = true;
//if last message
- //if(msgCon.isLastMessage())
- // resSeqHash.setLastMsg(msgCon.getMsgNumber());
-
+ if (msgCon.isLastMessage())
+ resSeqHash.setLastMsg(msgCon.getMsgNumber());
+
}
}
-
return successful;
}
- public boolean messagePresentInIncomingSequence(String sequenceId,
- Long messageNo) throws QueueException {
-
+ public boolean messagePresentInIncomingSequence(String sequenceId, Long messageNo)
+ throws QueueException {
IncomingSequence seqHash = (IncomingSequence) incomingMap.get(sequenceId);
if (seqHash == null)
- throw new QueueException("Sequence not present");
+ throw new QueueException(Constants.Queue.SEQUENCE_NOTPRESENT);
synchronized (seqHash) {
return seqHash.hasMessage(messageNo);
@@ -143,25 +137,19 @@
}
public boolean isIncomingSequenceExists(String seqId) {
-
synchronized (incomingMap) {
-
return incomingMap.containsKey(seqId);
}
}
public boolean isOutgoingSequenceExists(String resSeqId) {
-
synchronized (outgoingMap) {
-
return outgoingMap.containsKey(resSeqId);
}
}
public String nextIncomingSequenceIdToProcess() {
-
synchronized (incomingMap) {
-
int count = incomingMap.size();
Iterator it = incomingMap.keySet().iterator();
IncomingSequence sh = null;
@@ -175,21 +163,16 @@
break whileLoop;
}
}
-
return seqId;
}
}
- public RMMessageContext nextIncomingMessageToProcess(String sequenceId)
- throws QueueException {
-
+ public RMMessageContext nextIncomingMessageToProcess(String sequenceId) throws QueueException {
if (sequenceId == null)
return null;
IncomingSequence sh = (IncomingSequence) incomingMap.get(sequenceId);
-
synchronized (sh) {
-
if (sh == null)
throw new QueueException("Sequence id does not exist");
@@ -198,23 +181,18 @@
RMMessageContext msgCon = sh.getNextMessageToProcess();
return msgCon;
-
}
}
public RMMessageContext nextOutgoingMessageToSend() throws QueueException {
-
RMMessageContext msg = null;
-
synchronized (outgoingMap) {
-
Iterator it = outgoingMap.keySet().iterator();
whileLoop: while (it.hasNext()) {
RMMessageContext tempMsg;
String tempKey = (String) it.next();
- OutgoingSequence rsh = (OutgoingSequence) outgoingMap
- .get(tempKey);
+ OutgoingSequence rsh = (OutgoingSequence) outgoingMap.get(tempKey);
if (rsh.isOutSeqApproved()) {
tempMsg = rsh.getNextMessageToSend();
if (tempMsg != null) {
@@ -229,26 +207,22 @@
return msg;
}
- public void createNewIncomingSequence(String sequenceId)
- throws QueueException {
+ public void createNewIncomingSequence(String sequenceId) throws QueueException {
if (sequenceId == null)
- throw new QueueException("Sequence Id is null");
+ throw new QueueException(Constants.Queue.SEQUENCE_ID_NULL);
synchronized (incomingMap) {
-
IncomingSequence sh = new IncomingSequence(sequenceId);
incomingMap.put(sequenceId, sh);
}
}
- public void createNewOutgoingSequence(String sequenceId)
- throws QueueException {
+ public void createNewOutgoingSequence(String sequenceId) throws QueueException {
if (sequenceId == null)
- throw new QueueException("Sequence Id is null");
+ throw new QueueException(Constants.Queue.SEQUENCE_ID_NULL);
synchronized (outgoingMap) {
-
OutgoingSequence rsh = new OutgoingSequence(sequenceId);
outgoingMap.put(sequenceId, rsh);
}
@@ -259,29 +233,21 @@
* Adds a new message to the responses queue.
*/
public void addPriorityMessage(RMMessageContext msg) throws QueueException {
-
synchronized (highPriorityQueue) {
-
if (msg == null)
- throw new QueueException("Message is null");
-
+ throw new QueueException(Constants.Queue.MESSAGE_ID_NULL);
highPriorityQueue.add(msg);
}
}
-
public void addLowPriorityMessage(RMMessageContext msg) throws QueueException {
-
synchronized (lowPriorityQueue) {
-
if (msg == null)
- throw new QueueException("Message is null");
-
+ throw new QueueException(Constants.Queue.MESSAGE_ID_NULL);
lowPriorityQueue.add(msg);
}
}
-
public RMMessageContext nextPriorityMessageToSend() throws QueueException {
synchronized (highPriorityQueue) {
@@ -289,84 +255,47 @@
if (highPriorityQueue.size() <= 0)
return null;
- //RMMessageContext msg = (RMMessageContext) highPriorityQueue.get(0);
RMMessageContext msg = null;
int size = highPriorityQueue.size();
-
synchronized (highPriorityQueue) {
-
forLoop: //Label
for (int i = 0; i < size; i++) {
- RMMessageContext tempMsg = (RMMessageContext) highPriorityQueue
- .get(i);
+ RMMessageContext tempMsg = (RMMessageContext) highPriorityQueue.get(i);
if (tempMsg != null) {
-
switch (tempMsg.getMessageType()) {
- //Create seq messages will not be removed.
- case Constants.MSG_TYPE_CREATE_SEQUENCE_REQUEST:
- long lastSentTime = tempMsg.getLastSentTime();
- Date d = new Date();
- long currentTime = d.getTime();
- if (currentTime >= lastSentTime
- + Constants.RETRANSMISSION_INTERVAL) {
- tempMsg.setLastSentTime(currentTime);
- msg = tempMsg;
- break forLoop;
- }
- break;
-
- //Other msgs will be removed.
- //These include CreareSeqResponses and
- // Acknowledgements.
- default:
- highPriorityQueue.remove(i);
- queueBin.put(tempMsg.getMessageID(), tempMsg);
+ //Create seq messages will not be removed.
+ case Constants.MSG_TYPE_CREATE_SEQUENCE_REQUEST:
+ long lastSentTime = tempMsg.getLastSentTime();
+ Date d = new Date();
+ long currentTime = d.getTime();
+ if (currentTime >= lastSentTime + Constants.RETRANSMISSION_INTERVAL) {
+ tempMsg.setLastSentTime(currentTime);
msg = tempMsg;
break forLoop;
+ }
+ break;
+ default:
+ highPriorityQueue.remove(i);
+ queueBin.put(tempMsg.getMessageID(), tempMsg);
+ msg = tempMsg;
+ break forLoop;
}
-
}
}
}
-
return msg;
}
}
-
-
-
- /*
- * public RMMessageContext getNextToProcessIfHasNew(String sequenceId){
- * IncomingSequence sh = (IncomingSequence) incomingMap.get(sequenceId);
- * if(sh==null) return null;
- *
- * synchronized (sh) { if(!sh.hasNewMessages()) return null;
- *
- * Long key = sh. } }
- */
- public Vector nextAllMessagesToProcess(String sequenceId)
- throws QueueException {
+ public Vector nextAllMessagesToProcess(String sequenceId) throws QueueException {
IncomingSequence sh = (IncomingSequence) incomingMap.get(sequenceId);
-
synchronized (sh) {
Vector v = sh.getNextMessagesToProcess();
return v;
}
}
- //Folowing func. may cause errors.
- /*
- * public Vector nextAllResponseMessagesToSend(String sequenceId) throws
- * QueueException{ OutgoingSequence rsh = (OutgoingSequence)
- * outgoingMap.get(sequenceId); Vector v = new Vector(); synchronized (rsh){
- * RMMessageContext msg = nextAllResponseMessagesToSend()
- *
- * while(msg!=null){ v.add(msg); msg = rsh.getNextMessageToSend(); } return
- * v; } }
- */
-
public Vector nextAllSeqIdsToProcess() {
Vector ids = new Vector();
@@ -383,22 +312,9 @@
}
}
- /*
- * public Vector nextAllResponseSeqIdsToSend(){ Vector ids = new Vector();
- *
- * synchronized (outgoingMap){ Iterator it =
- * outgoingMap.keySet().iterator();
- *
- * while(it.hasNext()){ Object tempKey = it.next(); OutgoingSequence sh =
- * (OutgoingSequence) outgoingMap.get(tempKey);
- * if(sh.hasProcessableMessages()) ids.add(sh.getSequenceId()); } } return
- * ids; }
- */
-
public void clear(boolean yes) {
if (!yes)
return;
-
incomingMap.clear();
highPriorityQueue.clear();
outgoingMap.clear();
@@ -408,7 +324,6 @@
public void removeAllMsgsFromIncomingSeqence(String seqId, boolean yes) {
if (!yes)
return;
-
IncomingSequence sh = (IncomingSequence) incomingMap.get(seqId);
sh.clearSequence(yes);
}
@@ -416,7 +331,6 @@
public void removeAllMsgsFromOutgoingSeqence(String seqId, boolean yes) {
if (!yes)
return;
-
OutgoingSequence sh = (OutgoingSequence) outgoingMap.get(seqId);
sh.clearSequence(yes);
}
@@ -424,7 +338,6 @@
public void removeIncomingSequence(String sequenceId, boolean yes) {
if (!yes)
return;
-
incomingMap.remove(sequenceId);
}
@@ -453,8 +366,7 @@
public Set getAllReceivedMsgNumsOfOutgoingSeq(String sequenceId) {
Vector v = new Vector();
- OutgoingSequence rsh = (OutgoingSequence) outgoingMap
- .get(sequenceId);
+ OutgoingSequence rsh = (OutgoingSequence) outgoingMap.get(sequenceId);
synchronized (rsh) {
return rsh.getAllKeys();
}
@@ -473,7 +385,7 @@
OutgoingSequence rsh = (OutgoingSequence) outgoingMap.get(seqId);
if (rsh == null) {
- System.out.println("ERROR: RESPONSE SEQ IS NULL");
+ log.info("ERROR: RESPONSE SEQ IS NULL");
return;
}
@@ -483,11 +395,10 @@
}
public void setOutSequenceApproved(String seqId, boolean approved) {
- OutgoingSequence rsh = (OutgoingSequence) outgoingMap
- .get(seqId);
+ OutgoingSequence rsh = (OutgoingSequence) outgoingMap.get(seqId);
if (rsh == null) {
- System.out.println("ERROR: RESPONSE SEQ IS NULL");
+ log.info("ERROR: RESPONSE SEQ IS NULL");
return;
}
synchronized (rsh) {
@@ -501,17 +412,12 @@
return null;
}
- //Client will always handle a single seq
- //if(outSequence==Constants.CLIENT_DEFAULD_SEQUENCE_ID)
- // return outSequence;
-
Iterator it = outgoingMap.keySet().iterator();
synchronized (outgoingMap) {
while (it.hasNext()) {
String tempSeqId = (String) it.next();
- OutgoingSequence rsh = (OutgoingSequence) outgoingMap
- .get(tempSeqId);
+ OutgoingSequence rsh = (OutgoingSequence) outgoingMap.get(tempSeqId);
String tempOutSequence = rsh.getOutSequenceId();
if (outSequence.equals(tempOutSequence))
return tempSeqId;
@@ -529,15 +435,13 @@
while (it.hasNext()) {
String s = (String) it.next();
System.out.println("\n Sequence id - " + s);
- OutgoingSequence rsh = (OutgoingSequence) outgoingMap
- .get(s);
+ OutgoingSequence rsh = (OutgoingSequence) outgoingMap.get(s);
Iterator it1 = rsh.getAllKeys().iterator();
while (it1.hasNext()) {
Long l = (Long) it1.next();
String msgId = rsh.getMessageId(l);
- System.out.println("* key -" + l.longValue() + "- MessageID -"
- + msgId + "-");
+ System.out.println("* key -" + l.longValue() + "- MessageID -" + msgId + "-");
}
}
System.out.println("\n");
@@ -557,8 +461,7 @@
while (it1.hasNext()) {
Long l = (Long) it1.next();
String msgId = sh.getMessageId(l);
- System.out.println("* key -" + l.longValue() + "- MessageID -"
- + msgId + "-");
+ System.out.println("* key -" + l.longValue() + "- MessageID -" + msgId + "-");
}
}
System.out.println("\n");
@@ -583,11 +486,10 @@
public void moveOutgoingMsgToBin(String sequenceId, Long messageNo) {
String sequence = getSequenceOfOutSequence(sequenceId);
- OutgoingSequence rsh = (OutgoingSequence) outgoingMap
- .get(sequence);
+ OutgoingSequence rsh = (OutgoingSequence) outgoingMap.get(sequence);
if (rsh == null) {
- System.out.println("ERROR: RESPONSE SEQ IS NULL " + sequence);
+ System.out.println(Constants.Queue.RESPONSE_SEQ_NULL);
return;
}
@@ -598,9 +500,7 @@
if (msg != null) {
String msgId = msg.getMessageID();
- System.out
- .println("INFO: Moving out going messages to bin : msgId "
- + msgId);
+ log.info("INFO: Moving out going messages to bin");
//Add msg to bin if id isnt null.
if (msgId != null)
queueBin.put(msgId, msg);
@@ -611,11 +511,10 @@
public void markOutgoingMessageToDelete(String sequenceId, Long messageNo) {
String sequence = getSequenceOfOutSequence(sequenceId);
- OutgoingSequence rsh = (OutgoingSequence) outgoingMap
- .get(sequence);
+ OutgoingSequence rsh = (OutgoingSequence) outgoingMap.get(sequence);
if (rsh == null) {
- System.out.println("ERROR: RESPONSE SEQ IS NULL " + sequence);
+ log.error(Constants.Queue.RESPONSE_SEQ_NULL);
return;
}
@@ -627,15 +526,12 @@
}
-
public void movePriorityMsgToBin(String messageId) {
synchronized (highPriorityQueue) {
int size = highPriorityQueue.size();
for (int i = 0; i < size; i++) {
RMMessageContext msg = (RMMessageContext) highPriorityQueue.get(i);
-
if (msg.getMessageID().equals(messageId)) {
-
highPriorityQueue.remove(i);
queueBin.put(messageId, msg);
return;
@@ -646,12 +542,11 @@
public long getNextOutgoingMessageNumber(String seq) {
OutgoingSequence rsh = (OutgoingSequence) outgoingMap.get(seq);
-
if (rsh == null) { //saquence not created yet.
try {
createNewOutgoingSequence(seq);
} catch (QueueException q) {
- System.out.println(q.getStackTrace());
+ log.error(q.getStackTrace());
}
}
rsh = (OutgoingSequence) outgoingMap.get(seq);
@@ -659,31 +554,16 @@
Iterator keys = rsh.getAllKeys().iterator();
long msgNo = rsh.nextMessageNumber();
-
- /* while (keys.hasNext()) {
- System.out.println("HAS KEYS");
- long temp = ((Long) keys.next()).longValue();
- System.out.println("TEMP IS "+temp);
- if (temp > msgNo)
- msgNo = temp;
- }
-
- msgNo++;*/
-
- System.out.println("RETURNING MSG NUMBER " + msgNo);
return (msgNo);
}
}
public RMMessageContext checkForResponseMessage(String requestId, String seqId) {
IncomingSequence sh = (IncomingSequence) incomingMap.get(seqId);
- System.out.println("DEFAULT : " + requestId + " SEQ " + seqId);
-
if (sh == null) {
- System.out.println("ERROR: SEQ IS NULL");
+ log.error(Constants.Queue.SEQUENCE_ABSENT);
return null;
}
-
synchronized (sh) {
RMMessageContext msg = sh.getMessageRelatingTo(requestId);
return msg;
@@ -692,17 +572,14 @@
public boolean isRequestMsgPresent(String seqId, String messageId) {
OutgoingSequence rsh = (OutgoingSequence) outgoingMap.get(seqId);
-
if (rsh == null) {
- System.out.println("ERROR: SEQ IS NULL");
+ log.error(Constants.Queue.SEQUENCE_ABSENT);
return false;
}
-
boolean present = false;
synchronized (rsh) {
present = rsh.isMessagePresent(messageId);
}
-
return present;
}
@@ -711,7 +588,6 @@
Iterator it = outgoingMap.keySet().iterator();
String key = null;
-
while (it.hasNext()) {
key = (String) it.next();
Object obj = outgoingMap.get(key);
@@ -723,7 +599,6 @@
}
}
-
return key;
}
@@ -738,22 +613,20 @@
SequenceAcknowledgement seqAck = msg.getRMHeaders().getSequenceAcknowledgement();
String sId = seqAck.getIdentifier().getIdentifier();
- if (seqId != sId) //Sorry. Wrong sequence.
+ if (seqId != sId) //Sorry. Wrong sequence.
continue;
List ackList = seqAck.getAckRanges();
-
Iterator ackIt = ackList.iterator();
-
while (ackIt.hasNext()) {
AcknowledgementRange ackRng = (AcknowledgementRange) ackIt.next();
long min = ackRng.getMinValue();
long temp = min;
while (temp <= ackRng.getMaxValue()) {
Long lng = new Long(temp);
- if (!msgNumbers.contains(lng)) //vector cant hv duplicate entries.
+ if (!msgNumbers.contains(lng)) //vector cant hv duplicate
+ // entries.
msgNumbers.add(new Long(temp));
-
temp++;
}
}
@@ -763,24 +636,19 @@
public Vector getAllOutgoingMsgNumbers(String seqId) {
Vector msgNumbers = new Vector();
-
OutgoingSequence rsh = (OutgoingSequence) outgoingMap.get(seqId);
-
if (rsh == null) {
- System.out.println("ERROR: SEQ IS NULL " + seqId);
+ log.error(Constants.Queue.SEQUENCE_ABSENT);
return msgNumbers;
}
-
synchronized (rsh) {
msgNumbers = rsh.getReceivedMsgNumbers();
}
-
return msgNumbers;
}
public void setResponseReceived(RMMessageContext responseMsg) {
String requestMsgID = responseMsg.getAddressingHeaders().getRelatesTo().toString();
-
Iterator it = outgoingMap.keySet().iterator();
String key = null;
@@ -791,75 +659,52 @@
OutgoingSequence hash = (OutgoingSequence) obj;
boolean hasMsg = hash.hasMessageWithId(requestMsgID);
if (!hasMsg)
- //set the property response received
+ //set the property response received
hash.setResponseReceived(requestMsgID);
}
}
-
}
public void setAckReceived(String seqId, long msgNo) {
-
Iterator it = outgoingMap.keySet().iterator();
-
String key = null;
while (it.hasNext()) {
key = (String) it.next();
Object obj = outgoingMap.get(key);
-
if (obj != null) {
OutgoingSequence hash = (OutgoingSequence) obj;
- System.out.println("************** HASH SEQ IS " + hash.getSequenceId() + " SEQ IS " + seqId + " OUT SEQ IS " + hash.getOutSequenceId());
if (hash.getOutSequenceId().equals(seqId)) {
- hash.setAckReceived(msgNo);
+ hash.setAckReceived(msgNo);
}
}
}
}
-
public Vector getAllIncommingMsgNumbers(String seqId) {
Vector msgNumbers = new Vector();
incomingMap.get(seqId);
-
//Not implemented yet.
return msgNumbers;
}
public RMMessageContext getLowPriorityMessageIfAcked() {
-
int size = lowPriorityQueue.size();
-
RMMessageContext terminateMsg = null;
for (int i = 0; i < size; i++) {
RMMessageContext temp;
temp = (RMMessageContext) lowPriorityQueue.get(i);
String seqId = temp.getSequenceID();
- // System.out.println(" HASH NOT FOUND SEQ ID " + seqId);
-
OutgoingSequence hash = null;
hash = (OutgoingSequence) outgoingMap.get(seqId);
if (hash == null) {
- System.out.println("SandeshaQueue: ERROR: HASH NOT FOUND SEQ ID " + seqId);
+ log.error("ERROR: HASH NOT FOUND SEQ ID " + seqId);
}
-
- /*Iterator it1 = outgoingMap.keySet().iterator();
- while(it1.hasNext()){
- hash = (OutgoingSequence) it1.next();
- if(hash.getOutSequenceId().equals(seqId)){
- System.out.println("~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ FOUND SEQ "+ seqId);
- foundSeq = true;
- break;
- }
- } */
-
if (hash != null) {
boolean complete = hash.isAckComplete();
if (complete)
- //lowPriorityQueue.remove(i);
terminateMsg = temp;
if (terminateMsg != null) {
terminateMsg.setSequenceID(hash.getOutSequenceId());
@@ -867,71 +712,117 @@
break;
}
}
-
}
return terminateMsg;
}
-
- public void addSendMsgNo(String seqId,long msgNo){
- OutgoingSequence rsh = (OutgoingSequence) outgoingMap.get(seqId);
- System.out.println("5555555555555555555555555555555555555555555555555555555555555555");
+
+ public void addSendMsgNo(String seqId, long msgNo) {
+ OutgoingSequence rsh = (OutgoingSequence) outgoingMap.get(seqId);
if (rsh == null) {
- System.out.println("ERROR: SEQ IS NULL "+seqId+" "+msgNo);
+ log.error(Constants.Queue.SEQUENCE_ABSENT);
}
-
- synchronized(rsh){
- rsh.addMsgToSendList(msgNo);
- }
- }
-
- public boolean isSentMsg(String seqId,long msgNo){
- OutgoingSequence rsh = (OutgoingSequence) outgoingMap.get(seqId);
-
- if (rsh == null) {
- System.out.println("ERROR: SEQ IS NULL");
+ synchronized (rsh) {
+ rsh.addMsgToSendList(msgNo);
}
-
- synchronized(rsh){
- return rsh.isMsgInSentList(msgNo);
- }
- }
-
- public boolean hasLastMsgReceived(String seqId) {
-
- OutgoingSequence rsh = (OutgoingSequence) outgoingMap.get(seqId);
-
+ }
+
+ public boolean isSentMsg(String seqId, long msgNo) {
+ OutgoingSequence rsh = (OutgoingSequence) outgoingMap.get(seqId);
+
if (rsh == null) {
- System.out.println("ERROR: SEQ IS NULL");
+ log.error(Constants.Queue.SEQUENCE_ABSENT);
+ }
+ synchronized (rsh) {
+ return rsh.isMsgInSentList(msgNo);
+ }
+ }
+
+ public boolean hasLastOutgoingMsgReceived(String seqId) {
+ OutgoingSequence rsh = (OutgoingSequence) outgoingMap.get(seqId);
+
+ if (rsh == null) {
+ log.error(Constants.Queue.SEQUENCE_ABSENT);
+ }
+ synchronized (rsh) {
+ return rsh.hasLastMsgReceived();
}
-
- synchronized(rsh){
- return rsh.hasLastMsgReceived();
- }
}
- public long getLastMsgNo(String seqId){
- OutgoingSequence rsh = (OutgoingSequence) outgoingMap.get(seqId);
-
+ public long getLastOutgoingMsgNo(String seqId) {
+ OutgoingSequence rsh = (OutgoingSequence) outgoingMap.get(seqId);
+
if (rsh == null) {
- System.out.println("ERROR: SEQ IS NULL");
+ log.error(Constants.Queue.SEQUENCE_ABSENT);
+
+ }
+ synchronized (rsh) {
+ return rsh.getLastMsgNumber();
+ }
+ }
+
+ public boolean hasLastIncomingMsgReceived(String seqId) {
+
+ IncomingSequence sh = (IncomingSequence) incomingMap.get(seqId);
+
+ if (sh == null) {
+ log.error(Constants.Queue.SEQUENCE_ABSENT);
+ return false;
+ }
+ synchronized (sh) {
+ return sh.hasLastMsgReceived();
+ }
+ }
+
+ public long getLastIncomingMsgNo(String seqId) {
+ IncomingSequence sh = (IncomingSequence) incomingMap.get(seqId);
+ if (sh == null) {
+ log.error(Constants.Queue.SEQUENCE_ABSENT);
}
-
- synchronized(rsh){
- return rsh.getLastMsgNumber();
- }
+ synchronized (sh) {
+ return sh.getLastMsgNumber();
+ }
}
- public void addRequestedSequence(String seqId){
+ public void addRequestedSequence(String seqId) {
requestedSequences.add(seqId);
}
- public boolean isRequestedSeqPresent(String seqId){
- return requestedSequences.contains(seqId);
+ public boolean isRequestedSeqPresent(String seqId) {
+ return requestedSequences.contains(seqId);
}
+ public String getKeyFromIncomingSequenceId(String seqId) {
+ Iterator it = incomingMap.keySet().iterator();
+ while (it.hasNext()) {
+ String key = (String) it.next();
+ IncomingSequence is = (IncomingSequence) incomingMap.get(key);
+ String seq = is.getSequenceId();
+ if (seq == null)
+ continue;
+ if (seq.equals(seqId))
+ return key;
+ }
+ return null;
+ }
+
+ public String getKeyFromOutgoingSequenceId(String seqId) {
+ Iterator it = outgoingMap.keySet().iterator();
+ while (it.hasNext()) {
+ String key = (String) it.next();
+ OutgoingSequence os = (OutgoingSequence) outgoingMap.get(key);
+
+ String seq = os.getSequenceId();
+ if (seq == null)
+ continue;
+
+ if (seq.equals(seqId))
+ return key;
+ }
+ return null;
+ }
}