Author: chamikara
Date: Sun Oct 30 21:19:04 2005
New Revision: 329749
URL: http://svn.apache.org/viewcvs?rev=329749&view=rev
Log:
Added the SandeshsaGlobal handler. This is used to put logic that could be
carried out globally (to apply for all services). Because of the current way
Axis2 have been implemented, had to make sandesha a global module. The service
specific handlers will check a property in the services.xml to detect weather
RM is anabled for the service.
Added following loging to the global handler.
1. Eliminating duplicate messages.
2. Setting relatesTo value of ack messages to null (some framesowors (e.g.
indigo) tend to send a relatesTo value for the ack which could cause an invalid
dispatch).
3. Sending ack messages for duplicate messages if required.
did some corrections in the Storage classes.
Added:
webservices/sandesha/trunk/src/org/apache/sandesha2/handlers/SandeshaGlobalInHandler.java
Modified:
webservices/sandesha/trunk/src/org/apache/sandesha2/Constants.java
webservices/sandesha/trunk/src/org/apache/sandesha2/RMMsgContext.java
webservices/sandesha/trunk/src/org/apache/sandesha2/handlers/SandeshaOutHandler.java
webservices/sandesha/trunk/src/org/apache/sandesha2/msgprocessors/AcknowledgementProcessor.java
webservices/sandesha/trunk/src/org/apache/sandesha2/msgprocessors/ApplicationMsgProcessor.java
webservices/sandesha/trunk/src/org/apache/sandesha2/msgprocessors/TerminateSeqMsgProcessor.java
webservices/sandesha/trunk/src/org/apache/sandesha2/storage/beanmanagers/CreateSeqBeanMgr.java
webservices/sandesha/trunk/src/org/apache/sandesha2/storage/beanmanagers/NextMsgBeanMgr.java
webservices/sandesha/trunk/src/org/apache/sandesha2/storage/beanmanagers/RetransmitterBeanMgr.java
webservices/sandesha/trunk/src/org/apache/sandesha2/storage/beanmanagers/SequencePropertyBeanMgr.java
webservices/sandesha/trunk/src/org/apache/sandesha2/storage/beans/StorageMapBean.java
webservices/sandesha/trunk/src/org/apache/sandesha2/util/RMMsgCreator.java
webservices/sandesha/trunk/src/org/apache/sandesha2/util/SandeshaUtil.java
webservices/sandesha/trunk/src/org/apache/sandesha2/workers/InOrderInvoker.java
webservices/sandesha/trunk/src/org/apache/sandesha2/workers/Sender.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=329749&r1=329748&r2=329749&view=diff
==============================================================================
--- webservices/sandesha/trunk/src/org/apache/sandesha2/Constants.java
(original)
+++ webservices/sandesha/trunk/src/org/apache/sandesha2/Constants.java Sun Oct
30 21:19:04 2005
@@ -17,6 +17,8 @@
package org.apache.sandesha2;
+
+
/**
* @author Chamikara
* @author Sanka
@@ -97,7 +99,7 @@
}
public interface WSP {
- long RETRANSMISSION_INTERVAL = 20000;
+ long RETRANSMISSION_INTERVAL = 15000;
}
public interface MessageTypes {
@@ -228,6 +230,8 @@
String OUT_HANDLER_NAME = "SandeshaOutHandler";
+ String GLOBAL_IN_HANDLER_NAME= "GlobalInHandler";
+
String SEQUENCE_KEY = "SequenceKey";
//message context properties
@@ -250,5 +254,7 @@
String AcksTo = "AcksToProperty";
String OFFERED_SEQUENCE_ID = "OfferedSequenceId";
+
+ String ACK_PROCSSED = "AckProcessed";
}
Modified: webservices/sandesha/trunk/src/org/apache/sandesha2/RMMsgContext.java
URL:
http://svn.apache.org/viewcvs/webservices/sandesha/trunk/src/org/apache/sandesha2/RMMsgContext.java?rev=329749&r1=329748&r2=329749&view=diff
==============================================================================
--- webservices/sandesha/trunk/src/org/apache/sandesha2/RMMsgContext.java
(original)
+++ webservices/sandesha/trunk/src/org/apache/sandesha2/RMMsgContext.java Sun
Oct 30 21:19:04 2005
@@ -121,6 +121,14 @@
public String getMessageId() {
return msgContext.getMessageID();
}
+
+ public void setFaultTo (EndpointReference epr) {
+ msgContext.setFaultTo(epr);
+ }
+
+ public EndpointReference getFaultTo () {
+ return msgContext.getFaultTo();
+ }
public SOAPEnvelope getSOAPEnvelope() {
return msgContext.getEnvelope();
@@ -209,4 +217,7 @@
return msgContext.getSystemContext();
}
+ public void setSOAPAction (String SOAPAction) {
+ msgContext.setSoapAction(SOAPAction);
+ }
}
Added:
webservices/sandesha/trunk/src/org/apache/sandesha2/handlers/SandeshaGlobalInHandler.java
URL:
http://svn.apache.org/viewcvs/webservices/sandesha/trunk/src/org/apache/sandesha2/handlers/SandeshaGlobalInHandler.java?rev=329749&view=auto
==============================================================================
---
webservices/sandesha/trunk/src/org/apache/sandesha2/handlers/SandeshaGlobalInHandler.java
(added)
+++
webservices/sandesha/trunk/src/org/apache/sandesha2/handlers/SandeshaGlobalInHandler.java
Sun Oct 30 21:19:04 2005
@@ -0,0 +1,144 @@
+/*
+ * Copyright 1999-2004 The Apache Software Foundation.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not
+ * use this file except in compliance with the License. You may obtain a copy
of
+ * the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
under
+ * the License.
+ *
+ */
+
+/**
+ * @author Chamikara
+ * @author Sanka
+ */
+
+package org.apache.sandesha2.handlers;
+
+import java.util.ArrayList;
+import java.util.Collection;
+
+import javax.xml.namespace.QName;
+
+import org.apache.axis2.AxisFault;
+import org.apache.axis2.context.MessageContext;
+import org.apache.axis2.handlers.AbstractHandler;
+import org.apache.sandesha2.Constants;
+import org.apache.sandesha2.RMMsgContext;
+import org.apache.sandesha2.SandeshaException;
+import org.apache.sandesha2.msgprocessors.ApplicationMsgProcessor;
+import org.apache.sandesha2.msgprocessors.MsgProcessor;
+import org.apache.sandesha2.msgprocessors.MsgProcessorFactory;
+import org.apache.sandesha2.storage.AbstractBeanMgrFactory;
+import org.apache.sandesha2.storage.beanmanagers.SequencePropertyBeanMgr;
+import org.apache.sandesha2.storage.beanmanagers.StorageMapBeanMgr;
+import org.apache.sandesha2.storage.beans.SequencePropertyBean;
+import org.apache.sandesha2.storage.beans.StorageMapBean;
+import org.apache.sandesha2.util.MsgInitializer;
+import org.apache.sandesha2.util.SandeshaUtil;
+import org.apache.sandesha2.wsrm.Sequence;
+
+public class SandeshaGlobalInHandler extends AbstractHandler {
+
+ public void invoke(MessageContext msgContext) throws AxisFault {
+
+
+ try {
+ RMMsgContext rmMessageContext = MsgInitializer
+ .initializeMessage(msgContext);
+
+ //Dropping duplicates
+ boolean dropped = dropIfDuplicate (rmMessageContext);
+ if (dropped) {
+ processDroppedMessage (rmMessageContext);
+ return;
+ }
+
+ //Process if global processing possible. - Currently
none
+ if
(SandeshaUtil.isGloballyProcessableMessageType(rmMessageContext
+ .getMessageType())) {
+ doGlobalProcessing (rmMessageContext);
+ }
+
+ } catch (SandeshaException e) {
+ throw new AxisFault(e.getMessage());
+ }
+
+ }
+
+ private boolean dropIfDuplicate (RMMsgContext rmMsgContext) throws
SandeshaException {
+
+ boolean drop = false;
+
+ if
(rmMsgContext.getMessageType()==Constants.MessageTypes.APPLICATION) {
+ Sequence sequence = (Sequence)
rmMsgContext.getMessagePart(Constants.MessageParts.SEQUENCE);
+ String sequenceId = null;
+
+ if (sequence!=null) {
+ sequenceId =
sequence.getIdentifier().getIdentifier();
+ }
+
+ long msgNo =
sequence.getMessageNumber().getMessageNumber();
+
+ if (sequenceId!=null && msgNo>0) {
+ SequencePropertyBeanMgr seqPropMgr =
AbstractBeanMgrFactory.getInstance(rmMsgContext.getContext()).getSequencePropretyBeanMgr();
+ SequencePropertyBean receivedMsgsBean =
seqPropMgr.retrieve(sequenceId,Constants.SequenceProperties.RECEIVED_MESSAGES);
+ if (receivedMsgsBean!=null) {
+ String receivedMsgStr = (String)
receivedMsgsBean.getValue();
+ ArrayList msgNoArrList =
SandeshaUtil.getSplittedMsgNoArraylist(receivedMsgStr);
+
+ if (msgNoArrList.contains(new Long
(msgNo).toString())){
+ drop = true;
+ }
+ }
+ }
+ }
+
+ if (drop) {
+
rmMsgContext.getMessageContext().setPausedTrue(getName());
+ return true;
+ }
+
+ return false;
+ }
+
+ private void processDroppedMessage (RMMsgContext rmMsgContext) throws
SandeshaException {
+ if
(rmMsgContext.getMessageType()==Constants.MessageTypes.APPLICATION) {
+ Sequence sequence = (Sequence)
rmMsgContext.getMessagePart(Constants.MessageParts.SEQUENCE);
+ String sequenceId = null;
+
+ if (sequence!=null) {
+ sequenceId =
sequence.getIdentifier().getIdentifier();
+ }
+
+ SequencePropertyBeanMgr seqPropMgr =
AbstractBeanMgrFactory.getInstance(rmMsgContext.getContext()).getSequencePropretyBeanMgr();
+ SequencePropertyBean receivedMsgsBean =
seqPropMgr.retrieve(sequenceId,Constants.SequenceProperties.RECEIVED_MESSAGES);
+ String receivedMsgStr = (String)
receivedMsgsBean.getValue();
+
+ ApplicationMsgProcessor ackProcessor = new
ApplicationMsgProcessor ();
+ //Even though the duplicate message is dropped, hv to
send the ack if needed.
+
ackProcessor.sendAckIfNeeded(rmMsgContext,receivedMsgStr);
+
+
+ }
+ }
+
+ private void doGlobalProcessing (RMMsgContext rmMsgCtx) throws
SandeshaException {
+ switch (rmMsgCtx.getMessageType()) {
+ case Constants.MessageTypes.ACK:
+ rmMsgCtx.setRelatesTo(null); //Removing the
relatesTo part from ackMessageIf present.
+
//Some Frameworks tend to send this.
+ }
+ }
+
+ public QName getName () {
+ return new QName (Constants.GLOBAL_IN_HANDLER_NAME);
+ }
+}
\ No newline at end of file
Modified:
webservices/sandesha/trunk/src/org/apache/sandesha2/handlers/SandeshaOutHandler.java
URL:
http://svn.apache.org/viewcvs/webservices/sandesha/trunk/src/org/apache/sandesha2/handlers/SandeshaOutHandler.java?rev=329749&r1=329748&r2=329749&view=diff
==============================================================================
---
webservices/sandesha/trunk/src/org/apache/sandesha2/handlers/SandeshaOutHandler.java
(original)
+++
webservices/sandesha/trunk/src/org/apache/sandesha2/handlers/SandeshaOutHandler.java
Sun Oct 30 21:19:04 2005
@@ -75,7 +75,7 @@
} catch (SandeshaException ex) {
throw new AxisFault("Cant initialize the message");
}
-
+
//TODO recheck
//continue only if an possible application message
if (!(rmMsgCtx.getMessageType() ==
Constants.MessageTypes.UNKNOWN)) {
@@ -188,11 +188,19 @@
.getProperty(Constants.AcksTo);
//If acksTo is not anonymous. Start the
listner TODO: verify
- if
(!Constants.WSA.NS_URI_ANONYMOUS.equals(acksTo)) {
+ if
(!Constants.WSA.NS_URI_ANONYMOUS.equals(acksTo) && !serverSide) {
String transportIn = (String)
context.getProperty(MessageContext.TRANSPORT_IN);
if (transportIn==null)
transportIn =
org.apache.axis2.Constants.TRANSPORT_HTTP;
ListenerManager.makeSureStarted(transportIn,context);
+ }else if (acksTo==null && serverSide) {
+ String incomingSequencId =
SandeshaUtil.getServerSideIncomingSeqIdFromInternalSeqId(tempSequenceId);
+ SequencePropertyBean bean =
seqPropMgr.retrieve(incomingSequencId,Constants.SequenceProperties.REPLY_TO_EPR);
+ if (bean!=null) {
+ EndpointReference
acksToEPR = (EndpointReference) bean.getValue();
+ if (acksToEPR!=null)
+ acksTo =
(String) acksToEPR.getAddress();
+ }
}
addCreateSequenceMessage(rmMsgCtx,
tempSequenceId, acksTo);
@@ -294,18 +302,23 @@
//
("http://localhost:9070/somethingWorking";));
//Setting WSA Action if null
- //TODO: Recheck weather this action is
correct
- if (msgCtx.getWSAAction() == null) {
- EndpointReference toEPR =
msgCtx.getTo();
+ //TODO: Recheck weather this actions
are correct
+ EndpointReference toEPR =
msgCtx.getTo();
- if (toEPR == null)
- throw new
SandeshaException("To EPR is not found");
+ if (toEPR == null)
+ throw new SandeshaException("To
EPR is not found");
- String to = toEPR.getAddress();
- String operationName =
msgCtx.getOperationContext()
-
.getAxisOperation().getName()
- .getLocalPart();
+ String to = toEPR.getAddress();
+ String operationName =
msgCtx.getOperationContext()
+
.getAxisOperation().getName()
+ .getLocalPart();
+
+ if (msgCtx.getWSAAction() == null) {
msgCtx.setWSAAction(to + "/" +
operationName);
+ }
+
+ if (msgCtx.getSoapAction()==null) {
+ msgCtx.setSoapAction("\"" +
to+"/" + operationName + "\"");
}
//processing the response
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=329749&r1=329748&r2=329749&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 30 21:19:04 2005
@@ -55,13 +55,8 @@
public class AcknowledgementProcessor implements MsgProcessor {
-// public static void notifyAllWaitingOnKey () {
-//
-// SandeshaOutHandler.key.notifyAll();
-// }
-
public void processMessage(RMMsgContext rmMsgCtx) throws
SandeshaException {
-
+
SequenceAcknowledgement sequenceAck = (SequenceAcknowledgement)
rmMsgCtx
.getMessagePart(Constants.MessageParts.SEQ_ACKNOWLEDGEMENT);
if (sequenceAck == null)
@@ -71,32 +66,38 @@
if (context == null)
throw new SandeshaException("Context is null");
+
+ RetransmitterBeanMgr retransmitterMgr = AbstractBeanMgrFactory
+ .getInstance(context).getRetransmitterBeanMgr();
+ SequencePropertyBeanMgr seqPropMgr = AbstractBeanMgrFactory
+
.getInstance(context).getSequencePropretyBeanMgr();
+
Iterator ackRangeIterator =
sequenceAck.getAcknowledgementRanges()
.iterator();
-
+
Iterator nackIterator = sequenceAck.getNackList().iterator();
String outSequenceId =
sequenceAck.getIdentifier().getIdentifier();
if (outSequenceId == null || "".equals(outSequenceId))
throw new SandeshaException("OutSequenceId is null");
- RetransmitterBeanMgr retransmitterMgr = AbstractBeanMgrFactory
- .getInstance(context).getRetransmitterBeanMgr();
- SequencePropertyBeanMgr seqPropMgr = AbstractBeanMgrFactory
-
.getInstance(context).getSequencePropretyBeanMgr();
-
SequencePropertyBean tempSequenceBean = seqPropMgr.retrieve(
- outSequenceId,
- Constants.SequenceProperties.TEMP_SEQUENCE_ID);
-
- if (tempSequenceBean == null
- || tempSequenceBean.getValue() == null)
- throw new SandeshaException(
- "TempSequenceId is not set correctly");
+ outSequenceId,
Constants.SequenceProperties.TEMP_SEQUENCE_ID);
- String tempSequenceId = (String) tempSequenceBean.getValue();
-
+ if (tempSequenceBean == null || tempSequenceBean.getValue() ==
null)
+ throw new SandeshaException("TempSequenceId is not set
correctly");
+ String tempSequenceId = (String) tempSequenceBean.getValue();
+
+
+ //Following happens in the SandeshaGlobal handler
+
rmMsgCtx.getMessageContext().setProperty(Constants.ACK_PROCSSED,"true");
+
+ //Removing relatesTo - Some WSRM endpoints tend to set
relatesTo value for ack messages.
+ //Because of this dispatching may go wrong.
+ //So we set relatesTo value to null for ackMessages. (this
happens in the SandeshaGlobal handler)
+ rmMsgCtx.setRelatesTo(null);
+
RetransmitterBean input = new RetransmitterBean();
input.setTempSequenceId(tempSequenceId);
Collection retransmitterEntriesOfSequence = retransmitterMgr
@@ -125,35 +126,52 @@
//TODO - Process Nack
}
- //If all messages up to last message have been acknowledged.
- //Add terminate Sequence message.
- SequencePropertyBean lastOutMsgBean = seqPropMgr.retrieve(
- tempSequenceId,
- 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);
+
+
+
+// boolean justSendTerminateIfNeeded = false;
+// String ackProcessed = (String)
rmMsgCtx.getProperty(Constants.ACK_PROCSSED);
+// if (ackProcessed!=null && "true".equals(ackProcessed))
+// justSendTerminateIfNeeded = true;
- if (complete) {
- addTerminateSequenceMessage(rmMsgCtx,
outSequenceId,tempSequenceId);
- }
+ //following get called in the SandesaInHandler
+ //if (justSendTerminateIfNeeded) {
+ //If all messages up to last message have been
acknowledged.
+ //Add terminate Sequence message.
+ SequencePropertyBean lastOutMsgBean =
seqPropMgr.retrieve(
+ tempSequenceId,
+
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,
+ tempSequenceId);
+ }
+
+ //stopping the progress of the message further.
+ rmMsgCtx.getMessageContext().setPausedTrue(
+ new
QName(Constants.IN_HANDLER_NAME));
+
+ //return;
+ //}
}
-
- int i = 1;
- //stopping the progress of the message further.
- rmMsgCtx.getMessageContext().setPausedTrue(new QName
(Constants.IN_HANDLER_NAME));
+ int i = 1;
+
+
}
private RetransmitterBean getRetransmitterEntry(Collection collection,
@@ -169,32 +187,41 @@
}
public void addTerminateSequenceMessage(RMMsgContext incomingAckRMMsg,
- String outSequenceId,String tempSequenceId) throws
SandeshaException {
+ String outSequenceId, String tempSequenceId)
+ throws SandeshaException {
RMMsgContext terminateRMMessage = RMMsgCreator
.createTerminateSequenceMessage(incomingAckRMMsg, outSequenceId);
//detting addressing headers.
- SequencePropertyBeanMgr seqPropMgr =
AbstractBeanMgrFactory.getInstance(
-
incomingAckRMMsg.getContext()).getSequencePropretyBeanMgr();
- //SequencePropertyBean replyToBean =
seqPropMgr.retrieve(tempSequenceId,Constants.SequenceProperties.REPLY_TO_EPR);
- SequencePropertyBean toBean =
seqPropMgr.retrieve(tempSequenceId,Constants.SequenceProperties.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");
-
+ SequencePropertyBeanMgr seqPropMgr = AbstractBeanMgrFactory
+ .getInstance(incomingAckRMMsg.getContext())
+ .getSequencePropretyBeanMgr();
+ //SequencePropertyBean replyToBean =
+ //
seqPropMgr.retrieve(tempSequenceId,Constants.SequenceProperties.REPLY_TO_EPR);
+ SequencePropertyBean toBean =
seqPropMgr.retrieve(tempSequenceId,
+ Constants.SequenceProperties.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");
+
EndpointReference toEPR = (EndpointReference) toBean.getValue();
- if (toEPR==null)
- throw new SandeshaException ("To EPR has an invalid
value");
-
- terminateRMMessage.setTo(new EndpointReference
(toEPR.getAddress()));
-
- //terminateRMMessage.setFrom(new EndpointReference
(replyToEPR.getAddress()));
+ if (toEPR == null)
+ throw new SandeshaException("To EPR has an invalid
value");
+
+ terminateRMMessage.setTo(new
EndpointReference(toEPR.getAddress()));
+ terminateRMMessage.setFrom(new EndpointReference
(Constants.WSA.NS_URI_ANONYMOUS));
+ terminateRMMessage.setFaultTo(new EndpointReference
(Constants.WSA.NS_URI_ANONYMOUS));
+ //terminateRMMessage.setFrom(new EndpointReference
+ // (replyToEPR.getAddress()));
terminateRMMessage
.setWSAAction(Constants.WSRM.Actions.TERMINATE_SEQUENCE);
-
+ terminateRMMessage
+ .setSOAPAction("\"" +
Constants.WSRM.Actions.TERMINATE_SEQUENCE + "\"");
+
try {
terminateRMMessage.addSOAPEnvelope();
} catch (AxisFault e) {
@@ -205,20 +232,23 @@
.getMessageContext());
RetransmitterBean terminateBean = new RetransmitterBean();
terminateBean.setKey(key);
-
- //Set a retransmitter lastSentTime so that terminate will be
send with some delay.
+
+ //Set a retransmitter lastSentTime so that terminate will be
send with
+ // some delay.
//Otherwise this get send before return of the current request
(ack).
//TODO verify that the time given is correct
-
terminateBean.setLastSentTime(System.currentTimeMillis()+Constants.TERMINATE_DELAY);
-
+ terminateBean.setLastSentTime(System.currentTimeMillis()
+ + Constants.TERMINATE_DELAY);
+
terminateBean.setMessageId(terminateRMMessage.getMessageId());
terminateBean.setSend(true);
terminateBean.setReSend(false);
- RetransmitterBeanMgr retramsmitterMgr =
AbstractBeanMgrFactory.getInstance(
-
incomingAckRMMsg.getContext()).getRetransmitterBeanMgr();
+ RetransmitterBeanMgr retramsmitterMgr = AbstractBeanMgrFactory
+ .getInstance(incomingAckRMMsg.getContext())
+ .getRetransmitterBeanMgr();
retramsmitterMgr.insert(terminateBean);
-
+
}
}
Modified:
webservices/sandesha/trunk/src/org/apache/sandesha2/msgprocessors/ApplicationMsgProcessor.java
URL:
http://svn.apache.org/viewcvs/webservices/sandesha/trunk/src/org/apache/sandesha2/msgprocessors/ApplicationMsgProcessor.java?rev=329749&r1=329748&r2=329749&view=diff
==============================================================================
---
webservices/sandesha/trunk/src/org/apache/sandesha2/msgprocessors/ApplicationMsgProcessor.java
(original)
+++
webservices/sandesha/trunk/src/org/apache/sandesha2/msgprocessors/ApplicationMsgProcessor.java
Sun Oct 30 21:19:04 2005
@@ -31,6 +31,13 @@
import org.apache.axis2.context.ConfigurationContext;
import org.apache.axis2.context.MessageContext;
import org.apache.axis2.context.OperationContext;
+import org.apache.axis2.context.ServiceContext;
+import org.apache.axis2.context.ServiceGroupContext;
+import org.apache.axis2.description.AxisOperation;
+import org.apache.axis2.description.AxisOperationFactory;
+import org.apache.axis2.description.AxisService;
+import org.apache.axis2.description.AxisServiceGroup;
+import org.apache.axis2.engine.AxisConfiguration;
import org.apache.axis2.engine.AxisEngine;
import org.apache.axis2.soap.SOAPEnvelope;
import org.apache.sandesha2.Constants;
@@ -40,9 +47,11 @@
import org.apache.sandesha2.msgreceivers.RMMessageReceiver;
import org.apache.sandesha2.storage.AbstractBeanMgrFactory;
import org.apache.sandesha2.storage.beanmanagers.NextMsgBeanMgr;
+import org.apache.sandesha2.storage.beanmanagers.RetransmitterBeanMgr;
import org.apache.sandesha2.storage.beanmanagers.SequencePropertyBeanMgr;
import org.apache.sandesha2.storage.beanmanagers.StorageMapBeanMgr;
import org.apache.sandesha2.storage.beans.NextMsgBean;
+import org.apache.sandesha2.storage.beans.RetransmitterBean;
import org.apache.sandesha2.storage.beans.SequencePropertyBean;
import org.apache.sandesha2.storage.beans.StorageMapBean;
import org.apache.sandesha2.util.MsgInitializer;
@@ -51,6 +60,7 @@
import org.apache.sandesha2.util.SOAPAbstractFactory;
import org.apache.sandesha2.util.SandeshaUtil;
import org.apache.sandesha2.workers.InOrderInvoker;
+import org.apache.sandesha2.wsrm.AckRequested;
import org.apache.sandesha2.wsrm.AcknowledgementRange;
import org.apache.sandesha2.wsrm.Sequence;
import org.apache.sandesha2.wsrm.SequenceAcknowledgement;
@@ -130,64 +140,7 @@
msgsBean.setValue(messagesStr);
seqPropMgr.update(msgsBean);
- //Setting the ack depending on AcksTo.
- //TODO: Stop sending askc for every message.
- SequencePropertyBean acksToBean =
seqPropMgr.retrieve(sequenceId,
- Constants.SequenceProperties.ACKS_TO_EPR);
-
- EndpointReference acksTo = (EndpointReference)
acksToBean.getValue();
- String acksToStr = acksTo.getAddress();
-
- if (acksToStr == null || messagesStr == null)
- throw new SandeshaException(
- "Seqeunce properties are not set
correctly");
-
- //if (acksToStr.equals(Constants.WSA.NS_URI_ANONYMOUS)) {
-
- RMMsgContext ackRMMsgCtx = SandeshaUtil.deepCopy(rmMsgCtx);
- MessageContext ackMsgCtx = ackRMMsgCtx.getMessageContext();
-
ackMsgCtx.setServiceGroupContext(msgCtx.getServiceGroupContext());
-
ackMsgCtx.setServiceGroupContextId(msgCtx.getServiceGroupContextId());
- ackMsgCtx.setServiceContext(msgCtx.getServiceContext());
- ackMsgCtx.setServiceContextID(msgCtx.getServiceContextID());
-
- //TODO set a suitable operation description
- OperationContext ackOpContext = new OperationContext(msgCtx
- .getAxisOperation());
-
- try {
- ackOpContext.addMessageContext(ackMsgCtx);
- } catch (AxisFault e2) {
- throw new SandeshaException(e2.getMessage());
- }
- ackMsgCtx.setOperationContext(ackOpContext);
-
- //Set new envelope
- SOAPEnvelope envelope = SOAPAbstractFactory.getSOAPFactory(
-
Constants.SOAPVersion.DEFAULT).getDefaultEnvelope();
- try {
- ackMsgCtx.setEnvelope(envelope);
- } catch (AxisFault e3) {
- throw new SandeshaException(e3.getMessage());
- }
-
- //FIXME set acksTo instead of ReplyTo
- ackMsgCtx.setTo(acksTo);
- ackMsgCtx.setReplyTo(msgCtx.getTo());
- RMMsgCreator.addAckMessage(ackRMMsgCtx, sequenceId);
-
- AxisEngine engine = new
AxisEngine(ackRMMsgCtx.getMessageContext()
- .getSystemContext());
-
- //set CONTEXT_WRITTEN since acksto is anonymous
- rmMsgCtx.getMessageContext().getOperationContext().setProperty(
- org.apache.axis2.Constants.RESPONSE_WRITTEN,
"true");
- rmMsgCtx.getMessageContext().setProperty(Constants.ACK_WRITTEN,
"true");
- try {
- engine.send(ackRMMsgCtx.getMessageContext());
- } catch (AxisFault e1) {
- throw new SandeshaException(e1.getMessage());
- }
+ sendAckIfNeeded(rmMsgCtx,messagesStr);
// } else {
// //TODO Add async Ack
@@ -302,5 +255,118 @@
}
return false;
+ }
+
+ public void sendAckIfNeeded (RMMsgContext rmMsgCtx,String messagesStr)
throws SandeshaException {
+
+ MessageContext msgCtx = rmMsgCtx.getMessageContext();
+
+ SequencePropertyBeanMgr seqPropMgr = AbstractBeanMgrFactory
+ .getInstance(rmMsgCtx.getContext())
+ .getSequencePropretyBeanMgr();
+
+
+ Sequence sequence = (Sequence) rmMsgCtx
+ .getMessagePart(Constants.MessageParts.SEQUENCE);
+ String sequenceId = sequence.getIdentifier().getIdentifier();
+ ConfigurationContext configCtx = rmMsgCtx.getMessageContext()
+ .getSystemContext();
+ if (configCtx == null)
+ throw new SandeshaException("Configuration Context is
null");
+
+ //Setting the ack depending on AcksTo.
+ //TODO: Stop sending askc for every message.
+ SequencePropertyBean acksToBean =
seqPropMgr.retrieve(sequenceId,
+ Constants.SequenceProperties.ACKS_TO_EPR);
+
+ EndpointReference acksTo = (EndpointReference)
acksToBean.getValue();
+ String acksToStr = acksTo.getAddress();
+
+ if (acksToStr == null || messagesStr == null)
+ throw new SandeshaException(
+ "Seqeunce properties are not set
correctly");
+
+ //if (acksToStr.equals(Constants.WSA.NS_URI_ANONYMOUS)) {
+
+ AxisConfiguration axisConfig = configCtx.getAxisConfiguration();
+ AxisServiceGroup serviceGroup = new AxisServiceGroup
(axisConfig);
+ AxisService service = new AxisService (new QName
("RMClientService")); // This is a dummy service.
+ ServiceGroupContext serviceGroupContext = new
ServiceGroupContext (configCtx,serviceGroup);
+ ServiceContext serviceContext = new ServiceContext
(service,serviceGroupContext);
+
+
+ RMMsgContext ackRMMsgCtx = SandeshaUtil.deepCopy(rmMsgCtx);
+ MessageContext ackMsgCtx = ackRMMsgCtx.getMessageContext();
+
+ ackMsgCtx.setAxisServiceGroup(serviceGroup);
+ ackMsgCtx.setServiceGroupContext(serviceGroupContext);
+ ackMsgCtx.setAxisService(service);
+ ackMsgCtx.setServiceContext(serviceContext);
+
+ try {
+ AxisOperation ackOperation =
AxisOperationFactory.getOperetionDescription(AxisOperationFactory.MEP_CONSTANT_IN_ONLY);
+
+ AxisOperation rmMsgOperation =
rmMsgCtx.getMessageContext().getAxisOperation();
+ if (rmMsgOperation!=null) {
+ ArrayList outFlow =
rmMsgOperation.getPhasesOutFlow();
+ if (outFlow!=null) {
+ ackOperation.setPhasesOutFlow(outFlow);
+
ackOperation.setPhasesOutFaultFlow(outFlow);
+ }
+ }
+
+ OperationContext ackOpContext = new OperationContext
(ackOperation);
+ ackMsgCtx.setAxisOperation(ackOperation);
+ ackMsgCtx.setOperationContext(ackOpContext);
+ ackOpContext.addMessageContext(ackMsgCtx);
+ ackMsgCtx.setOperationContext(ackOpContext);
+
+ } catch (AxisFault e) {
+ throw new SandeshaException (e.getMessage());
+ }
+
+ //Set new envelope
+ SOAPEnvelope envelope = SOAPAbstractFactory.getSOAPFactory(
+
Constants.SOAPVersion.DEFAULT).getDefaultEnvelope();
+ try {
+ ackMsgCtx.setEnvelope(envelope);
+ } catch (AxisFault e3) {
+ throw new SandeshaException(e3.getMessage());
+ }
+
+ //FIXME set acksTo instead of ReplyTo
+ ackMsgCtx.setTo(acksTo);
+ ackMsgCtx.setReplyTo(msgCtx.getTo());
+ RMMsgCreator.addAckMessage(ackRMMsgCtx, sequenceId);
+
+ if (Constants.WSA.NS_URI_ANONYMOUS.equals(acksTo.getAddress()))
{
+ AxisEngine engine = new
AxisEngine(ackRMMsgCtx.getMessageContext()
+ .getSystemContext());
+
+
+ //set CONTEXT_WRITTEN since acksto is anonymous
+
rmMsgCtx.getMessageContext().getOperationContext().setProperty(
+
org.apache.axis2.Constants.RESPONSE_WRITTEN, "true");
+
rmMsgCtx.getMessageContext().setProperty(Constants.ACK_WRITTEN, "true");
+ try {
+ engine.send(ackRMMsgCtx.getMessageContext());
+ } catch (AxisFault e1) {
+ throw new SandeshaException(e1.getMessage());
+ }
+ } else {
+ RetransmitterBeanMgr retransmitterBeanMgr =
AbstractBeanMgrFactory.getInstance(configCtx).getRetransmitterBeanMgr();
+
+ String key =
SandeshaUtil.storeMessageContext(ackMsgCtx);
+ RetransmitterBean ackBean = new RetransmitterBean ();
+ ackBean.setKey(key);
+ ackBean.setMessageId(ackMsgCtx.getMessageID());
+ ackBean.setReSend(false);
+ ackBean.setSend(true);
+
+ retransmitterBeanMgr.insert(ackBean);
+
+ SandeshaUtil.startSenderIfStopped(configCtx);
+ }
+
}
}
Modified:
webservices/sandesha/trunk/src/org/apache/sandesha2/msgprocessors/TerminateSeqMsgProcessor.java
URL:
http://svn.apache.org/viewcvs/webservices/sandesha/trunk/src/org/apache/sandesha2/msgprocessors/TerminateSeqMsgProcessor.java?rev=329749&r1=329748&r2=329749&view=diff
==============================================================================
---
webservices/sandesha/trunk/src/org/apache/sandesha2/msgprocessors/TerminateSeqMsgProcessor.java
(original)
+++
webservices/sandesha/trunk/src/org/apache/sandesha2/msgprocessors/TerminateSeqMsgProcessor.java
Sun Oct 30 21:19:04 2005
@@ -17,13 +17,22 @@
package org.apache.sandesha2.msgprocessors;
+import java.io.SequenceInputStream;
+
import javax.xml.namespace.QName;
+import org.apache.axis2.context.ConfigurationContext;
import org.apache.axis2.context.MessageContext;
import org.apache.sandesha2.Constants;
import org.apache.sandesha2.RMMsgContext;
import org.apache.sandesha2.SandeshaException;
+import org.apache.sandesha2.storage.AbstractBeanMgrFactory;
+import org.apache.sandesha2.storage.beanmanagers.CreateSeqBeanMgr;
+import org.apache.sandesha2.storage.beanmanagers.NextMsgBeanMgr;
+import org.apache.sandesha2.storage.beanmanagers.SequencePropertyBeanMgr;
+import org.apache.sandesha2.storage.beanmanagers.StorageMapBeanMgr;
import org.apache.sandesha2.wsrm.SequenceAcknowledgement;
+import org.apache.sandesha2.wsrm.TerminateSequence;
/**
* @author Chamikara
@@ -45,7 +54,6 @@
//Processing the terminate message
//TODO Add terminate sequence message logic.
-
terminateSeqMsg.setPausedTrue(new QName
(Constants.IN_HANDLER_NAME));
Modified:
webservices/sandesha/trunk/src/org/apache/sandesha2/storage/beanmanagers/CreateSeqBeanMgr.java
URL:
http://svn.apache.org/viewcvs/webservices/sandesha/trunk/src/org/apache/sandesha2/storage/beanmanagers/CreateSeqBeanMgr.java?rev=329749&r1=329748&r2=329749&view=diff
==============================================================================
---
webservices/sandesha/trunk/src/org/apache/sandesha2/storage/beanmanagers/CreateSeqBeanMgr.java
(original)
+++
webservices/sandesha/trunk/src/org/apache/sandesha2/storage/beanmanagers/CreateSeqBeanMgr.java
Sun Oct 30 21:19:04 2005
@@ -75,18 +75,37 @@
public Collection find(CreateSeqBean bean) {
ArrayList beans = new ArrayList();
+ if (bean==null)
+ return beans;
+
Iterator iterator = table.values().iterator();
CreateSeqBean temp;
while (iterator.hasNext()) {
temp = (CreateSeqBean) iterator.next();
- if ((bean.getCreateSeqMsgId() != null &&
bean.getCreateSeqMsgId()
- .equals(temp.getCreateSeqMsgId()))
- && (bean.getSequenceId() != null &&
bean.getSequenceId()
-
.equals(bean.getSequenceId()))) {
+// if ((bean.getCreateSeqMsgId() != null &&
bean.getCreateSeqMsgId()
+// .equals(temp.getCreateSeqMsgId()))
+// && (bean.getSequenceId() != null &&
bean.getSequenceId()
+//
.equals(bean.getSequenceId()))) {
+// beans.add(temp);
+//
+// }
+
+ boolean equal = true;
+
+ if (bean.getCreateSeqMsgId()!=null &&
!bean.getCreateSeqMsgId().equals(temp.getCreateSeqMsgId()))
+ equal = false;
+
+ if (bean.getSequenceId()!=null &&
!bean.getSequenceId().equals(temp.getSequenceId()))
+ equal = false;
+
+ if (bean.getTempSequenceId()!=null &&
!bean.getTempSequenceId().equals(temp.getTempSequenceId()))
+ equal = false;
+
+
+ if (equal)
beans.add(temp);
-
- }
+
}
return beans;
}
Modified:
webservices/sandesha/trunk/src/org/apache/sandesha2/storage/beanmanagers/NextMsgBeanMgr.java
URL:
http://svn.apache.org/viewcvs/webservices/sandesha/trunk/src/org/apache/sandesha2/storage/beanmanagers/NextMsgBeanMgr.java?rev=329749&r1=329748&r2=329749&view=diff
==============================================================================
---
webservices/sandesha/trunk/src/org/apache/sandesha2/storage/beanmanagers/NextMsgBeanMgr.java
(original)
+++
webservices/sandesha/trunk/src/org/apache/sandesha2/storage/beanmanagers/NextMsgBeanMgr.java
Sun Oct 30 21:19:04 2005
@@ -71,20 +71,36 @@
ArrayList beans = new ArrayList();
Iterator iterator = table.values().iterator();
+ if (bean==null)
+ return beans;
+
NextMsgBean temp;
while (iterator.hasNext()) {
temp = (NextMsgBean) iterator.next();
- if ((bean.getSequenceId() != null &&
bean.getSequenceId().equals(
- temp.getSequenceId()))
- /*
- * && (bean.getNextMsgNoToProcess() !=
null &&
- *
bean.getNextMsgNoToProcess().equals(temp.getNextMsgNoToProcess()))
- */
- && (bean.getNextMsgNoToProcess() > 0)) {
-
+// if ((bean.getSequenceId() != null &&
bean.getSequenceId().equals(
+// temp.getSequenceId()))
+// /*
+// * && (bean.getNextMsgNoToProcess() !=
null &&
+// *
bean.getNextMsgNoToProcess().equals(temp.getNextMsgNoToProcess()))
+// */
+// && (bean.getNextMsgNoToProcess() > 0)) {
+//
+// beans.add(temp);
+// }
+
+
+ boolean equal = true;
+
+ if (bean.getNextMsgNoToProcess()>0 &&
bean.getNextMsgNoToProcess()!=temp.getNextMsgNoToProcess())
+ equal = false;
+
+ if (bean.getSequenceId()!=null &&
!bean.getSequenceId().equals(temp.getSequenceId()))
+ equal = false;
+
+ if (equal)
beans.add(temp);
- }
+
}
return beans;
Modified:
webservices/sandesha/trunk/src/org/apache/sandesha2/storage/beanmanagers/RetransmitterBeanMgr.java
URL:
http://svn.apache.org/viewcvs/webservices/sandesha/trunk/src/org/apache/sandesha2/storage/beanmanagers/RetransmitterBeanMgr.java?rev=329749&r1=329748&r2=329749&view=diff
==============================================================================
---
webservices/sandesha/trunk/src/org/apache/sandesha2/storage/beanmanagers/RetransmitterBeanMgr.java
(original)
+++
webservices/sandesha/trunk/src/org/apache/sandesha2/storage/beanmanagers/RetransmitterBeanMgr.java
Sun Oct 30 21:19:04 2005
@@ -113,8 +113,7 @@
int count = temp.getSentCount();
long timeNow = System.currentTimeMillis();
if (count == 0
- || (timeNow > lastSentTime
- +
Constants.WSP.RETRANSMISSION_INTERVAL)) {
+ || (timeNow > (lastSentTime +
Constants.WSP.RETRANSMISSION_INTERVAL))) {
beans.add(temp);
}
}
Modified:
webservices/sandesha/trunk/src/org/apache/sandesha2/storage/beanmanagers/SequencePropertyBeanMgr.java
URL:
http://svn.apache.org/viewcvs/webservices/sandesha/trunk/src/org/apache/sandesha2/storage/beanmanagers/SequencePropertyBeanMgr.java?rev=329749&r1=329748&r2=329749&view=diff
==============================================================================
---
webservices/sandesha/trunk/src/org/apache/sandesha2/storage/beanmanagers/SequencePropertyBeanMgr.java
(original)
+++
webservices/sandesha/trunk/src/org/apache/sandesha2/storage/beanmanagers/SequencePropertyBeanMgr.java
Sun Oct 30 21:19:04 2005
@@ -67,21 +67,40 @@
public Collection find(SequencePropertyBean bean) {
ArrayList beans = new ArrayList();
+
+ if (bean==null)
+ return beans;
+
Iterator iterator = table.values().iterator();
SequencePropertyBean temp;
while (iterator.hasNext()) {
temp = (SequencePropertyBean) iterator.next();
- if ((bean.getSequenceId() != null &&
bean.getSequenceId().equals(
- temp.getSequenceId()))
- && (bean.getName() != null &&
bean.getName().equals(
- temp.getName()))
- && (bean.getValue() != null &&
bean.getValue().equals(
- temp.getValue()))) {
-
+// if ((bean.getSequenceId() != null &&
bean.getSequenceId().equals(
+// temp.getSequenceId()))
+// && (bean.getName() != null &&
bean.getName().equals(
+// temp.getName()))
+// && (bean.getValue() != null &&
bean.getValue().equals(
+// temp.getValue()))) {
+//
+// beans.add(temp);
+// }
+
+ boolean equal = true;
+
+ if (bean.getSequenceId()!=null &&
!bean.getSequenceId().equals(temp.getSequenceId()))
+ equal = false;
+
+ if (bean.getName()!=null &&
!bean.getName().equals(temp.getName()))
+ equal = false;
+
+ if (bean.getValue()!=null &&
!bean.getValue().equals(temp.getValue()))
+ equal = false;
+
+ if (equal)
beans.add(temp);
- }
+
}
return beans;
}
Modified:
webservices/sandesha/trunk/src/org/apache/sandesha2/storage/beans/StorageMapBean.java
URL:
http://svn.apache.org/viewcvs/webservices/sandesha/trunk/src/org/apache/sandesha2/storage/beans/StorageMapBean.java?rev=329749&r1=329748&r2=329749&view=diff
==============================================================================
---
webservices/sandesha/trunk/src/org/apache/sandesha2/storage/beans/StorageMapBean.java
(original)
+++
webservices/sandesha/trunk/src/org/apache/sandesha2/storage/beans/StorageMapBean.java
Sun Oct 30 21:19:04 2005
@@ -29,6 +29,8 @@
private long MsgNo;
private String sequenceId;
+
+ private boolean invoked = false;
public StorageMapBean() {
@@ -83,5 +85,13 @@
*/
public void setSequenceId(String sequenceId) {
this.sequenceId = sequenceId;
+ }
+
+ public boolean isInvoked() {
+ return invoked;
+ }
+
+ public void setInvoked(boolean invoked) {
+ this.invoked = invoked;
}
}
Modified:
webservices/sandesha/trunk/src/org/apache/sandesha2/util/RMMsgCreator.java
URL:
http://svn.apache.org/viewcvs/webservices/sandesha/trunk/src/org/apache/sandesha2/util/RMMsgCreator.java?rev=329749&r1=329748&r2=329749&view=diff
==============================================================================
--- webservices/sandesha/trunk/src/org/apache/sandesha2/util/RMMsgCreator.java
(original)
+++ webservices/sandesha/trunk/src/org/apache/sandesha2/util/RMMsgCreator.java
Sun Oct 30 21:19:04 2005
@@ -17,6 +17,9 @@
package org.apache.sandesha2.util;
+import java.util.ArrayList;
+
+import javax.xml.namespace.QName;
import javax.xml.stream.XMLOutputFactory;
import javax.xml.stream.XMLStreamWriter;
@@ -28,8 +31,15 @@
import org.apache.axis2.context.ConfigurationContext;
import org.apache.axis2.context.MessageContext;
import org.apache.axis2.context.OperationContext;
+import org.apache.axis2.context.ServiceContext;
+import org.apache.axis2.context.ServiceGroupContext;
import org.apache.axis2.description.AxisOperation;
import org.apache.axis2.description.AxisOperationFactory;
+import org.apache.axis2.description.AxisService;
+import org.apache.axis2.description.AxisServiceGroup;
+import org.apache.axis2.engine.AxisConfiguration;
+import org.apache.axis2.engine.AxisConfigurationImpl;
+import org.apache.axis2.engine.Phase;
import org.apache.axis2.om.OMAbstractFactory;
import org.apache.axis2.om.impl.MIMEOutputUtils;
import org.apache.axis2.soap.SOAPEnvelope;
@@ -109,10 +119,12 @@
try {
AxisOperation appMsgOperationDesc =
applicationMsgContext.getAxisOperation();
AxisOperation createSeqOperationDesc =
AxisOperationFactory.getOperetionDescription(AxisOperationFactory.MEP_CONSTANT_OUT_IN);
-
createSeqOperationDesc.setPhasesOutFlow(appMsgOperationDesc.getPhasesOutFlow());
-
createSeqOperationDesc.setPhasesOutFaultFlow(appMsgOperationDesc.getPhasesOutFaultFlow());
-
createSeqOperationDesc.setPhasesInFaultFlow(appMsgOperationDesc.getPhasesInFaultFlow());
-
createSeqOperationDesc.setRemainingPhasesInFlow(appMsgOperationDesc.getRemainingPhasesInFlow());
+ if (appMsgOperationDesc!=null) {
+
createSeqOperationDesc.setPhasesOutFlow(appMsgOperationDesc.getPhasesOutFlow());
+
createSeqOperationDesc.setPhasesOutFaultFlow(appMsgOperationDesc.getPhasesOutFaultFlow());
+
createSeqOperationDesc.setPhasesInFaultFlow(appMsgOperationDesc.getPhasesInFaultFlow());
+
createSeqOperationDesc.setRemainingPhasesInFlow(appMsgOperationDesc.getRemainingPhasesInFlow());
+ }
createSeqmsgContext.setAxisOperation(createSeqOperationDesc);
//TODO set a suitable ope. description
@@ -201,7 +213,7 @@
}
createSeqRMMsg.setAction(Constants.WSRM.ACTION_CREATE_SEQ);
-
+ createSeqRMMsg.setSOAPAction("\"" +
Constants.WSRM.ACTION_CREATE_SEQ + "\"");
createSeqRMMsg.setMessageId(createSeqMsgId);
MessageContext createSeqMsg =
createSeqRMMsg.getMessageContext();
@@ -230,30 +242,43 @@
terminateMessage.setMessageInformationHeaders(newMessageInfoHeaders);
terminateMessage.setMessageID(SandeshaUtil.getUUID());
- terminateMessage.setServiceGroupContext(referenceMessage
- .getServiceGroupContext());
- terminateMessage.setServiceGroupContextId(referenceMessage
- .getServiceGroupContextId());
- terminateMessage.setServiceContext(referenceMessage
- .getServiceContext());
- terminateMessage.setServiceContextID(referenceMessage
- .getServiceContextID());
-
terminateMessage.setAxisOperation(referenceMessage.getAxisOperation());
- OperationContext newOperationCtx = new OperationContext
(terminateMessage.getAxisOperation());
+ ConfigurationContext configCtx =
referenceMessage.getSystemContext();
+ if (configCtx==null)
+ throw new SandeshaException ("Configuration Context is
null");
+
+ AxisConfiguration axisConfig = configCtx.getAxisConfiguration();
+ AxisServiceGroup serviceGroup = new AxisServiceGroup
(axisConfig);
+ AxisService service = new AxisService (new QName
("RMClientService")); // This is a dummy service.
+ ServiceGroupContext serviceGroupContext = new
ServiceGroupContext (configCtx,serviceGroup);
+ ServiceContext serviceContext = new ServiceContext
(service,serviceGroupContext);
+
+ terminateMessage.setAxisServiceGroup(serviceGroup);
+ terminateMessage.setServiceGroupContext(serviceGroupContext);
+ terminateMessage.setAxisService(service);
+ terminateMessage.setServiceContext(serviceContext);
+
try {
- newOperationCtx.addMessageContext(terminateMessage);
+ AxisOperation terminateOperaiton =
AxisOperationFactory.getOperetionDescription(AxisOperationFactory.MEP_CONSTANT_IN_ONLY);
+ AxisOperation referenceMsgOperation =
referenceMessage.getAxisOperation();
+ if (referenceMsgOperation!=null) {
+ ArrayList outPhases =
referenceMsgOperation.getPhasesOutFlow();
+ if (outPhases!=null) {
+
terminateOperaiton.setPhasesOutFlow(outPhases);
+
terminateOperaiton.setPhasesOutFaultFlow(outPhases);
+ }
+ }
+
+ OperationContext terminateOpContext = new
OperationContext (terminateOperaiton);
+ terminateMessage.setAxisOperation(terminateOperaiton);
+
terminateMessage.setOperationContext(terminateOpContext);
+ terminateOpContext.addMessageContext(terminateMessage);
+
terminateMessage.setOperationContext(terminateOpContext);
+
} 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);
@@ -306,7 +331,8 @@
Constants.SOAPVersion.DEFAULT).getDefaultEnvelope();
response.toOMElement(envelope.getBody());
outMessage.setWSAAction(Constants.WSRM.NS_URI_CREATE_SEQ_RESPONSE);
-
+
outMessage.setSoapAction(Constants.WSRM.NS_URI_CREATE_SEQ_RESPONSE);
+
String newMessageId = SandeshaUtil.getUUID();
outMessage.setMessageID(newMessageId);
@@ -357,6 +383,7 @@
sequenceAck.toOMElement(envelope.getHeader());
applicationMsg.setAction(Constants.WSRM.ACTION_SEQ_ACK);
+ applicationMsg.setSOAPAction(Constants.WSRM.ACTION_SEQ_ACK);
applicationMsg.setMessageId(SandeshaUtil.getUUID());
}
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=329749&r1=329748&r2=329749&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 30 21:19:04 2005
@@ -148,7 +148,7 @@
return input;
}
- private static long[] getLongArr(String[] strings) {
+ public static long[] getLongArr(String[] strings) {
int length = strings.length;
long[] longs = new long[length];
for (int i = 0; i < length; i++) {
@@ -192,8 +192,8 @@
newMessageContext.setProcessingFault(msgCtx.isProcessingFault());
newMessageContext.setResponseWritten(msgCtx.isResponseWritten());
newMessageContext.setRestThroughPOST(msgCtx.isRestThroughPOST());
- newMessageContext.setAxisOperation(msgCtx
- .getAxisOperation());
+ if (msgCtx.getAxisOperation()!=null)
+
newMessageContext.setAxisOperation(msgCtx.getAxisOperation());
if (msgCtx.getEnvelope() != null)
newMessageContext.setEnvelope(msgCtx.getEnvelope());
@@ -411,5 +411,47 @@
default:
return "Error";
}
+ }
+
+ public static boolean isGloballyProcessableMessageType (int type) {
+ if (type==Constants.MessageTypes.ACK ||
type==Constants.MessageTypes.TERMINATE_SEQ) {
+ return true;
+ }
+
+ return false;
+ }
+
+ public static boolean isDuplicateDropRequiredMsgType (int
rmMessageType) {
+ if (rmMessageType==Constants.MessageTypes.APPLICATION)
+ return true;
+
+ if (rmMessageType==Constants.MessageTypes.CREATE_SEQ_RESPONSE)
+ return true;
+
+ return false;
+ }
+
+ //TODO: correct following to work for long.
+ public static ArrayList getSplittedMsgNoArraylist (String str) {
+ String[] splitted = str.split(",");
+ ArrayList results = new ArrayList ();
+
+ long count = splitted.length;
+ for (int i=0;i<count;i++) {
+ String s = splitted[i];
+ results.add(s);
+ }
+
+ return results;
+ }
+
+ public static String getServerSideIncomingSeqIdFromInternalSeqId
(String internalSequenceId) {
+ String incomingSequenceId = internalSequenceId;
+ return incomingSequenceId;
+ }
+
+ public static String getServerSideInternalSeqIdFromIncomingSeqId
(String incomingSequenceId) {
+ String internalSequenceId = incomingSequenceId;
+ return internalSequenceId;
}
}
Modified:
webservices/sandesha/trunk/src/org/apache/sandesha2/workers/InOrderInvoker.java
URL:
http://svn.apache.org/viewcvs/webservices/sandesha/trunk/src/org/apache/sandesha2/workers/InOrderInvoker.java?rev=329749&r1=329748&r2=329749&view=diff
==============================================================================
---
webservices/sandesha/trunk/src/org/apache/sandesha2/workers/InOrderInvoker.java
(original)
+++
webservices/sandesha/trunk/src/org/apache/sandesha2/workers/InOrderInvoker.java
Sun Oct 30 21:19:04 2005
@@ -82,6 +82,8 @@
Thread.sleep(1000);
} catch (InterruptedException ex) {
System.out.println("Invoker was
Inturrepted....");
+ ex.printStackTrace();
+ System.out.println("End printing Interrupt...");
}
@@ -127,17 +129,17 @@
MessageContext msgToInvoke =
SandeshaUtil.getStoredMessageContext(key);
- //removing the storage map
entry.
- storageMapMgr.delete(key);
-
RMMsgContext rmMsg =
MsgInitializer.initializeMessage(msgToInvoke);
Sequence seq = (Sequence)
rmMsg.getMessagePart(Constants.MessageParts.SEQUENCE);
long msgNo =
seq.getMessageNumber().getMessageNumber();
- System.out.println("Invoking
message number " + msgNo + " of the sequence " + sequenceId);
try {
//Invoking the message.
new AxisEngine
(msgToInvoke.getSystemContext()).receive(msgToInvoke);
+
+ //deleting the message
entry.
+
storageMapMgr.delete(key);
+
} catch (AxisFault e) {
throw new
SandeshaException (e.getMessage());
}
Modified:
webservices/sandesha/trunk/src/org/apache/sandesha2/workers/Sender.java
URL:
http://svn.apache.org/viewcvs/webservices/sandesha/trunk/src/org/apache/sandesha2/workers/Sender.java?rev=329749&r1=329748&r2=329749&view=diff
==============================================================================
--- webservices/sandesha/trunk/src/org/apache/sandesha2/workers/Sender.java
(original)
+++ webservices/sandesha/trunk/src/org/apache/sandesha2/workers/Sender.java Sun
Oct 30 21:19:04 2005
@@ -17,6 +17,7 @@
package org.apache.sandesha2.workers;
import java.lang.reflect.InvocationTargetException;
+import java.util.ArrayList;
import java.util.Collection;
import java.util.Iterator;
@@ -107,6 +108,8 @@
e1.printStackTrace();
} catch (SandeshaException e2) {
e2.printStackTrace();
+ } catch (Exception e3) {
+ e3.printStackTrace();
}
//changing the values of the sent bean.
@@ -127,6 +130,8 @@
} catch (InterruptedException e1) {
//e1.printStackTrace();
System.out.println("Sender was interupted...");
+ e1.printStackTrace();
+ System.out.println("End printing Interrupt...");
}
}
@@ -190,6 +195,13 @@
response.setServiceContext(msgCtx.getServiceContext());
response.setAxisService (msgCtx.getAxisService());
response.setAxisServiceGroup(msgCtx.getAxisServiceGroup());
+
+ //setting the in-flow.
+ //ArrayList inPhaseHandlers =
response.getAxisOperation().getRemainingPhasesInFlow();
+ /*if (inPhaseHandlers==null ||
inPhaseHandlers.isEmpty()) {
+ ArrayList phases =
msgCtx.getSystemContext().getAxisConfiguration().getInPhasesUptoAndIncludingPostDispatch();
+
response.getAxisOperation().setRemainingPhasesInFlow(phases);
+ }*/
//Changed following from TransportUtils to SandeshaUtil
since op. context is anavailable.
SOAPEnvelope resenvelope = null;
---------------------------------------------------------------------
To unsubscribe, e-mail: [EMAIL PROTECTED]
For additional commands, e-mail: [EMAIL PROTECTED]
