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 ja...@apache.org on 2004/09/20 14:17:08 UTC
cvs commit: ws-fx/sandesha/src/org/apache/sandesha/server/queue QueueException.java ResponseSequenceHash.java ServerQueue.java
jaliya 2004/09/20 05:17:08
Modified: sandesha/src/org/apache/sandesha/server/queue
QueueException.java ResponseSequenceHash.java
ServerQueue.java
Log:
Changes were done to support for the client side implementation
Revision Changes Path
1.3 +0 -1 ws-fx/sandesha/src/org/apache/sandesha/server/queue/QueueException.java
Index: QueueException.java
===================================================================
RCS file: /home/cvs/ws-fx/sandesha/src/org/apache/sandesha/server/queue/QueueException.java,v
retrieving revision 1.2
retrieving revision 1.3
diff -u -r1.2 -r1.3
--- QueueException.java 13 Sep 2004 11:26:51 -0000 1.2
+++ QueueException.java 20 Sep 2004 12:17:07 -0000 1.3
@@ -16,7 +16,6 @@
*/
package org.apache.sandesha.server.queue;
-
/**
* @author Chamikara Jayalath
* @author Jaliya Ekanayaka
1.3 +1 -1 ws-fx/sandesha/src/org/apache/sandesha/server/queue/ResponseSequenceHash.java
Index: ResponseSequenceHash.java
===================================================================
RCS file: /home/cvs/ws-fx/sandesha/src/org/apache/sandesha/server/queue/ResponseSequenceHash.java,v
retrieving revision 1.2
retrieving revision 1.3
diff -u -r1.2 -r1.3
--- ResponseSequenceHash.java 13 Sep 2004 11:26:51 -0000 1.2
+++ ResponseSequenceHash.java 20 Sep 2004 12:17:07 -0000 1.3
@@ -40,7 +40,7 @@
*/
public class ResponseSequenceHash {
- //following concepts was removed from responseQueue.
+ //following concepts was removed from priorityQueue.
//This was to reduce complexity. (Since here time is also taken into
// account)
//private long lastProcessed;
1.3 +123 -100 ws-fx/sandesha/src/org/apache/sandesha/server/queue/ServerQueue.java
Index: ServerQueue.java
===================================================================
RCS file: /home/cvs/ws-fx/sandesha/src/org/apache/sandesha/server/queue/ServerQueue.java,v
retrieving revision 1.2
retrieving revision 1.3
diff -u -r1.2 -r1.3
--- ServerQueue.java 13 Sep 2004 11:26:51 -0000 1.2
+++ ServerQueue.java 20 Sep 2004 12:17:07 -0000 1.3
@@ -40,20 +40,20 @@
private static ServerQueue queue = null;
- HashMap sequenceMap; //In comming messages.
+ HashMap incomingMap; //In comming messages.
- HashMap responseMap; //Response messages
+ HashMap outgoingMap; //Response messages
- ArrayList responseQueue; // Acks and create seq. responses.
+ ArrayList priorityQueue; // Acks and create seq. responses.
- HashMap responseQueueBin; // Messaged processed from out queue will be moved
+ HashMap queueBin; // Messaged processed from out queue will be moved
// to this.
private ServerQueue() {
- sequenceMap = new HashMap();
- responseMap = new HashMap();
- responseQueue = new ArrayList();
- responseQueueBin = new HashMap();
+ incomingMap = new HashMap();
+ outgoingMap = new HashMap();
+ priorityQueue = new ArrayList();
+ queueBin = new HashMap();
}
public static ServerQueue getInstance() {
@@ -84,23 +84,23 @@
* public RMSequenceContext getRMSequenceContext(String sequenceID){
* if(sequenceID==null) return null;
*
- * Object obj = sequenceMap.get(sequenceID); if(obj!=null && (obj instanceof
+ * Object obj = incomingMap.get(sequenceID); if(obj!=null && (obj instanceof
* RMSequenceContext)) return (RMSequenceContext)
- * sequenceMap.get(sequenceID); else return null; }
+ * incomingMap.get(sequenceID); else return null; }
*/
/**
* This will not replace messages automatically.
*/
- public boolean addMessageToSequence(String seqId, Long messageNo,
+ 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");
- if (isSequenceExists(seqId)) {
- SequenceHash seqHash = (SequenceHash) sequenceMap.get(seqId);
+ if (isIncomingSequenceExists(seqId)) {
+ SequenceHash seqHash = (SequenceHash) incomingMap.get(seqId);
synchronized (seqHash) {
@@ -121,15 +121,15 @@
/**
*
*/
- public boolean addMessageToResponseSequence(String seqId,
+ public boolean addMessageToOutgoingSequence(String seqId,
RMMessageContext msgCon) throws QueueException {
boolean successful = false;
if (seqId == null || msgCon == null)
throw new QueueException("Error in adding message");
- if (isResponseSequenceExists(seqId)) {
- ResponseSequenceHash resSeqHash = (ResponseSequenceHash) responseMap
+ if (isOutgoingSequenceExists(seqId)) {
+ ResponseSequenceHash resSeqHash = (ResponseSequenceHash) outgoingMap
.get(seqId);
synchronized (resSeqHash) {
@@ -144,10 +144,10 @@
return successful;
}
- public boolean messagePresentInSequence(String sequenceId, Long messageNo)
+ public boolean messagePresentInIncomingSequence(String sequenceId, Long messageNo)
throws QueueException {
- SequenceHash seqHash = (SequenceHash) sequenceMap.get(sequenceId);
+ SequenceHash seqHash = (SequenceHash) incomingMap.get(sequenceId);
if (seqHash == null)
throw new QueueException("Sequence not present");
@@ -157,34 +157,34 @@
}
}
- public boolean isSequenceExists(String seqId) {
+ public boolean isIncomingSequenceExists(String seqId) {
- synchronized (sequenceMap) {
+ synchronized (incomingMap) {
- return sequenceMap.containsKey(seqId);
+ return incomingMap.containsKey(seqId);
}
}
- public boolean isResponseSequenceExists(String resSeqId) {
+ public boolean isOutgoingSequenceExists(String resSeqId) {
- synchronized (responseMap) {
+ synchronized (outgoingMap) {
- return responseMap.containsKey(resSeqId);
+ return outgoingMap.containsKey(resSeqId);
}
}
- public String nextSequenceIdToProcess() {
+ public String nextIncomingSequenceIdToProcess() {
- synchronized (sequenceMap) {
+ synchronized (incomingMap) {
- int count = sequenceMap.size();
- Iterator it = sequenceMap.keySet().iterator();
+ int count = incomingMap.size();
+ Iterator it = incomingMap.keySet().iterator();
SequenceHash sh = null;
String seqId = null;
whileLoop: while (it.hasNext()) {
String tempSeqId = (String) it.next();
- sh = (SequenceHash) sequenceMap.get(tempSeqId);
+ sh = (SequenceHash) incomingMap.get(tempSeqId);
if (sh.hasProcessableMessages()) {
seqId = tempSeqId;
break whileLoop;
@@ -195,13 +195,13 @@
}
}
- public RMMessageContext nextMessageToProcess(String sequenceId)
+ public RMMessageContext nextIncomingMessageToProcess(String sequenceId)
throws QueueException {
if (sequenceId == null)
return null;
- SequenceHash sh = (SequenceHash) sequenceMap.get(sequenceId);
+ SequenceHash sh = (SequenceHash) incomingMap.get(sequenceId);
synchronized (sh) {
@@ -217,19 +217,19 @@
}
}
- public RMMessageContext nextResponseMessageToSend() throws QueueException {
+ public RMMessageContext nextOutgoingMessageToSend() throws QueueException {
RMMessageContext msg = null;
- synchronized (responseMap) {
+ synchronized (outgoingMap) {
- Iterator it = responseMap.keySet().iterator();
+ Iterator it = outgoingMap.keySet().iterator();
whileLoop: while (it.hasNext()) {
RMMessageContext tempMsg;
String tempKey = (String) it.next();
- ResponseSequenceHash rsh = (ResponseSequenceHash) responseMap
+ ResponseSequenceHash rsh = (ResponseSequenceHash) outgoingMap
.get(tempKey);
if (rsh.isOutSeqApproved()) {
tempMsg = rsh.getNextMessageToSend();
@@ -243,26 +243,26 @@
return msg;
}
- public void createNewSequence(String sequenceId) throws QueueException {
+ public void createNewIncomingSequence(String sequenceId) throws QueueException {
if (sequenceId == null)
throw new QueueException("Sequence Id is null");
- synchronized (sequenceMap) {
+ synchronized (incomingMap) {
SequenceHash sh = new SequenceHash(sequenceId);
- sequenceMap.put(sequenceId, sh);
+ incomingMap.put(sequenceId, sh);
}
}
- public void createNewResponseSequence(String sequenceId)
+ public void createNewOutgoingSequence(String sequenceId)
throws QueueException {
if (sequenceId == null)
throw new QueueException("Sequence Id is null");
- synchronized (responseMap) {
+ synchronized (outgoingMap) {
ResponseSequenceHash rsh = new ResponseSequenceHash(sequenceId);
- responseMap.put(sequenceId, rsh);
+ outgoingMap.put(sequenceId, rsh);
}
}
@@ -272,31 +272,31 @@
*/
public void addPriorityMessage(RMMessageContext msg) throws QueueException {
- synchronized (responseQueue) {
+ synchronized (priorityQueue) {
if (msg == null)
throw new QueueException("Message is null");
- responseQueue.add(msg);
+ priorityQueue.add(msg);
}
}
public RMMessageContext nextPriorityMessageToSend() throws QueueException {
- synchronized (responseQueue) {
+ synchronized (priorityQueue) {
- if (responseQueue.size() <= 0)
+ if (priorityQueue.size() <= 0)
return null;
- //RMMessageContext msg = (RMMessageContext) responseQueue.get(0);
+ //RMMessageContext msg = (RMMessageContext) priorityQueue.get(0);
RMMessageContext msg = null;
- int size = responseQueue.size();
+ int size = priorityQueue.size();
- synchronized (responseQueue) {
+ synchronized (priorityQueue) {
forLoop: //Label
for (int i = 0; i < size; i++) {
- RMMessageContext tempMsg = (RMMessageContext) responseQueue
+ RMMessageContext tempMsg = (RMMessageContext) priorityQueue
.get(i);
if (tempMsg != null) {
@@ -318,8 +318,8 @@
//These include CreareSeqResponses and
// Acknowledgements.
default:
- responseQueue.remove(i);
- responseQueueBin.put(tempMsg.getMessageID(),
+ priorityQueue.remove(i);
+ queueBin.put(tempMsg.getMessageID(),
tempMsg);
msg = tempMsg;
break forLoop;
@@ -336,7 +336,7 @@
/*
* public RMMessageContext getNextToProcessIfHasNew(String sequenceId){
- * SequenceHash sh = (SequenceHash) sequenceMap.get(sequenceId);
+ * SequenceHash sh = (SequenceHash) incomingMap.get(sequenceId);
* if(sh==null) return null;
*
* synchronized (sh) { if(!sh.hasNewMessages()) return null;
@@ -346,7 +346,7 @@
public Vector nextAllMessagesToProcess(String sequenceId)
throws QueueException {
- SequenceHash sh = (SequenceHash) sequenceMap.get(sequenceId);
+ SequenceHash sh = (SequenceHash) incomingMap.get(sequenceId);
synchronized (sh) {
Vector v = sh.getNextMessagesToProcess();
@@ -358,7 +358,7 @@
/*
* public Vector nextAllResponseMessagesToSend(String sequenceId) throws
* QueueException{ ResponseSequenceHash rsh = (ResponseSequenceHash)
- * responseMap.get(sequenceId); Vector v = new Vector(); synchronized (rsh){
+ * outgoingMap.get(sequenceId); Vector v = new Vector(); synchronized (rsh){
* RMMessageContext msg = nextAllResponseMessagesToSend()
*
* while(msg!=null){ v.add(msg); msg = rsh.getNextMessageToSend(); } return
@@ -368,12 +368,12 @@
public Vector nextAllSeqIdsToProcess() {
Vector ids = new Vector();
- synchronized (sequenceMap) {
- Iterator it = sequenceMap.keySet().iterator();
+ synchronized (incomingMap) {
+ Iterator it = incomingMap.keySet().iterator();
while (it.hasNext()) {
Object tempKey = it.next();
- SequenceHash sh = (SequenceHash) sequenceMap.get(tempKey);
+ SequenceHash sh = (SequenceHash) incomingMap.get(tempKey);
if (sh.hasProcessableMessages() && !sh.isSequenceLocked())
ids.add(sh.getSequenceId());
}
@@ -384,11 +384,11 @@
/*
* public Vector nextAllResponseSeqIdsToSend(){ Vector ids = new Vector();
*
- * synchronized (responseMap){ Iterator it =
- * responseMap.keySet().iterator();
+ * synchronized (outgoingMap){ Iterator it =
+ * outgoingMap.keySet().iterator();
*
* while(it.hasNext()){ Object tempKey = it.next(); ResponseSequenceHash sh =
- * (ResponseSequenceHash) responseMap.get(tempKey);
+ * (ResponseSequenceHash) outgoingMap.get(tempKey);
* if(sh.hasProcessableMessages()) ids.add(sh.getSequenceId()); } } return
* ids; }
*/
@@ -397,66 +397,66 @@
if (!yes)
return;
- sequenceMap.clear();
- responseQueue.clear();
- responseMap.clear();
- responseQueueBin.clear();
+ incomingMap.clear();
+ priorityQueue.clear();
+ outgoingMap.clear();
+ queueBin.clear();
}
- public void removeAllMsgsFromSeqence(String seqId, boolean yes) {
+ public void removeAllMsgsFromIncomingSeqence(String seqId, boolean yes) {
if (!yes)
return;
- SequenceHash sh = (SequenceHash) sequenceMap.get(seqId);
+ SequenceHash sh = (SequenceHash) incomingMap.get(seqId);
sh.clearSequence(yes);
}
- public void removeAllMsgsFromResponseSeqence(String seqId, boolean yes) {
+ public void removeAllMsgsFromOutgoingSeqence(String seqId, boolean yes) {
if (!yes)
return;
- ResponseSequenceHash sh = (ResponseSequenceHash) responseMap.get(seqId);
+ ResponseSequenceHash sh = (ResponseSequenceHash) outgoingMap.get(seqId);
sh.clearSequence(yes);
}
- public void removeSequence(String sequenceId, boolean yes) {
+ public void removeIncomingSequence(String sequenceId, boolean yes) {
if (!yes)
return;
- sequenceMap.remove(sequenceId);
+ incomingMap.remove(sequenceId);
}
- public void removeResponseSequence(String sequenceId, boolean yes) {
+ public void removeOutgoingSequence(String sequenceId, boolean yes) {
if (!yes)
return;
- synchronized (responseMap) {
- responseMap.remove(sequenceId);
+ synchronized (outgoingMap) {
+ outgoingMap.remove(sequenceId);
}
}
public void setSequenceLock(String sequenceId, boolean lock) {
- SequenceHash sh = (SequenceHash) sequenceMap.get(sequenceId);
+ SequenceHash sh = (SequenceHash) incomingMap.get(sequenceId);
sh.setProcessLock(lock);
}
- public Set getAllReceivedMsgNumsOfSeq(String sequenceId) {
+ public Set getAllReceivedMsgNumsOfIncomingSeq(String sequenceId) {
Vector v = new Vector();
- SequenceHash sh = (SequenceHash) sequenceMap.get(sequenceId);
+ SequenceHash sh = (SequenceHash) incomingMap.get(sequenceId);
return sh.getAllKeys();
}
- public Set getAllReceivedMsgNumsOfResponseSeq(String sequenceId) {
+ public Set getAllReceivedMsgNumsOfOutgoingSeq(String sequenceId) {
Vector v = new Vector();
- ResponseSequenceHash rsh = (ResponseSequenceHash) responseMap
+ ResponseSequenceHash rsh = (ResponseSequenceHash) outgoingMap
.get(sequenceId);
synchronized (rsh) {
return rsh.getAllKeys();
}
}
- public boolean isMessageExists(String sequenceId, Long messageNo) {
- SequenceHash sh = (SequenceHash) sequenceMap.get(sequenceId);
+ public boolean isIncomingMessageExists(String sequenceId, Long messageNo) {
+ SequenceHash sh = (SequenceHash) incomingMap.get(sequenceId);
//sh can be null if there are no messages at the initial point.
if (sh != null)
return sh.hasMessage(messageNo);
@@ -465,7 +465,7 @@
}
public void setOutSequence(String seqId, String outSeqId) {
- ResponseSequenceHash rsh = (ResponseSequenceHash) responseMap
+ ResponseSequenceHash rsh = (ResponseSequenceHash) outgoingMap
.get(seqId);
if (rsh == null) {
@@ -479,7 +479,7 @@
}
public void setOutSequenceApproved(String seqId, boolean approved) {
- ResponseSequenceHash rsh = (ResponseSequenceHash) responseMap
+ ResponseSequenceHash rsh = (ResponseSequenceHash) outgoingMap
.get(seqId);
if (rsh == null) {
@@ -495,11 +495,11 @@
if (outSequence == null)
return null;
- Iterator it = responseMap.keySet().iterator();
- synchronized (responseMap) {
+ Iterator it = outgoingMap.keySet().iterator();
+ synchronized (outgoingMap) {
while (it.hasNext()) {
String tempSeqId = (String) it.next();
- ResponseSequenceHash rsh = (ResponseSequenceHash) responseMap
+ ResponseSequenceHash rsh = (ResponseSequenceHash) outgoingMap
.get(tempSeqId);
String tempOutSequence = rsh.getOutSequenceId();
@@ -511,15 +511,15 @@
return null;
}
- public void displayResponseMap() {
- Iterator it = responseMap.keySet().iterator();
+ public void displayOutgoingMap() {
+ Iterator it = outgoingMap.keySet().iterator();
System.out.println("***************************************");
System.out.println(" DISPLAYING RESPONSE MAP");
System.out.println(" -----------------------");
while (it.hasNext()) {
String s = (String) it.next();
System.out.println("\n Sequence id - " + s);
- ResponseSequenceHash rsh = (ResponseSequenceHash) responseMap
+ ResponseSequenceHash rsh = (ResponseSequenceHash) outgoingMap
.get(s);
Iterator it1 = rsh.getAllKeys().iterator();
@@ -533,15 +533,15 @@
System.out.println("***************************************");
}
- public void displaySequenceMap() {
- Iterator it = sequenceMap.keySet().iterator();
+ public void displayIncomingMap() {
+ Iterator it = incomingMap.keySet().iterator();
System.out.println("***************************************");
System.out.println(" DISPLAYING SEQUENCE MAP");
System.out.println(" -----------------------");
while (it.hasNext()) {
String s = (String) it.next();
System.out.println("\n Sequence id - " + s);
- SequenceHash sh = (SequenceHash) sequenceMap.get(s);
+ SequenceHash sh = (SequenceHash) incomingMap.get(s);
Iterator it1 = sh.getAllKeys().iterator();
while (it1.hasNext()) {
@@ -560,7 +560,7 @@
System.out.println(" DISPLAYING PRIORITY QUEUE");
System.out.println(" -------------------------");
- Iterator it = responseQueue.iterator();
+ Iterator it = priorityQueue.iterator();
while (it.hasNext()) {
RMMessageContext msg = (RMMessageContext) it.next();
String id = msg.getMessageID();
@@ -571,8 +571,8 @@
System.out.println("***************************************");
}
- public void moveResponseMsgToBin(String sequenceId, Long messageNo) {
- ResponseSequenceHash rsh = (ResponseSequenceHash) responseMap
+ public void moveOutgoingMsgToBin(String sequenceId, Long messageNo) {
+ ResponseSequenceHash rsh = (ResponseSequenceHash) outgoingMap
.get(sequenceId);
if (rsh == null) {
@@ -587,25 +587,48 @@
//Add msg to bin if id isnt null.
if (msgId != null)
- responseQueueBin.put(msgId, msg);
+ queueBin.put(msgId, msg);
}
}
public void movePriorityMsgToBin(String messageId) {
- synchronized (responseQueue) {
+ synchronized (priorityQueue) {
- int size = responseQueue.size();
+ int size = priorityQueue.size();
for (int i = 0; i < size; i++) {
- RMMessageContext msg = (RMMessageContext) responseQueue.get(i);
+ RMMessageContext msg = (RMMessageContext) priorityQueue.get(i);
if (msg.getMessageID().equals(messageId)) {
- responseQueue.remove(i);
- responseQueueBin.put(messageId, msg);
+ priorityQueue.remove(i);
+ queueBin.put(messageId, msg);
return;
}
}
+ }
+ }
+
+ public long getNextOutgoingMessageNumber (String seq){
+ ResponseSequenceHash rsh = (ResponseSequenceHash) outgoingMap.get(seq);
+ if(rsh==null) //saquence not created yet.
+ return 1;
+
+ synchronized (rsh){
+
+
+ Iterator keys = rsh.getAllKeys().iterator();
+
+ long l = 0;
+
+ while(keys.hasNext()){
+ long temp = ((Long) keys.next()).longValue();
+ if(temp>l)
+ l = temp;
+ }
+
+ return (l+1);
+
}
}
}