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;
}
lmpx.com only provides a reader for public news (NNTP) servers. It is not
affiliated with the servers or forums shown here and is not responsible for
the content of articles, which is written by their respective authors.