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/10/09 10:50:23 UTC
cvs commit: ws-fx/sandesha/src/org/apache/sandesha/server/queue ServerQueue.java
jaliya 2004/10/09 01:50:23
Modified: sandesha/src/org/apache/sandesha/server
CreateSequenceProcessor.java Sender.java
RMInvoker.java ServerStorageManager.java
AcknowledgementProcessor.java
CreateSequenceResponseProcessor.java
CompositeProcessor.java
RMMessageProcessorIdentifier.java
sandesha/src/org/apache/sandesha/server/queue
ServerQueue.java
Log:
I went through the Chamikara's patch and applied the changes.
Changes are mainly fixing of bugs in the server side message queue
and the SOAP message creation methods
Revision Changes Path
1.8 +14 -7 ws-fx/sandesha/src/org/apache/sandesha/server/CreateSequenceProcessor.java
Index: CreateSequenceProcessor.java
===================================================================
RCS file: /home/cvs/ws-fx/sandesha/src/org/apache/sandesha/server/CreateSequenceProcessor.java,v
retrieving revision 1.7
retrieving revision 1.8
diff -u -r1.7 -r1.8
--- CreateSequenceProcessor.java 13 Sep 2004 11:26:32 -0000 1.7
+++ CreateSequenceProcessor.java 9 Oct 2004 08:50:23 -0000 1.8
@@ -77,19 +77,26 @@
//Set the message type.
rmMessageContext
.setMessageType(org.apache.sandesha.Constants.MSG_TYPE_CREATE_SEQUENCE_RESPONSE);
-
+
+ //FIX THIS FIX THIS
+ //Need to change the ANONYMOUS URI to the new one after completion.
//We have some synchronous stuff here
- if (addrHeaders.getReplyTo().getAddress().toString().equals(
- Constants.NS_URI_ANONYMOUS)) {
- //Do we need anything here??
- //Inform that we have a synchronous response.
- rmMessageContext.getMsgContext().setResponseMessage(
+
+
+ if((addrHeaders.getReplyTo().getAddress().toString().equals(
+ "http://schemas.xmlsoap.org/ws/2003/03/addressing/role/anonymous"))||(addrHeaders.getReplyTo().getAddress().toString().equals(Constants.NS_URI_ANONYMOUS))){
+
+ //Inform that we have a synchronous response.
+
+
+ rmMessageContext.getMsgContext().setResponseMessage(
new Message(resEnvelope));
return true;
} else {
//FIX THIS FIX THIS
//This should be storageManger.addCreateSequenceResponse();
-
+
+
MessageContext msgContext = new MessageContext(rmMessageContext
.getMsgContext().getAxisEngine());
msgContext.setResponseMessage(new Message(resEnvelope));
1.8 +24 -9 ws-fx/sandesha/src/org/apache/sandesha/server/Sender.java
Index: Sender.java
===================================================================
RCS file: /home/cvs/ws-fx/sandesha/src/org/apache/sandesha/server/Sender.java,v
retrieving revision 1.7
retrieving revision 1.8
diff -u -r1.7 -r1.8
--- Sender.java 1 Oct 2004 12:51:26 -0000 1.7
+++ Sender.java 9 Oct 2004 08:50:23 -0000 1.8
@@ -28,6 +28,7 @@
import org.apache.sandesha.IStorageManager;
import org.apache.sandesha.RMException;
import org.apache.sandesha.RMMessageContext;
+import org.apache.sandesha.server.queue.ServerQueue;
/**
* @author JEkanayake
@@ -62,9 +63,9 @@
if (rmMessageContext.getMsgContext() == null)
System.out
.println("rmMessageContext.getMsgContext() == null");
- if (rmMessageContext.getMsgContext().getResponseMessage() == null)
+ if (rmMessageContext.getMsgContext().getRequestMessage() == null)
System.out
- .println("rmMessageContext.getMsgContext().getResponseMessage() == null");
+ .println("rmMessageContext.getMsgContext().getRequestMessage() == null");
switch (rmMessageContext.getMessageType()) {
case Constants.MSG_TYPE_CREATE_SEQUENCE_REQUEST: {
@@ -204,14 +205,25 @@
//RMMessageContext a field to store the long
// lastProcessedTime
//Another field to hold retransmission count.
-
+
+ System.out
+ .println("SENDING RESPONSE MESSAGE .....\n");
+
+ ServerQueue sq = ServerQueue.getInstance();
+ sq.displayPriorityQueue();
+ sq.displayOutgoingMap();
+ sq.displayIncomingMap();
+
SOAPEnvelope responseEnvelope = null;
if (rmMessageContext.getReTransmissionCount() <= Constants.MAXIMUM_RETRANSMISSION_COUNT) {
if ((System.currentTimeMillis() - rmMessageContext
.getLastPrecessedTime()) > Constants.RETRANSMISSION_INTERVAL) {
-
- if (rmMessageContext.getReTransmissionCount() == 0) {
+
+ //TODO
+ //We should do this only once and then need to
+ //Save the respones message.
+ //if (rmMessageContext.getReTransmissionCount() == 0) {
//Need to create the response envelope.
responseEnvelope = EnvelopeCreator
.createServiceResponseEnvelope(rmMessageContext);
@@ -219,14 +231,14 @@
.setRequestMessage(
new Message(
responseEnvelope));
- }
+ System.out.println(responseEnvelope);
+ //}
- System.out
- .println("SENDING RESPONSE MESSAGE .....\n");
+
System.out.println(rmMessageContext
.getAddressingHeaders().getReplyTo()
.getAddress().toString());
- System.out.println(responseEnvelope);
+
try {
Service service = new Service();
Call call = (Call) service.createCall();
@@ -239,6 +251,8 @@
//NOTE: WE USE THE REQUEST MESSAGE TO SEND
// THE RESPONSE.
+
+
call.setRequestMessage(rmMessageContext
.getMsgContext()
.getRequestMessage());
@@ -255,6 +269,7 @@
//We are not expecting the ack over the
// same HTTP connection.
call.invoke();
+ //System.out.println(call.getResponseMessage().getSOAPPartAsString());
} catch (AxisFault e) {
e.printStackTrace();
break;
1.3 +8 -7 ws-fx/sandesha/src/org/apache/sandesha/server/RMInvoker.java
Index: RMInvoker.java
===================================================================
RCS file: /home/cvs/ws-fx/sandesha/src/org/apache/sandesha/server/RMInvoker.java,v
retrieving revision 1.2
retrieving revision 1.3
diff -u -r1.2 -r1.3
--- RMInvoker.java 6 Oct 2004 13:06:55 -0000 1.2
+++ RMInvoker.java 9 Oct 2004 08:50:23 -0000 1.3
@@ -131,14 +131,16 @@
System.out.println("NO RESPONSE SEQUENCE");
RMMessageContext rmMsgContext = new RMMessageContext();
rmMessageContext.copyContents(rmMsgContext);
-
+
MessageContext msgContext = new MessageContext(
rmMessageContext.getMsgContext()
.getAxisEngine());
- RMMessageContext.copyMessageContext(
- rmMessageContext.getMsgContext(),
- msgContext);
+
+ // RMMessageContext.copyMessageContext(
+ // rmMessageContext.getMsgContext(),
+ // msgContext);
//Set this new msgContext to the rmMsgContext.
+
rmMessageContext.setMsgContext(msgContext);
rmMsgContext
@@ -156,9 +158,8 @@
SOAPEnvelope createSequenceEnvelope = EnvelopeCreator
.createCreateSequenceEnvelope(id,
rmMsgContext, Constants.SERVER);
- System.out.println("aaaaa");
- System.out.println("bbbbb");
+
rmMsgContext.getMsgContext().setRequestMessage(
new Message(createSequenceEnvelope));
@@ -167,7 +168,7 @@
.getAddressingHeaders().getReplyTo()
.getAddress().toString());
- System.out.println("ddddd");
+
rmMsgContext.setMessageID("uuid:"+id);
storageManager
1.5 +7 -4 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.4
retrieving revision 1.5
diff -u -r1.4 -r1.5
--- ServerStorageManager.java 29 Sep 2004 11:03:15 -0000 1.4
+++ ServerStorageManager.java 9 Oct 2004 08:50:23 -0000 1.5
@@ -158,7 +158,8 @@
IServerDAO accessor =
ServerDAOFactory.getStorageAccessor(
Constants.SERVER_QUEUE_ACCESSOR);
- return accessor.isIncomingSequenceExists(sequenceID);
+ //return accessor.isIncomingSequenceExists(sequenceID);
+ return accessor.isOutgoingSequenceExists(sequenceID);
}
/**
@@ -328,13 +329,15 @@
boolean done = false;
String sequenceID = accessor.getSequenceOfOutSequence(oldOutsequenceId);
- if(sequenceID==null)
+ if(sequenceID==null){
+ System.out.println("REOVING REMOVING REMOVING REMO " +oldOutsequenceId);
return false;
-
+ }
+ System.out.println("REOVING REMOVING " +oldOutsequenceId);
accessor.setOutSequence(sequenceID,newOutSequenceId);
accessor.setOutSequenceApproved(sequenceID,true);
- //Deleting create sequence message from the priority queue.
+ //Deleting create sequence message from the priority queue.
accessor.removeCreateSequenceMsg(oldOutsequenceId);
return true;
}
1.8 +5 -3 ws-fx/sandesha/src/org/apache/sandesha/server/AcknowledgementProcessor.java
Index: AcknowledgementProcessor.java
===================================================================
RCS file: /home/cvs/ws-fx/sandesha/src/org/apache/sandesha/server/AcknowledgementProcessor.java,v
retrieving revision 1.7
retrieving revision 1.8
diff -u -r1.7 -r1.8
--- AcknowledgementProcessor.java 6 Oct 2004 12:59:03 -0000 1.7
+++ AcknowledgementProcessor.java 9 Oct 2004 08:50:23 -0000 1.8
@@ -43,6 +43,7 @@
}
public boolean processMessage(RMMessageContext rmMessageContext) {
+
SequenceAcknowledgement seqAcknowledgement = rmMessageContext
.getRMHeaders().getSequenceAcknowledgement();
String seqID = seqAcknowledgement.getIdentifier().getIdentifier();
@@ -51,8 +52,9 @@
while (ite.hasNext()) {
AcknowledgementRange ackRange = (AcknowledgementRange) ite.next();
- long msgNumber = ackRange.getMinValue();
+ long msgNumber = ackRange.getMaxValue();
while (ackRange.getMaxValue() >= msgNumber) {
+
storageManger.setAcknowledged(seqID, msgNumber);
//storageManager.setAcknowledged(String seqID,long msgNumber)
msgNumber++;
@@ -164,7 +166,7 @@
if (size > 1) {
for (long i = 1; i <= size; i++) {
- // System.out.println(i);
+
if (i + 1 > size) {
found = true;
@@ -186,7 +188,7 @@
ackRange.setMaxValue(max);
ackRange.setMinValue(min);
vec.add(ackRange);
- //System.out.println("Range " + min + " " + max);
+
min = ((Long) listOfMsgNumbers.get(new Long(i + 1)))
.longValue();
}
1.6 +4 -2 ws-fx/sandesha/src/org/apache/sandesha/server/CreateSequenceResponseProcessor.java
Index: CreateSequenceResponseProcessor.java
===================================================================
RCS file: /home/cvs/ws-fx/sandesha/src/org/apache/sandesha/server/CreateSequenceResponseProcessor.java,v
retrieving revision 1.5
retrieving revision 1.6
diff -u -r1.5 -r1.6
--- CreateSequenceResponseProcessor.java 1 Oct 2004 12:51:26 -0000 1.5
+++ CreateSequenceResponseProcessor.java 9 Oct 2004 08:50:23 -0000 1.6
@@ -44,11 +44,13 @@
RelatesTo relatesTo = (RelatesTo) rmMessageContext
.getAddressingHeaders().getRelatesTo().get(0);
String sequenceID = createSeqRes.getIdentifier().toString();
+
+
storageManager.setApprovedOutSequence(relatesTo.getURI().toString(),sequenceID);
-
+
//Delete the CreateSequence request message in the create sequence
// request queue.
- //storageManger.removeCreateSequenceRequest(String messageID)
+ //storageManager..removeCreateSequenceRequest(messageID);
//No response to this message.
return false;
1.6 +7 -3 ws-fx/sandesha/src/org/apache/sandesha/server/CompositeProcessor.java
Index: CompositeProcessor.java
===================================================================
RCS file: /home/cvs/ws-fx/sandesha/src/org/apache/sandesha/server/CompositeProcessor.java,v
retrieving revision 1.5
retrieving revision 1.6
diff -u -r1.5 -r1.6
--- CompositeProcessor.java 13 Sep 2004 11:26:32 -0000 1.5
+++ CompositeProcessor.java 9 Oct 2004 08:50:23 -0000 1.6
@@ -38,12 +38,13 @@
//if the message is and Ack then process the ack
//if the message has a body then insert it to the inqueue
-
+
RMHeaders rmHeaders = rmMessageContext.getRMHeaders();
AcknowledgementProcessor ackProcessor = new AcknowledgementProcessor(
this.storageManger);
if (rmHeaders.getSequenceAcknowledgement() != null) {
+
ackProcessor.processMessage(rmMessageContext);
}
@@ -54,7 +55,6 @@
.getMessageNumber();
if (storageManger.isMessageExist(sequenceID, messageNumber) != true) {
-
//Create a copy of the RMMessageContext.
RMMessageContext rmMsgContext = new RMMessageContext();
@@ -80,6 +80,10 @@
// context.
msgContext.setRequestMessage(reqMsg);
msgContext.setResponseMessage(resMsg);
+
+ rmMsgContext.setReqEnv(reqMsg.getSOAPEnvelope());
+ rmMsgContext.setResEnv(resMsg.getSOAPEnvelope());
+
rmMsgContext.setMsgContext(msgContext);
//Set the message type for this message.
rmMsgContext
@@ -99,7 +103,7 @@
//This should be changed according to the WS-policy.
return ackProcessor.sendAcknowledgement(rmMessageContext);
}
-
+
return false;
}
1.6 +2 -1 ws-fx/sandesha/src/org/apache/sandesha/server/RMMessageProcessorIdentifier.java
Index: RMMessageProcessorIdentifier.java
===================================================================
RCS file: /home/cvs/ws-fx/sandesha/src/org/apache/sandesha/server/RMMessageProcessorIdentifier.java,v
retrieving revision 1.5
retrieving revision 1.6
diff -u -r1.5 -r1.6
--- RMMessageProcessorIdentifier.java 13 Sep 2004 11:26:32 -0000 1.5
+++ RMMessageProcessorIdentifier.java 9 Oct 2004 08:50:23 -0000 1.6
@@ -32,7 +32,7 @@
AddressingHeaders addrHeaders = rmMessageContext.getAddressingHeaders();
RMHeaders rmHeaders = rmMessageContext.getRMHeaders();
-
+
if (addrHeaders.getAction().toString().equals(
Constants.ACTION_CREATE_SEQUENCE)) {
return new CreateSequenceProcessor(storageManager);
@@ -44,6 +44,7 @@
return new TerminateSequenceProcessor(storageManager);
} else if ((rmHeaders.getSequenceAcknowledgement() != null)
|| (rmHeaders.getSequence().getMessageNumber() != null)) {
+
return new CompositeProcessor(storageManager);
} else
return new FaultProcessor(storageManager);
1.4 +16 -8 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.3
retrieving revision 1.4
diff -u -r1.3 -r1.4
--- ServerQueue.java 20 Sep 2004 12:17:07 -0000 1.3
+++ ServerQueue.java 9 Oct 2004 08:50:23 -0000 1.4
@@ -160,7 +160,7 @@
public boolean isIncomingSequenceExists(String seqId) {
synchronized (incomingMap) {
-
+
return incomingMap.containsKey(seqId);
}
}
@@ -235,6 +235,8 @@
tempMsg = rsh.getNextMessageToSend();
if (tempMsg != null) {
msg = tempMsg;
+ msg.setSequenceID(rsh.getOutSequenceId());
+ msg.setOldSequenceID(rsh.getSequenceId());
break whileLoop;
}
}
@@ -251,6 +253,7 @@
SequenceHash sh = new SequenceHash(sequenceId);
incomingMap.put(sequenceId, sh);
+
}
}
@@ -492,17 +495,20 @@
}
public String getSequenceOfOutSequence(String outSequence) {
- if (outSequence == null)
- return null;
-
+ if (outSequence == null){
+ return null;
+ }
Iterator it = outgoingMap.keySet().iterator();
synchronized (outgoingMap) {
while (it.hasNext()) {
+
+
String tempSeqId = (String) it.next();
+
ResponseSequenceHash rsh = (ResponseSequenceHash) outgoingMap
.get(tempSeqId);
String tempOutSequence = rsh.getOutSequenceId();
-
+
if (outSequence.equals(tempOutSequence))
return tempSeqId;
@@ -572,9 +578,11 @@
}
public void moveOutgoingMsgToBin(String sequenceId, Long messageNo) {
+ String sequence = getSequenceOfOutSequence(sequenceId);
ResponseSequenceHash rsh = (ResponseSequenceHash) outgoingMap
- .get(sequenceId);
-
+ .get(sequence);
+
+
if (rsh == null) {
System.out.println("ERROR: RESPONSE SEQ IS NULL");
return;
@@ -627,7 +635,7 @@
l = temp;
}
- return (l+1);
+ return (l);
}
}