Author: gatfora
Date: Mon Jun 23 06:07:30 2008
New Revision: 670554
URL: http://svn.apache.org/viewvc?rev=670554&view=rev
Log:
Applying patch for SANDESHA2-165, Thanks David, Sara
Modified:
webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/msgprocessors/AcknowledgementProcessor.java
webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/storage/beanmanagers/RMSBeanMgr.java
webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/storage/beanmanagers/SenderBeanMgr.java
webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/storage/inmemory/InMemoryBeanMgr.java
webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/storage/inmemory/InMemoryInvokerBeanMgr.java
webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/storage/inmemory/InMemoryRMDBeanMgr.java
webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/storage/inmemory/InMemoryRMSBeanMgr.java
webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/storage/inmemory/InMemorySenderBeanMgr.java
webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/util/FaultManager.java
webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/util/SandeshaUtil.java
Modified:
webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/msgprocessors/AcknowledgementProcessor.java
URL:
http://svn.apache.org/viewvc/webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/msgprocessors/AcknowledgementProcessor.java?rev=670554&r1=670553&r2=670554&view=diff
==============================================================================
---
webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/msgprocessors/AcknowledgementProcessor.java
(original)
+++
webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/msgprocessors/AcknowledgementProcessor.java
Mon Jun 23 06:07:30 2008
@@ -156,13 +156,7 @@
for(long messageNo =
newRanges[rangeIndex].lowerValue; messageNo<=newRanges[rangeIndex].upperValue;
messageNo++){
numberOfNewMessagesAcked++;
- SenderBean matcher = new
SenderBean();
-
matcher.setSequenceID(outSequenceId);
-
-
matcher.setMessageNumber(messageNo);
-
- SenderBean retransmitterBean =
retransmitterMgr.findUnique(matcher);
- if (retransmitterBean != null) {
+ SenderBean retransmitterBean =
retransmitterMgr.retrieve(outSequenceId, messageNo);
if (retransmitterBean != null) {
// Check we haven't got
an Ack for a message that hasn't been sent yet !
if
(retransmitterBean.getSentCount() == 0) {
FaultManager.makeInvalidAcknowledgementFault(rmMsgCtx, sequenceAck, ackRange,
Modified:
webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/storage/beanmanagers/RMSBeanMgr.java
URL:
http://svn.apache.org/viewvc/webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/storage/beanmanagers/RMSBeanMgr.java?rev=670554&r1=670553&r2=670554&view=diff
==============================================================================
---
webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/storage/beanmanagers/RMSBeanMgr.java
(original)
+++
webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/storage/beanmanagers/RMSBeanMgr.java
Mon Jun 23 06:07:30 2008
@@ -44,4 +44,7 @@
public RMSBean findUnique (RMSBean bean) throws SandeshaException;
-}
\ No newline at end of file
+ public RMSBean retrieveBySequenceID(String seqId) throws
SandeshaStorageException;
+
+ public RMSBean retrieveByInternalSequenceID(String internalSeqId)
throws SandeshaStorageException;
+}
Modified:
webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/storage/beanmanagers/SenderBeanMgr.java
URL:
http://svn.apache.org/viewvc/webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/storage/beanmanagers/SenderBeanMgr.java?rev=670554&r1=670553&r2=670554&view=diff
==============================================================================
---
webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/storage/beanmanagers/SenderBeanMgr.java
(original)
+++
webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/storage/beanmanagers/SenderBeanMgr.java
Mon Jun 23 06:07:30 2008
@@ -53,4 +53,7 @@
* @return
*/
public SenderBean retrieveFromMessageRefKey (String
messageContextRefKey);
+
+ public SenderBean retrieve(String sequnceId, long messageNumber) throws
SandeshaStorageException;
+
}
Modified:
webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/storage/inmemory/InMemoryBeanMgr.java
URL:
http://svn.apache.org/viewvc/webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/storage/inmemory/InMemoryBeanMgr.java?rev=670554&r1=670553&r2=670554&view=diff
==============================================================================
---
webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/storage/inmemory/InMemoryBeanMgr.java
(original)
+++
webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/storage/inmemory/InMemoryBeanMgr.java
Mon Jun 23 06:07:30 2008
@@ -66,14 +66,14 @@
return wasInserted;
}
- protected boolean delete(Object key) throws SandeshaStorageException {
+ protected RMBean delete(Object key) throws SandeshaStorageException {
if(LoggingControl.isAnyTracingEnabled() &&
log.isDebugEnabled()) log.debug("Entry: InMemoryBeanMgr " + this.getClass() + "
delete " + key);
RMBean bean = (RMBean) table.remove(key);
if(bean != null) {
mgr.enlistBean(bean);
}
if(LoggingControl.isAnyTracingEnabled() &&
log.isDebugEnabled()) log.debug("Exit: InMemoryBeanMgr " + this.getClass() + "
delete " + bean);
- return bean != null;
+ return bean;
}
protected RMBean retrieve(Object key) throws SandeshaStorageException {
Modified:
webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/storage/inmemory/InMemoryInvokerBeanMgr.java
URL:
http://svn.apache.org/viewvc/webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/storage/inmemory/InMemoryInvokerBeanMgr.java?rev=670554&r1=670553&r2=670554&view=diff
==============================================================================
---
webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/storage/inmemory/InMemoryInvokerBeanMgr.java
(original)
+++
webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/storage/inmemory/InMemoryInvokerBeanMgr.java
Mon Jun 23 06:07:30 2008
@@ -64,7 +64,7 @@
public boolean delete(String key) throws SandeshaStorageException {
lock.lock();
- boolean result = super.delete(key);
+ boolean result = (super.delete(key)!=null);
lock.unlock();
return result;
}
Modified:
webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/storage/inmemory/InMemoryRMDBeanMgr.java
URL:
http://svn.apache.org/viewvc/webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/storage/inmemory/InMemoryRMDBeanMgr.java?rev=670554&r1=670553&r2=670554&view=diff
==============================================================================
---
webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/storage/inmemory/InMemoryRMDBeanMgr.java
(original)
+++
webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/storage/inmemory/InMemoryRMDBeanMgr.java
Mon Jun 23 06:07:30 2008
@@ -36,7 +36,8 @@
}
public boolean delete(String sequenceId) throws
SandeshaStorageException {
- return super.delete(sequenceId);
+ return (super.delete(sequenceId)!=null);
+
}
public RMDBean retrieve(String sequenceId) throws
SandeshaStorageException {
Modified:
webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/storage/inmemory/InMemoryRMSBeanMgr.java
URL:
http://svn.apache.org/viewvc/webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/storage/inmemory/InMemoryRMSBeanMgr.java?rev=670554&r1=670553&r2=670554&view=diff
==============================================================================
---
webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/storage/inmemory/InMemoryRMSBeanMgr.java
(original)
+++
webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/storage/inmemory/InMemoryRMSBeanMgr.java
Mon Jun 23 06:07:30 2008
@@ -20,6 +20,9 @@
package org.apache.sandesha2.storage.inmemory;
import java.util.List;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
import org.apache.axis2.context.AbstractContext;
import org.apache.sandesha2.Sandesha2Constants;
@@ -29,23 +32,43 @@
public class InMemoryRMSBeanMgr extends InMemoryBeanMgr implements RMSBeanMgr {
+ private Lock lock = new ReentrantLock();
+
+ private ConcurrentHashMap seqID2csm = new ConcurrentHashMap();
+ private ConcurrentHashMap intSeqID2csm = new ConcurrentHashMap();
+
public InMemoryRMSBeanMgr(InMemoryStorageManager mgr, AbstractContext
context) {
super(mgr, context, Sandesha2Constants.BeanMAPs.CREATE_SEQUECE);
}
public boolean insert(RMSBean bean) throws SandeshaStorageException {
- boolean res = false;
- synchronized (table) {
- RMSBean findBean = new RMSBean();
- findBean.setInternalSequenceID(bean.getInternalSequenceID());
- if (findUniqueNoLock(findBean) == null)
- res = super.insert(bean.getCreateSeqMsgID(), bean);
- }
+ boolean res = false;
+ lock.lock();
+ if(intSeqID2csm.get(bean.getInternalSequenceID())==null){
+ res = super.insert(bean.getCreateSeqMsgID(), bean);
+ if(res){
+ if(bean.getInternalSequenceID()!=null){
+
intSeqID2csm.put(bean.getInternalSequenceID(), bean.getCreateSeqMsgID());
+ }
+ if(bean.getSequenceID()!=null){
+ seqID2csm.put(bean.getSequenceID(),
bean.getCreateSeqMsgID());
+ }
+ }
+ }
+ lock.unlock();
+
+
return res;
}
public boolean delete(String msgId) throws SandeshaStorageException {
- return super.delete(msgId);
+ RMSBean removed = (RMSBean) super.delete(msgId);
+ if(removed!=null){
+ seqID2csm.remove(removed.getSequenceID());
+ intSeqID2csm.remove(removed.getInternalSequenceID());
+ }
+ return removed!=null;
+
}
public RMSBean retrieve(String msgId) throws SandeshaStorageException {
@@ -53,7 +76,15 @@
}
public boolean update(RMSBean bean) throws SandeshaStorageException {
- return super.update(bean.getCreateSeqMsgID(), bean);
+ boolean result = super.update(bean.getCreateSeqMsgID(), bean);
+ if(bean.getInternalSequenceID()!=null){
+ intSeqID2csm.put(bean.getInternalSequenceID(),
bean.getCreateSeqMsgID());
+ }
+ if(bean.getSequenceID()!=null){
+ seqID2csm.put(bean.getSequenceID(),
bean.getCreateSeqMsgID());
+ }
+ return result;
+
}
public List find(RMSBean bean) throws SandeshaStorageException {
@@ -63,5 +94,33 @@
public RMSBean findUnique (RMSBean bean) throws
SandeshaStorageException {
return (RMSBean) super.findUnique(bean);
}
+
+ public RMSBean retrieveBySequenceID(String seqId) throws
SandeshaStorageException {
+ String csid = (String) seqID2csm.get(seqId);
+ RMSBean bean = null;
+ if(csid!=null){
+ bean = retrieve(csid);
+ }
+ if(bean == null){
+ RMSBean finder = new RMSBean();
+ finder.setSequenceID(seqId);
+ bean = findUnique(finder);
+ }
+ return bean;
+ }
+
+ public RMSBean retrieveByInternalSequenceID(String internalSeqId)
throws SandeshaStorageException {
+ String csid = (String) intSeqID2csm.get(internalSeqId);
+ RMSBean bean = null;
+ if(csid!=null){
+ bean = retrieve(csid);
+ }
+ if(bean == null){
+ RMSBean finder = new RMSBean();
+ finder.setInternalSequenceID(internalSeqId);
+ bean = findUnique(finder);
+ }
+ return bean;
+ }
}
Modified:
webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/storage/inmemory/InMemorySenderBeanMgr.java
URL:
http://svn.apache.org/viewvc/webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/storage/inmemory/InMemorySenderBeanMgr.java?rev=670554&r1=670553&r2=670554&view=diff
==============================================================================
---
webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/storage/inmemory/InMemorySenderBeanMgr.java
(original)
+++
webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/storage/inmemory/InMemorySenderBeanMgr.java
Mon Jun 23 06:07:30 2008
@@ -21,6 +21,7 @@
import java.util.Iterator;
import java.util.List;
+import java.util.concurrent.ConcurrentHashMap;
import org.apache.axis2.context.AbstractContext;
import org.apache.commons.logging.Log;
@@ -38,23 +39,42 @@
private static final Log log =
LogFactory.getLog(InMemorySenderBeanMgr.class);
+ ConcurrentHashMap sequenceIdandMessNum2MessageId = new
ConcurrentHashMap();
+
public InMemorySenderBeanMgr(InMemoryStorageManager mgr,
AbstractContext context) {
super(mgr, context, Sandesha2Constants.BeanMAPs.RETRANSMITTER);
}
public boolean delete(String MessageId) throws SandeshaStorageException
{
- return super.delete(MessageId);
+ SenderBean bean =(SenderBean) super.delete(MessageId);
+ if(bean.getSequenceID()!=null && bean.getMessageNumber()>0){
+
sequenceIdandMessNum2MessageId.remove(bean.getSequenceID()+":"+bean.getMessageNumber());
+ }
+ return bean!=null;
+
}
public SenderBean retrieve(String MessageId) throws
SandeshaStorageException {
return (SenderBean) super.retrieve(MessageId);
}
+ public SenderBean retrieve(String sequnceId, long messageNumber) throws
SandeshaStorageException {
+ String MessageId = (String)
sequenceIdandMessNum2MessageId.get(sequnceId+":"+messageNumber);
+ if(MessageId == null){
+ return null;
+ }
+ return (SenderBean) super.retrieve(MessageId);
+ }
+
+
public boolean insert(SenderBean bean) throws SandeshaStorageException {
if (bean.getMessageID() == null)
throw new
SandeshaStorageException(SandeshaMessageHelper.getMessage(
SandeshaMessageKeys.nullMsgId));
boolean result = super.insert(bean.getMessageID(), bean);
+ if(bean.getSequenceID()!=null && bean.getMessageNumber()>0){
+
sequenceIdandMessNum2MessageId.put(bean.getSequenceID()+":"+bean.getMessageNumber(),
bean.getMessageID());
+ }
mgr.getInMemoryTransaction().setSentMessages(true);
return result;
}
@@ -79,7 +99,7 @@
matcher.setTimeToSend(System.currentTimeMillis());
matcher.setTransportAvailable(true);
- List matches = super.find(matcher);
+ List matches = super.findNoLock(matcher);
if(LoggingControl.isAnyTracingEnabled() &&
log.isDebugEnabled()) log.debug("Found " + matches.size() + " messages");
// Look for the message with the lowest send time, and send
that one.
@@ -103,6 +123,14 @@
result = bean;
}
}
+ // Because the beans weren't locked before, need to do a
retrieve to get a locked copy.
+ // And then check that it's still valid
+ if(result!=null){
+ result = retrieve(result.getMessageID());
+ if(!result.match(matcher)){
+ result = null;
+ }
+ }
if(LoggingControl.isAnyTracingEnabled() &&
log.isDebugEnabled()) log.debug("Exit:
InMemorySenderBeanManager::getNextMessageToSend " + result);
return result;
@@ -110,6 +138,9 @@
public boolean update(SenderBean bean) throws SandeshaStorageException {
boolean result = super.update(bean.getMessageID(), bean);
+ if(bean.getSequenceID()!=null && bean.getMessageNumber()>0){
+
sequenceIdandMessNum2MessageId.put(bean.getSequenceID()+":"+bean.getMessageNumber(),
bean.getMessageID());
+ }
mgr.getInMemoryTransaction().setSentMessages(true);
return result;
}
Modified:
webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/util/FaultManager.java
URL:
http://svn.apache.org/viewvc/webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/util/FaultManager.java?rev=670554&r1=670553&r2=670554&view=diff
==============================================================================
---
webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/util/FaultManager.java
(original)
+++
webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/util/FaultManager.java
Mon Jun 23 06:07:30 2008
@@ -718,11 +718,19 @@
SOAPFaultDetail detail = faultPart.getDetail();
if (detail != null && !isSOAP11SequenceUnknownFault)
{
- OMElement identifierOM =
detail.getFirstChildWithName(new QName(rmMsgCtx.getRMNamespaceValue(),
-
Sandesha2Constants.WSRM_COMMON.IDENTIFIER));
- if (identifierOM != null)
+ // At this point we may not know what RM NS is in use.
+ OMElement identifierOM =
detail.getFirstChildWithName(new QName(Sandesha2Constants.SPEC_2005_02.NS_URI,
+
Sandesha2Constants.WSRM_COMMON.IDENTIFIER));
+ if (identifierOM != null){
identifier = identifierOM.getText();
- }
+ }else{
+ identifierOM = detail.getFirstChildWithName(new
QName(Sandesha2Constants.SPEC_2007_02.NS_URI,
+
Sandesha2Constants.WSRM_COMMON.IDENTIFIER));
+ if (identifierOM != null){
+ identifier = identifierOM.getText();
+ }
+ }
+ }
if
(Sandesha2Constants.SOAPFaults.Subcodes.CREATE_SEQUENCE_REFUSED.equals(soapFaultSubcode))
{
processCreateSequenceRefusedFault(rmMsgCtx, fault);
Modified:
webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/util/SandeshaUtil.java
URL:
http://svn.apache.org/viewvc/webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/util/SandeshaUtil.java?rev=670554&r1=670553&r2=670554&view=diff
==============================================================================
---
webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/util/SandeshaUtil.java
(original)
+++
webservices/sandesha/trunk/java/modules/core/src/main/java/org/apache/sandesha2/util/SandeshaUtil.java
Mon Jun 23 06:07:30 2008
@@ -609,23 +609,13 @@
throws SandeshaException {
RMSBeanMgr rmsBeanMgr = storageManager.getRMSBeanMgr();
- RMSBean bean = new RMSBean();
- bean.setInternalSequenceID(internalSequenceID);
-
- bean = rmsBeanMgr.findUnique(bean);
-
+ RMSBean bean =
rmsBeanMgr.retrieveByInternalSequenceID(internalSequenceID);
return bean;
}
- public static final RMSBean getRMSBeanFromSequenceId(StorageManager
storageManager, String sequenceID)
-
- throws SandeshaException {
+ public static final RMSBean getRMSBeanFromSequenceId(StorageManager
storageManager, String sequenceID) throws SandeshaException {
RMSBeanMgr rmsBeanMgr = storageManager.getRMSBeanMgr();
- RMSBean bean = new RMSBean();
- bean.setSequenceID(sequenceID);
-
- bean = rmsBeanMgr.findUnique(bean);
-
+ RMSBean bean = rmsBeanMgr.retrieveBySequenceID(sequenceID);
return bean;
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [EMAIL PROTECTED]
For additional commands, e-mail: [EMAIL PROTECTED]
