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/10/02 19:54:08 UTC
svn commit: r293139 - in
/webservices/sandesha/trunk/src/org/apache/sandesha2: Constants.java
RMMsgCreator.java handlers/ServerOutHandler.java
msgprocessors/AcknowledgementProcessor.java util/SandeshaUtil.java
Author: chamikara
Date: Sun Oct 2 10:53:05 2005
New Revision: 293139
URL: http://svn.apache.org/viewcvs?rev=293139&view=rev
Log:
Complete message exchange is working for the echo case single input/output message. I.e.
Sandesha1 client Vs Sandesha2 server
------------------------------------
--> create sequence
<-- create sequence response
--> application message
<-- acknowledgement
--> terminate sequence
<-- create sequence request
--> create sequence response
<-- application message (retransmitted)
--> acknowledgement
<-- terminate sequence
Modified:
webservices/sandesha/trunk/src/org/apache/sandesha2/Constants.java
webservices/sandesha/trunk/src/org/apache/sandesha2/RMMsgCreator.java
webservices/sandesha/trunk/src/org/apache/sandesha2/handlers/ServerOutHandler.java
webservices/sandesha/trunk/src/org/apache/sandesha2/msgprocessors/AcknowledgementProcessor.java
webservices/sandesha/trunk/src/org/apache/sandesha2/util/SandeshaUtil.java
Modified: webservices/sandesha/trunk/src/org/apache/sandesha2/Constants.java
URL: http://svn.apache.org/viewcvs/webservices/sandesha/trunk/src/org/apache/sandesha2/Constants.java?rev=293139&r1=293138&r2=293139&view=diff
==============================================================================
--- webservices/sandesha/trunk/src/org/apache/sandesha2/Constants.java (original)
+++ webservices/sandesha/trunk/src/org/apache/sandesha2/Constants.java Sun Oct 2 10:53:05 2005
@@ -73,6 +73,17 @@
String IDENTIFIER = "Identifier";
String ACCEPT = "Accept";
+
+ public interface Actions {
+
+ String CREATE_SEQUENCE = "http://schemas.xmlsoap.org/ws/2005/02/rm/CreateSequence";
+
+ String CREATE_SEQUENCE_RESPONSE = "http://schemas.xmlsoap.org/ws/2005/02/rm/CreateSequenceResponse";
+
+ String SEQUENCE_ACKNOWLEDGEMENT = "http://schemas.xmlsoap.org/ws/2005/02/rm/SequenceAcknowledgement";
+
+ String TERMINATE_SEQUENCE = "http://schemas.xmlsoap.org/ws/2005/02/rm/SequenceAcknowledgement";
+ }
}
public interface WSA {
@@ -145,6 +156,8 @@
String OUT_CREATE_SEQUENCE_SENT = "OutCreateSeqSent";
String NEXT_MESSAGE_NUMBER = "NextMsgNo";
+
+ String LAST_OUT_MESSAGE = "LastOutMessage";
}
public interface SOAPVersion {
Modified: webservices/sandesha/trunk/src/org/apache/sandesha2/RMMsgCreator.java
URL: http://svn.apache.org/viewcvs/webservices/sandesha/trunk/src/org/apache/sandesha2/RMMsgCreator.java?rev=293139&r1=293138&r2=293139&view=diff
==============================================================================
--- webservices/sandesha/trunk/src/org/apache/sandesha2/RMMsgCreator.java (original)
+++ webservices/sandesha/trunk/src/org/apache/sandesha2/RMMsgCreator.java Sun Oct 2 10:53:05 2005
@@ -51,6 +51,7 @@
import org.apache.sandesha2.wsrm.Identifier;
import org.apache.sandesha2.wsrm.Sequence;
import org.apache.sandesha2.wsrm.SequenceAcknowledgement;
+import org.apache.sandesha2.wsrm.TerminateSequence;
import org.apache.wsdl.WSDLConstants;
/**
@@ -160,6 +161,56 @@
createSeqMsg.setServiceContextID(applicationMsg.getServiceContextID());
return createSeqRMMsg;
+ }
+
+ public static RMMsgContext createTerminateSequenceMessage (RMMsgContext referenceRMMessage, String sequenceId) throws SandeshaException {
+ MessageContext referenceMessage = referenceRMMessage.getMessageContext();
+ if (referenceMessage==null)
+ throw new SandeshaException ("MessageContext is null");
+
+ RMMsgContext terminateRMMessage = SandeshaUtil.shallowCopy(referenceRMMessage);
+ MessageContext terminateMessage = terminateRMMessage.getMessageContext();
+ if (terminateMessage==null)
+ throw new SandeshaException ("MessageContext is null");
+
+ MessageInformationHeaders newMessageInfoHeaders = new MessageInformationHeaders ();
+ terminateMessage.setMessageInformationHeaders(newMessageInfoHeaders);
+ terminateMessage.setMessageID(SandeshaUtil.getUUID());
+
+ terminateMessage.setServiceGroupContext(referenceMessage
+ .getServiceGroupContext());
+ terminateMessage.setServiceGroupContextId(referenceMessage
+ .getServiceGroupContextId());
+ terminateMessage.setServiceContext(referenceMessage
+ .getServiceContext());
+ terminateMessage.setServiceContextID(referenceMessage
+ .getServiceContextID());
+
+ terminateMessage.setOperationDescription(referenceMessage.getOperationDescription());
+ OperationContext newOperationCtx = new OperationContext (terminateMessage.getOperationDescription());
+ try {
+ newOperationCtx.addMessageContext(terminateMessage);
+ } catch (AxisFault e) {
+ throw new SandeshaException (e.getMessage());
+ }
+
+ terminateMessage.setOperationContext(newOperationCtx);
+
+ ConfigurationContext configCtx = terminateMessage.getSystemContext();
+ if (configCtx==null)
+ throw new SandeshaException ("Configuration Context is null");
+ configCtx.registerOperationContext(terminateMessage.getMessageID(),newOperationCtx);
+
+ SOAPEnvelope envelope = SOAPAbstractFactory.getSOAPFactory(Constants.SOAPVersion.DEFAULT).getDefaultEnvelope();
+ terminateRMMessage.setSOAPEnvelop(envelope);
+
+ TerminateSequence terminateSequencePart = new TerminateSequence ();
+ Identifier identifier = new Identifier ();
+ identifier.setIndentifer(sequenceId);
+ terminateSequencePart.setIdentifier(identifier);
+ terminateRMMessage.setMessagePart(Constants.MessageParts.TERMINATE_SEQ,terminateSequencePart);
+
+ return terminateRMMessage;
}
public static RMMsgContext createCreateSeqResponseMsg(
Modified: webservices/sandesha/trunk/src/org/apache/sandesha2/handlers/ServerOutHandler.java
URL: http://svn.apache.org/viewcvs/webservices/sandesha/trunk/src/org/apache/sandesha2/handlers/ServerOutHandler.java?rev=293139&r1=293138&r2=293139&view=diff
==============================================================================
--- webservices/sandesha/trunk/src/org/apache/sandesha2/handlers/ServerOutHandler.java (original)
+++ webservices/sandesha/trunk/src/org/apache/sandesha2/handlers/ServerOutHandler.java Sun Oct 2 10:53:05 2005
@@ -53,6 +53,7 @@
import org.apache.sandesha2.storage.beans.StorageMapBean;
import org.apache.sandesha2.util.SandeshaUtil;
import org.apache.sandesha2.wsrm.Identifier;
+import org.apache.sandesha2.wsrm.LastMessage;
import org.apache.sandesha2.wsrm.MessageNumber;
import org.apache.sandesha2.wsrm.Sequence;
import org.apache.wsdl.MessageReference;
@@ -292,6 +293,27 @@
appMsgEntry.setMessageNumber(nextMsgNo);
+ //setting last message
+ if (msg.isServerSide()) {
+ Sequence requestSequence = (Sequence) reqRMMsg
+ .getMessagePart(Constants.MessageParts.SEQUENCE);
+ if (requestSequence == null)
+ throw new SandeshaException("Request Sequence is null");
+
+ if (requestSequence.getLastMessage() != null) {
+ //FIX ME - This fails if request last message has more than one
+ // responses.
+ sequence.setLastMessage(new LastMessage());
+
+ //saving the last message no.
+ SequencePropertyBean lastOutMsgBean = new SequencePropertyBean(
+ incomingSeqId,
+ Constants.SequenceProperties.LAST_OUT_MESSAGE,
+ new Long(nextMsgNo));
+ sequencePropertyMgr.insert(lastOutMsgBean);
+ }
+ }
+
if (outSequenceBean == null || outSequenceBean.getValue() == null) {
//setting tempSequenceId;
@@ -355,7 +377,6 @@
String tempSequeneId = (String) tempSequenceBean.getValue();
appMsgEntry.setTempSequenceId(tempSequeneId);
retransmitterMgr.insert(appMsgEntry);
-
}
private long getNextMsgNo(ConfigurationContext context, String incomingSeqId) {
Modified: webservices/sandesha/trunk/src/org/apache/sandesha2/msgprocessors/AcknowledgementProcessor.java
URL: http://svn.apache.org/viewcvs/webservices/sandesha/trunk/src/org/apache/sandesha2/msgprocessors/AcknowledgementProcessor.java?rev=293139&r1=293138&r2=293139&view=diff
==============================================================================
--- webservices/sandesha/trunk/src/org/apache/sandesha2/msgprocessors/AcknowledgementProcessor.java (original)
+++ webservices/sandesha/trunk/src/org/apache/sandesha2/msgprocessors/AcknowledgementProcessor.java Sun Oct 2 10:53:05 2005
@@ -21,17 +21,28 @@
import java.util.Iterator;
import java.util.List;
+import javax.xml.stream.FactoryConfigurationError;
+import javax.xml.stream.XMLOutputFactory;
+import javax.xml.stream.XMLStreamException;
+import javax.xml.stream.XMLStreamWriter;
+
+import org.apache.axis2.AxisFault;
+import org.apache.axis2.addressing.EndpointReference;
import org.apache.axis2.context.AbstractContext;
import org.apache.axis2.context.ConfigurationContext;
+import org.apache.axis2.soap.SOAPEnvelope;
import org.apache.sandesha2.Constants;
import org.apache.sandesha2.RMMsgContext;
+import org.apache.sandesha2.RMMsgCreator;
import org.apache.sandesha2.SandeshaException;
import org.apache.sandesha2.storage.AbstractBeanMgrFactory;
import org.apache.sandesha2.storage.beanmanagers.RetransmitterBeanMgr;
import org.apache.sandesha2.storage.beanmanagers.SequencePropertyBeanMgr;
import org.apache.sandesha2.storage.beans.RetransmitterBean;
import org.apache.sandesha2.storage.beans.SequencePropertyBean;
+import org.apache.sandesha2.util.SandeshaUtil;
import org.apache.sandesha2.wsrm.AcknowledgementRange;
+import org.apache.sandesha2.wsrm.LastMessage;
import org.apache.sandesha2.wsrm.Nack;
import org.apache.sandesha2.wsrm.SequenceAcknowledgement;
@@ -109,6 +120,31 @@
//TODO - Process Nack
}
+
+ //If all messages up to last message have been acknowledged.
+ //Add terminate Sequence message.
+ SequencePropertyBean lastOutMsgBean = seqPropMgr.retrieve(
+ incomingSequenceId,
+ Constants.SequenceProperties.LAST_OUT_MESSAGE);
+ if (lastOutMsgBean != null) {
+ Long lastOutMsgNoLng = (Long) lastOutMsgBean.getValue();
+ if (lastOutMsgNoLng == null)
+ throw new SandeshaException(
+ "Invalid object set for the Last Out Message");
+
+ long lastOutMessageNo = lastOutMsgNoLng.longValue();
+ if (lastOutMessageNo <= 0)
+ throw new SandeshaException(
+ "Invalid value set for the last out message");
+
+ boolean complete = SandeshaUtil.verifySequenceCompletion(
+ sequenceAck.getAcknowledgementRanges().iterator(),
+ lastOutMessageNo);
+ if (complete) {
+ addTerminateSequenceMessage(rmMsgCtx, outSequenceId,incomingSequenceId);
+ }
+ }
+
}
private RetransmitterBean getRetransmitterEntry(Collection collection,
@@ -122,4 +158,59 @@
return null;
}
+
+ public void addTerminateSequenceMessage(RMMsgContext incomingAckRMMsg,
+ String outSequenceId,String incomingSequenceId) throws SandeshaException {
+ RMMsgContext terminateRMMessage = RMMsgCreator
+ .createTerminateSequenceMessage(incomingAckRMMsg, outSequenceId);
+
+ //detting addressing headers.
+ SequencePropertyBeanMgr seqPropMgr = AbstractBeanMgrFactory.getInstance(
+ incomingAckRMMsg.getContext()).getSequencePropretyBeanMgr();
+ SequencePropertyBean replyToBean = seqPropMgr.retrieve(incomingSequenceId,Constants.SequenceProperties.REPLY_TO_EPR);
+ if (replyToBean==null)
+ throw new SandeshaException ("ReplyTo property is not set");
+
+ EndpointReference replyToEPR = (EndpointReference) replyToBean.getValue();
+ if (replyToEPR==null)
+ throw new SandeshaException ("ReplyTo EPR has an invalid value");
+
+ terminateRMMessage.setTo(new EndpointReference (replyToEPR.getAddress()));
+ terminateRMMessage.setFrom(incomingAckRMMsg.getTo());
+ terminateRMMessage
+ .setWSAAction(Constants.WSRM.Actions.TERMINATE_SEQUENCE);
+
+ try {
+ terminateRMMessage.addSOAPEnvelope();
+ } catch (AxisFault e) {
+ throw new SandeshaException(e.getMessage());
+ }
+
+ String key = SandeshaUtil.storeMessageContext(terminateRMMessage
+ .getMessageContext());
+ RetransmitterBean terminateBean = new RetransmitterBean();
+ terminateBean.setKey(key);
+ terminateBean.setLastSentTime(0);
+ terminateBean.setMessageId(terminateRMMessage.getMessageId());
+ terminateBean.setSend(true);
+
+ RetransmitterBeanMgr retramsmitterMgr = AbstractBeanMgrFactory.getInstance(
+ incomingAckRMMsg.getContext()).getRetransmitterBeanMgr();
+ retramsmitterMgr.insert(terminateBean);
+
+
+ try {
+ System.out.println("SERIALIZING TERMINATE MSG");
+ SOAPEnvelope envel = terminateRMMessage.getSOAPEnvelope();
+ XMLStreamWriter writer = XMLOutputFactory.newInstance().createXMLStreamWriter(System.out);
+ envel.serialize(writer);
+ } catch (XMLStreamException e1) {
+ // TODO Auto-generated catch block
+ e1.printStackTrace();
+ } catch (FactoryConfigurationError e1) {
+ // TODO Auto-generated catch block
+ e1.printStackTrace();
+ }
+ }
+
}
Modified: webservices/sandesha/trunk/src/org/apache/sandesha2/util/SandeshaUtil.java
URL: http://svn.apache.org/viewcvs/webservices/sandesha/trunk/src/org/apache/sandesha2/util/SandeshaUtil.java?rev=293139&r1=293138&r2=293139&view=diff
==============================================================================
--- webservices/sandesha/trunk/src/org/apache/sandesha2/util/SandeshaUtil.java (original)
+++ webservices/sandesha/trunk/src/org/apache/sandesha2/util/SandeshaUtil.java Sun Oct 2 10:53:05 2005
@@ -18,7 +18,10 @@
import java.awt.datatransfer.StringSelection;
import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
import java.util.Hashtable;
+import java.util.Iterator;
import java.util.StringTokenizer;
import org.apache.axis2.AxisFault;
@@ -197,8 +200,9 @@
TransportInDescription transportIn = msgCtx.getTransportIn();
TransportOutDescription transportOut = msgCtx.getTransportOut();
MessageInformationHeaders msgInfoHeaders1 = new MessageInformationHeaders();
- MessageInformationHeaders oldMsgInfoHeaders = msgCtx.getMessageInformationHeaders();
-
+ MessageInformationHeaders oldMsgInfoHeaders = msgCtx
+ .getMessageInformationHeaders();
+
msgInfoHeaders1.setTo(oldMsgInfoHeaders.getTo());
msgInfoHeaders1.setFrom(oldMsgInfoHeaders.getFrom());
msgInfoHeaders1.setReplyTo(oldMsgInfoHeaders.getReplyTo());
@@ -206,10 +210,11 @@
msgInfoHeaders1.setMessageId(getUUID());
msgInfoHeaders1.setRelatesTo(oldMsgInfoHeaders.getRelatesTo());
msgInfoHeaders1.setAction(oldMsgInfoHeaders.getAction());
- msgInfoHeaders1.setReferenceParameters(oldMsgInfoHeaders.getReferenceParameters());
-
+ msgInfoHeaders1.setReferenceParameters(oldMsgInfoHeaders
+ .getReferenceParameters());
+
try {
-
+
MessageContext newMessageContext = new MessageContext(configCtx,
transportIn, transportOut);
newMessageContext.setProperty(MessageContext.TRANSPORT_OUT, msgCtx
@@ -253,9 +258,51 @@
return newRMMsgCtx;
}
+ public static RMMsgContext shallowCopy(RMMsgContext rmMsgContext)
+ throws SandeshaException {
+ MessageContext msgCtx = null;
+ if (rmMsgContext.getMessageContext() != null)
+ msgCtx = shallowCopy(rmMsgContext.getMessageContext());
+
+ RMMsgContext newRMMsgCtx = new RMMsgContext();
+ if (msgCtx != null)
+ newRMMsgCtx.setMessageContext(msgCtx);
+
+ return newRMMsgCtx;
+ }
+
public static void startSenderIfStopped(ConfigurationContext context) {
if (!sender.isSenderStarted()) {
sender.start(context);
}
+ }
+
+ public static boolean verifySequenceCompletion(Iterator ackRangesIterator,
+ long lastMessageNo) {
+ HashMap startMap = new HashMap();
+
+ while (ackRangesIterator.hasNext()) {
+ AcknowledgementRange temp = (AcknowledgementRange) ackRangesIterator
+ .next();
+ startMap.put(new Long(temp.getLowerValue()), temp);
+ }
+
+ long start = 1;
+ boolean loop = true;
+ while (loop) {
+ AcknowledgementRange temp = (AcknowledgementRange) startMap
+ .get(new Long(start));
+ if (temp == null) {
+ loop = false;
+ continue;
+ }
+
+ if (temp.getUpperValue() >= lastMessageNo)
+ return true;
+
+ start = temp.getUpperValue() + 1;
+ }
+
+ return false;
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: sandesha-dev-unsubscribe@ws.apache.org
For additional commands, e-mail: sandesha-dev-help@ws.apache.org