Author: sergeyb
Date: Fri Apr 17 15:14:09 2009
New Revision: 766045
URL: http://svn.apache.org/viewvc?rev=766045&view=rev
Log:
Merged revisions 766013 via svnmerge from
https://svn.apache.org/repos/asf/cxf/trunk
........
r766013 | sergeyb | 2009-04-17 14:58:12 +0100 (Fri, 17 Apr 2009) | 1 line
CXF-2002 : support for maxSuspendedContinuations in JMS
........
Added:
cxf/branches/2.1.x-fixes/systests/src/test/java/org/apache/cxf/systest/jms/continuations/HelloWorldContinuationsThrottleTest.java
- copied unchanged from r766013,
cxf/trunk/systests/src/test/java/org/apache/cxf/systest/jms/continuations/HelloWorldContinuationsThrottleTest.java
cxf/branches/2.1.x-fixes/systests/src/test/java/org/apache/cxf/systest/jms/continuations/HelloWorldWithContinuationsJMS2.java
- copied unchanged from r766013,
cxf/trunk/systests/src/test/java/org/apache/cxf/systest/jms/continuations/HelloWorldWithContinuationsJMS2.java
cxf/branches/2.1.x-fixes/systests/src/test/java/org/apache/cxf/systest/jms/continuations/Server3.java
- copied unchanged from r766013,
cxf/trunk/systests/src/test/java/org/apache/cxf/systest/jms/continuations/Server3.java
cxf/branches/2.1.x-fixes/systests/src/test/java/org/apache/cxf/systest/jms/continuations/test2.wsdl
- copied unchanged from r766013,
cxf/trunk/systests/src/test/java/org/apache/cxf/systest/jms/continuations/test2.wsdl
Modified:
cxf/branches/2.1.x-fixes/ (props changed)
cxf/branches/2.1.x-fixes/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/JMSConfiguration.java
cxf/branches/2.1.x-fixes/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/JMSDestination.java
cxf/branches/2.1.x-fixes/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/JMSFactory.java
cxf/branches/2.1.x-fixes/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/JMSOldConfigHolder.java
cxf/branches/2.1.x-fixes/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/continuations/JMSContinuation.java
cxf/branches/2.1.x-fixes/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/continuations/JMSContinuationProvider.java
cxf/branches/2.1.x-fixes/rt/transports/jms/src/test/java/org/apache/cxf/transport/jms/JMSDestinationTest.java
cxf/branches/2.1.x-fixes/rt/transports/jms/src/test/java/org/apache/cxf/transport/jms/continuations/JMSContinuationProviderTest.java
cxf/branches/2.1.x-fixes/rt/transports/jms/src/test/java/org/apache/cxf/transport/jms/continuations/JMSContinuationTest.java
cxf/branches/2.1.x-fixes/rt/transports/jms/src/test/resources/jms_test_config.xml
cxf/branches/2.1.x-fixes/systests/src/test/java/org/apache/cxf/systest/jms/continuations/HelloWorldContinuationsClientServerTest.java
cxf/branches/2.1.x-fixes/systests/src/test/java/org/apache/cxf/systest/jms/continuations/jms_test_config.xml
Propchange: cxf/branches/2.1.x-fixes/
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Fri Apr 17 15:14:09 2009
@@ -1 +1 @@
-/cxf/trunk:743446,753380,753397,753421,754585,755365,757499,757859,757899,757935,757951,758195,758303,758308,758378,758690,758910,759890,759961,759963-759964,759966,760029,760073,760150,760171,760178,760198,760212,760456,760468,760582,760938,761094,761113,761120,761317,761759,761789,762393,762518,762567,763200,763272,763495,763854,763931,763942,763953,764033-764034,764581,764599-764606,764887
+/cxf/trunk:743446,753380,753397,753421,754585,755365,757499,757859,757899,757935,757951,758195,758303,758308,758378,758690,758910,759890,759961,759963-759964,759966,760029,760073,760150,760171,760178,760198,760212,760456,760468,760582,760938,761094,761113,761120,761317,761759,761789,762393,762518,762567,763200,763272,763495,763854,763931,763942,763953,764033-764034,764581,764599-764606,764887,766013
Propchange: cxf/branches/2.1.x-fixes/
------------------------------------------------------------------------------
Binary property 'svnmerge-integrated' - no diff available.
Modified:
cxf/branches/2.1.x-fixes/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/JMSConfiguration.java
URL:
http://svn.apache.org/viewvc/cxf/branches/2.1.x-fixes/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/JMSConfiguration.java?rev=766045&r1=766044&r2=766045&view=diff
==============================================================================
---
cxf/branches/2.1.x-fixes/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/JMSConfiguration.java
(original)
+++
cxf/branches/2.1.x-fixes/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/JMSConfiguration.java
Fri Apr 17 15:14:09 2009
@@ -64,6 +64,7 @@
private int concurrentConsumers = 1;
private int maxConcurrentConsumers = 1;
private int maxConcurrentTasks = 10;
+ private int maxSuspendedContinuations = DEFAULT_VALUE;
private volatile String messageSelector;
private boolean subscriptionDurable;
@@ -318,6 +319,14 @@
public void setMaxConcurrentConsumers(int maxConcurrentConsumers) {
this.maxConcurrentConsumers = maxConcurrentConsumers;
}
+
+ public int getMaxSuspendedContinuations() {
+ return maxSuspendedContinuations;
+ }
+
+ public void setMaxSuspendedContinuations(int maxSuspendedContinuations) {
+ this.maxSuspendedContinuations = maxSuspendedContinuations;
+ }
public TaskExecutor getTaskExecutor() {
return taskExecutor;
Modified:
cxf/branches/2.1.x-fixes/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/JMSDestination.java
URL:
http://svn.apache.org/viewvc/cxf/branches/2.1.x-fixes/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/JMSDestination.java?rev=766045&r1=766044&r2=766045&view=diff
==============================================================================
---
cxf/branches/2.1.x-fixes/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/JMSDestination.java
(original)
+++
cxf/branches/2.1.x-fixes/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/JMSDestination.java
Fri Apr 17 15:14:09 2009
@@ -176,12 +176,15 @@
inMessage.put(JMSConstants.JMS_SERVER_RESPONSE_HEADERS, new
JMSMessageHeadersType());
inMessage.put(JMSConstants.JMS_REQUEST_MESSAGE, message);
inMessage.setDestination(this);
-
- inMessage.put(ContinuationProvider.class.getName(),
- new JMSContinuationProvider(bus,
- inMessage,
- incomingObserver,
- continuations));
+ if (jmsConfig.getMaxSuspendedContinuations() != 0) {
+ inMessage.put(ContinuationProvider.class.getName(),
+ new JMSContinuationProvider(bus,
+ inMessage,
+ incomingObserver,
+ continuations,
+ jmsListener,
+ jmsConfig));
+ }
BusFactory.setThreadDefaultBus(bus);
Modified:
cxf/branches/2.1.x-fixes/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/JMSFactory.java
URL:
http://svn.apache.org/viewvc/cxf/branches/2.1.x-fixes/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/JMSFactory.java?rev=766045&r1=766044&r2=766045&view=diff
==============================================================================
---
cxf/branches/2.1.x-fixes/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/JMSFactory.java
(original)
+++
cxf/branches/2.1.x-fixes/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/JMSFactory.java
Fri Apr 17 15:14:09 2009
@@ -18,6 +18,8 @@
*/
package org.apache.cxf.transport.jms;
+import java.util.logging.Logger;
+
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
@@ -26,6 +28,7 @@
import javax.jms.Session;
import javax.naming.NamingException;
+import org.apache.cxf.common.logging.LogUtils;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import
org.springframework.jms.connection.UserCredentialsConnectionFactoryAdapter;
import org.springframework.jms.core.JmsTemplate;
@@ -40,6 +43,8 @@
*/
public final class JMSFactory {
+ private static final Logger LOG = LogUtils.getL7dLogger(JMSFactory.class);
+
private JMSFactory() {
}
@@ -141,6 +146,12 @@
} else if (jmsConfig.getCacheLevel() !=
JMSConfiguration.DEFAULT_VALUE) {
jmsListener.setCacheLevel(jmsConfig.getCacheLevel());
}
+ if (jmsListener.getCacheLevel() >=
DefaultMessageListenerContainer.CACHE_CONSUMER
+ && jmsConfig.getMaxSuspendedContinuations() > 0) {
+ LOG.info("maxSuspendedContinuations value will be ignored - "
+ + ", please set cacheLevel to the value less than "
+ + "
org.springframework.jms.listener.DefaultMessageListenerContainer.CACHE_CONSUMER");
+ }
String staticSelectorPrefix = jmsConfig.getConduitSelectorPrefix();
if (!userCID && messageSelectorPrefix != null &&
jmsConfig.isUseConduitIdSelector()) {
jmsListener.setMessageSelector("JMSCorrelationID LIKE '"
Modified:
cxf/branches/2.1.x-fixes/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/JMSOldConfigHolder.java
URL:
http://svn.apache.org/viewvc/cxf/branches/2.1.x-fixes/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/JMSOldConfigHolder.java?rev=766045&r1=766044&r2=766045&view=diff
==============================================================================
---
cxf/branches/2.1.x-fixes/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/JMSOldConfigHolder.java
(original)
+++
cxf/branches/2.1.x-fixes/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/JMSOldConfigHolder.java
Fri Apr 17 15:14:09 2009
@@ -87,7 +87,9 @@
}
jmsConfig.setDurableSubscriptionName(serverBehavior.getDurableSubscriberName());
jmsConfig.setExplicitQosEnabled(true);
- jmsConfig.setMessageSelector(serverBehavior.getMessageSelector());
+ if (jmsConfig.getMessageSelector() == null) {
+
jmsConfig.setMessageSelector(serverBehavior.getMessageSelector());
+ }
if (isConduit && runtimePolicy.isSetMessageType()) {
jmsConfig.setMessageType(runtimePolicy.getMessageType().value());
}
Modified:
cxf/branches/2.1.x-fixes/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/continuations/JMSContinuation.java
URL:
http://svn.apache.org/viewvc/cxf/branches/2.1.x-fixes/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/continuations/JMSContinuation.java?rev=766045&r1=766044&r2=766045&view=diff
==============================================================================
---
cxf/branches/2.1.x-fixes/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/continuations/JMSContinuation.java
(original)
+++
cxf/branches/2.1.x-fixes/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/continuations/JMSContinuation.java
Fri Apr 17 15:14:09 2009
@@ -22,20 +22,31 @@
import java.util.Collection;
import java.util.Timer;
import java.util.TimerTask;
+import java.util.logging.Logger;
import org.apache.cxf.Bus;
import org.apache.cxf.BusFactory;
+import org.apache.cxf.common.logging.LogUtils;
import org.apache.cxf.continuations.Continuation;
import org.apache.cxf.continuations.SuspendedInvocationException;
import org.apache.cxf.message.Message;
import org.apache.cxf.transport.MessageObserver;
+import org.apache.cxf.transport.jms.JMSConfiguration;
+import org.springframework.jms.listener.DefaultMessageListenerContainer;
public class JMSContinuation implements Continuation {
+ static final String BOGUS_MESSAGE_SELECTOR =
"org.apache.cxf.transports.jms.continuations=too-many";
+ private static final Logger LOG =
LogUtils.getL7dLogger(JMSContinuation.class);
+
private Bus bus;
private Message inMessage;
private MessageObserver incomingObserver;
private Collection<JMSContinuation> continuations;
+ private DefaultMessageListenerContainer jmsListener;
+ private JMSConfiguration jmsConfig;
+
+ private String currentMessageSelector = BOGUS_MESSAGE_SELECTOR;
private Object userObject;
@@ -44,14 +55,16 @@
private boolean isResumed;
private Timer timer = new Timer();
- public JMSContinuation(Bus b,
- Message m,
- MessageObserver observer,
- Collection<JMSContinuation> cList) {
+ public JMSContinuation(Bus b, Message m, MessageObserver observer,
+ Collection<JMSContinuation> cList,
+ DefaultMessageListenerContainer jmsListener,
+ JMSConfiguration jmsConfig) {
bus = b;
inMessage = m;
incomingObserver = observer;
continuations = cList;
+ this.jmsListener = jmsListener;
+ this.jmsConfig = jmsConfig;
}
public Object getObject() {
@@ -87,7 +100,8 @@
}
protected void doResume() {
- continuations.remove(this);
+
+ updateContinuations(true);
BusFactory.setThreadDefaultBus(bus);
try {
@@ -108,8 +122,8 @@
return false;
}
- continuations.add(this);
-
+ updateContinuations(false);
+
isNew = false;
isResumed = false;
isPending = true;
@@ -136,4 +150,47 @@
protected void cancelTimerTask() {
timer.cancel();
}
+
+ protected void updateContinuations(boolean remove) {
+
+ modifyList(remove);
+
+ if (jmsConfig.getMaxSuspendedContinuations() < 0
+ || jmsListener.getCacheLevel() >=
DefaultMessageListenerContainer.CACHE_CONSUMER) {
+ return;
+ }
+
+ // throttle the flow if there're too many continuation instances in
memory
+ if (remove && !BOGUS_MESSAGE_SELECTOR.equals(currentMessageSelector)) {
+ LOG.fine("A number of continuations has dropped below the limit of
"
+ + jmsConfig.getMaxSuspendedContinuations()
+ + ", resetting JMS MessageSelector to " +
currentMessageSelector);
+ jmsListener.setMessageSelector(currentMessageSelector);
+ currentMessageSelector = BOGUS_MESSAGE_SELECTOR;
+ } else if (!remove && continuations.size() >=
jmsConfig.getMaxSuspendedContinuations()) {
+ currentMessageSelector = jmsListener.getMessageSelector();
+ if (!BOGUS_MESSAGE_SELECTOR.equals(currentMessageSelector)) {
+ LOG.fine("A number of continuations has reached the limit of "
+ + jmsConfig.getMaxSuspendedContinuations()
+ + ", setting JMS MessageSelector to " +
BOGUS_MESSAGE_SELECTOR);
+ jmsListener.setMessageSelector(BOGUS_MESSAGE_SELECTOR);
+
+ }
+ }
+
+ }
+
+ protected void modifyList(boolean remove) {
+ if (remove) {
+ continuations.remove(this);
+ } else {
+ continuations.add(this);
+ }
+ }
+
+ String getCurrentMessageSelector() {
+ return currentMessageSelector;
+ }
+
+
}
Modified:
cxf/branches/2.1.x-fixes/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/continuations/JMSContinuationProvider.java
URL:
http://svn.apache.org/viewvc/cxf/branches/2.1.x-fixes/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/continuations/JMSContinuationProvider.java?rev=766045&r1=766044&r2=766045&view=diff
==============================================================================
---
cxf/branches/2.1.x-fixes/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/continuations/JMSContinuationProvider.java
(original)
+++
cxf/branches/2.1.x-fixes/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/continuations/JMSContinuationProvider.java
Fri Apr 17 15:14:09 2009
@@ -26,6 +26,8 @@
import org.apache.cxf.continuations.ContinuationProvider;
import org.apache.cxf.message.Message;
import org.apache.cxf.transport.MessageObserver;
+import org.apache.cxf.transport.jms.JMSConfiguration;
+import org.springframework.jms.listener.DefaultMessageListenerContainer;
public class JMSContinuationProvider implements ContinuationProvider {
@@ -33,15 +35,21 @@
private Message inMessage;
private MessageObserver incomingObserver;
private Collection<JMSContinuation> continuations;
+ private DefaultMessageListenerContainer jmsListener;
+ private JMSConfiguration jmsConfig;
public JMSContinuationProvider(Bus b,
Message m,
MessageObserver observer,
- Collection<JMSContinuation> cList) {
+ Collection<JMSContinuation> cList,
+ DefaultMessageListenerContainer jmsListener,
+ JMSConfiguration jmsConfig) {
bus = b;
inMessage = m;
incomingObserver = observer;
continuations = cList;
+ this.jmsListener = jmsListener;
+ this.jmsConfig = jmsConfig;
}
public Continuation getContinuation() {
@@ -50,10 +58,8 @@
}
JMSContinuation cw = inMessage.get(JMSContinuation.class);
if (cw == null) {
- cw = new JMSContinuation(bus,
- inMessage,
- incomingObserver,
- continuations);
+ cw = new JMSContinuation(bus, inMessage, incomingObserver,
continuations,
+ jmsListener, jmsConfig);
inMessage.put(JMSContinuation.class, cw);
}
return cw;
Modified:
cxf/branches/2.1.x-fixes/rt/transports/jms/src/test/java/org/apache/cxf/transport/jms/JMSDestinationTest.java
URL:
http://svn.apache.org/viewvc/cxf/branches/2.1.x-fixes/rt/transports/jms/src/test/java/org/apache/cxf/transport/jms/JMSDestinationTest.java?rev=766045&r1=766044&r2=766045&view=diff
==============================================================================
---
cxf/branches/2.1.x-fixes/rt/transports/jms/src/test/java/org/apache/cxf/transport/jms/JMSDestinationTest.java
(original)
+++
cxf/branches/2.1.x-fixes/rt/transports/jms/src/test/java/org/apache/cxf/transport/jms/JMSDestinationTest.java
Fri Apr 17 15:14:09 2009
@@ -127,6 +127,8 @@
jmsConfig = destination.getJmsConfig();*/
assertEquals("The concurrentConsumer should be set",
jmsConfig.getConcurrentConsumers(), 3);
assertEquals("The maxConcurrentConsumer should be set",
jmsConfig.getMaxConcurrentConsumers(), 5);
+ assertEquals("The maxSuspendedContinuations should be set",
+ jmsConfig.getMaxSuspendedContinuations(), 2);
assertNotNull("The connectionFactory should not be null",
jmsConfig.getConnectionFactory());
assertTrue("Should get the instance of ActiveMQConnectionFactory",
jmsConfig.getConnectionFactory() instanceof
ActiveMQConnectionFactory);
Modified:
cxf/branches/2.1.x-fixes/rt/transports/jms/src/test/java/org/apache/cxf/transport/jms/continuations/JMSContinuationProviderTest.java
URL:
http://svn.apache.org/viewvc/cxf/branches/2.1.x-fixes/rt/transports/jms/src/test/java/org/apache/cxf/transport/jms/continuations/JMSContinuationProviderTest.java?rev=766045&r1=766044&r2=766045&view=diff
==============================================================================
---
cxf/branches/2.1.x-fixes/rt/transports/jms/src/test/java/org/apache/cxf/transport/jms/continuations/JMSContinuationProviderTest.java
(original)
+++
cxf/branches/2.1.x-fixes/rt/transports/jms/src/test/java/org/apache/cxf/transport/jms/continuations/JMSContinuationProviderTest.java
Fri Apr 17 15:14:09 2009
@@ -36,7 +36,8 @@
exchange.setOneWay(true);
Message m = new MessageImpl();
m.setExchange(exchange);
- JMSContinuationProvider provider = new JMSContinuationProvider(null,
m, null, null);
+ JMSContinuationProvider provider =
+ new JMSContinuationProvider(null, m, null, null, null, null);
assertNull(provider.getContinuation());
}
@@ -44,7 +45,8 @@
public void testGetNewContinuation() {
Message m = new MessageImpl();
m.setExchange(new ExchangeImpl());
- JMSContinuationProvider provider = new JMSContinuationProvider(null,
m, null, null);
+ JMSContinuationProvider provider =
+ new JMSContinuationProvider(null, m, null, null, null, null);
Continuation cw = provider.getContinuation();
assertTrue(cw.isNew());
assertSame(cw, m.get(JMSContinuation.class));
@@ -54,9 +56,9 @@
public void testGetExistingContinuation() {
Message m = new MessageImpl();
m.setExchange(new ExchangeImpl());
- JMSContinuation cw = new JMSContinuation(null, m, null, null);
+ JMSContinuation cw = new JMSContinuation(null, m, null, null, null,
null);
m.put(JMSContinuation.class, cw);
- JMSContinuationProvider provider = new JMSContinuationProvider(null,
m, null, null);
+ JMSContinuationProvider provider = new JMSContinuationProvider(null,
m, null, null, null, null);
assertSame(cw, provider.getContinuation());
assertSame(cw, m.get(JMSContinuation.class));
}
Modified:
cxf/branches/2.1.x-fixes/rt/transports/jms/src/test/java/org/apache/cxf/transport/jms/continuations/JMSContinuationTest.java
URL:
http://svn.apache.org/viewvc/cxf/branches/2.1.x-fixes/rt/transports/jms/src/test/java/org/apache/cxf/transport/jms/continuations/JMSContinuationTest.java?rev=766045&r1=766044&r2=766045&view=diff
==============================================================================
---
cxf/branches/2.1.x-fixes/rt/transports/jms/src/test/java/org/apache/cxf/transport/jms/continuations/JMSContinuationTest.java
(original)
+++
cxf/branches/2.1.x-fixes/rt/transports/jms/src/test/java/org/apache/cxf/transport/jms/continuations/JMSContinuationTest.java
Fri Apr 17 15:14:09 2009
@@ -28,10 +28,13 @@
import org.apache.cxf.message.Message;
import org.apache.cxf.message.MessageImpl;
import org.apache.cxf.transport.MessageObserver;
+import org.apache.cxf.transport.jms.JMSConfiguration;
import org.easymock.classextension.EasyMock;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
+import org.springframework.jms.listener.DefaultMessageListenerContainer;
+
public class JMSContinuationTest extends Assert {
@@ -52,7 +55,7 @@
@Test
public void testInitialStatus() {
JMSContinuation cw =
- new JMSContinuation(b, m, observer, continuations);
+ new JMSContinuation(b, m, observer, continuations, null, null);
assertTrue(cw.isNew());
assertFalse(cw.isPending());
assertFalse(cw.isResumed());
@@ -61,7 +64,7 @@
@Test
public void testSuspendResume() {
TestJMSContinuationWrapper cw =
- new TestJMSContinuationWrapper(b, m, observer, continuations);
+ new TestJMSContinuationWrapper(b, m, observer, continuations,
null, new JMSConfiguration());
try {
cw.suspend(5000);
fail("SuspendInvocation exception expected");
@@ -96,9 +99,58 @@
}
@Test
+ public void testThrottleWithMessageSelector() {
+
+ DefaultMessageListenerContainer springContainer = new
DefaultMessageListenerContainer();
+ springContainer.setCacheLevel(2);
+ JMSConfiguration config = new JMSConfiguration();
+ config.setMaxSuspendedContinuations(1);
+
+ TestJMSContinuationWrapper cw =
+ new TestJMSContinuationWrapper(b, m, observer, continuations,
+ springContainer, config);
+
+ assertNull(springContainer.getMessageSelector());
+ assertEquals(JMSContinuation.BOGUS_MESSAGE_SELECTOR,
cw.getCurrentMessageSelector());
+
+ suspendResumeCheckSelector(cw, springContainer);
+ EasyMock.reset(observer);
+ suspendResumeCheckSelector(cw, springContainer);
+
+ }
+
+ private void suspendResumeCheckSelector(JMSContinuation cw,
+ DefaultMessageListenerContainer
springContainer) {
+ try {
+ cw.suspend(5000);
+ fail("SuspendInvocation exception expected");
+ } catch (SuspendedInvocationException ex) {
+ // ignore
+ }
+ assertEquals(continuations.size(), 1);
+ assertSame(continuations.get(0), cw);
+
+ assertFalse(cw.suspend(1000));
+
+ assertEquals(JMSContinuation.BOGUS_MESSAGE_SELECTOR,
springContainer.getMessageSelector());
+ assertNull(cw.getCurrentMessageSelector());
+
+ observer.onMessage(m);
+ EasyMock.expectLastCall();
+ EasyMock.replay(observer);
+
+ cw.resume();
+
+ assertEquals(continuations.size(), 0);
+ EasyMock.verify(observer);
+
+ assertNull(springContainer.getMessageSelector());
+ assertEquals(JMSContinuation.BOGUS_MESSAGE_SELECTOR,
cw.getCurrentMessageSelector());
+ }
+
+ @Test
public void testUserObject() {
- JMSContinuation cw =
- new JMSContinuation(b, m, observer, continuations);
+ JMSContinuation cw = new JMSContinuation(b, m, observer,
continuations, null, null);
assertNull(cw.getObject());
Object userObject = new Object();
cw.setObject(userObject);
@@ -113,8 +165,10 @@
public TestJMSContinuationWrapper(Bus b,
Message m,
MessageObserver observer,
- List<JMSContinuation> cList) {
- super(b, m, observer, cList);
+ List<JMSContinuation> cList,
+ DefaultMessageListenerContainer
jmsListener,
+ JMSConfiguration jmsConfig) {
+ super(b, m, observer, cList, jmsListener, jmsConfig);
}
public void createTimerTask(long timeout) {
Modified:
cxf/branches/2.1.x-fixes/rt/transports/jms/src/test/resources/jms_test_config.xml
URL:
http://svn.apache.org/viewvc/cxf/branches/2.1.x-fixes/rt/transports/jms/src/test/resources/jms_test_config.xml?rev=766045&r1=766044&r2=766045&view=diff
==============================================================================
---
cxf/branches/2.1.x-fixes/rt/transports/jms/src/test/resources/jms_test_config.xml
(original)
+++
cxf/branches/2.1.x-fixes/rt/transports/jms/src/test/resources/jms_test_config.xml
Fri Apr 17 15:14:09 2009
@@ -107,6 +107,7 @@
p:usingEndpointInfo="false"
p:concurrentConsumers="3"
p:maxConcurrentConsumers="5"
+ p:maxSuspendedContinuations="2"
/>
<!--
Modified:
cxf/branches/2.1.x-fixes/systests/src/test/java/org/apache/cxf/systest/jms/continuations/HelloWorldContinuationsClientServerTest.java
URL:
http://svn.apache.org/viewvc/cxf/branches/2.1.x-fixes/systests/src/test/java/org/apache/cxf/systest/jms/continuations/HelloWorldContinuationsClientServerTest.java?rev=766045&r1=766044&r2=766045&view=diff
==============================================================================
---
cxf/branches/2.1.x-fixes/systests/src/test/java/org/apache/cxf/systest/jms/continuations/HelloWorldContinuationsClientServerTest.java
(original)
+++
cxf/branches/2.1.x-fixes/systests/src/test/java/org/apache/cxf/systest/jms/continuations/HelloWorldContinuationsClientServerTest.java
Fri Apr 17 15:14:09 2009
@@ -61,7 +61,7 @@
launchServer(EmbeddedJMSBrokerLauncher.class, props, null));
assertTrue("server did not launch correctly",
- launchServer(Server2.class, false));
+ launchServer(Server2.class));
serversStarted = true;
}
Modified:
cxf/branches/2.1.x-fixes/systests/src/test/java/org/apache/cxf/systest/jms/continuations/jms_test_config.xml
URL:
http://svn.apache.org/viewvc/cxf/branches/2.1.x-fixes/systests/src/test/java/org/apache/cxf/systest/jms/continuations/jms_test_config.xml?rev=766045&r1=766044&r2=766045&view=diff
==============================================================================
---
cxf/branches/2.1.x-fixes/systests/src/test/java/org/apache/cxf/systest/jms/continuations/jms_test_config.xml
(original)
+++
cxf/branches/2.1.x-fixes/systests/src/test/java/org/apache/cxf/systest/jms/continuations/jms_test_config.xml
Fri Apr 17 15:14:09 2009
@@ -31,6 +31,11 @@
<jms:conduit
name="{http://cxf.apache.org/systest/jaxws}HelloContinuationPort.jms-conduit";>
<jms:clientConfig clientReceiveTimeout="500000"
messageTimeToLive="500000"/>
+ <jms:address jndiConnectionFactoryName="ConnectionFactory"
+ jndiDestinationName="dynamicQueues/test.jmstransport.text">
+ <jms:JMSNamingProperty name="java.naming.factory.initial"
value="org.apache.activemq.jndi.ActiveMQInitialContextFactory"/>
+ <jms:JMSNamingProperty name="java.naming.provider.url"
value="tcp://localhost:61500"/>
+ </jms:address>
<jms:jmsConfig-ref>jmsConf1</jms:jmsConfig-ref>
</jms:conduit>
@@ -44,7 +49,25 @@
<bean id="jmsConf1" class="org.apache.cxf.transport.jms.JMSConfiguration"
p:connectionFactory-ref="singleConnectionFactory"
- p:concurrentConsumers="10"
- p:maxConcurrentConsumers="10"/>
+ p:concurrentConsumers="1"
+ p:maxConcurrentConsumers="1"/>
+
+ <jms:destination
name="{http://cxf.apache.org/systest/jaxws}HelloContinuationPort.jms-destination";>
+ <jms:address jndiConnectionFactoryName="ConnectionFactory"
+ jndiDestinationName="dynamicQueues/test.jmstransport.text">
+ <jms:JMSNamingProperty name="java.naming.factory.initial"
value="org.apache.activemq.jndi.ActiveMQInitialContextFactory"/>
+ <jms:JMSNamingProperty name="java.naming.provider.url"
value="tcp://localhost:61500"/>
+ </jms:address>
+ <jms:jmsConfig-ref>jmsConf2</jms:jmsConfig-ref>
+ </jms:destination>
+
+ <bean id="jmsConf2" class="org.apache.cxf.transport.jms.JMSConfiguration"
+ p:connectionFactory-ref="jmsConnectionFactory"
+ p:timeToLive="500000"
+ p:concurrentConsumers="1"
+ p:maxConcurrentConsumers="1"
+ p:maxSuspendedContinuations="1"
+ p:cacheLevel="2"
+ />
</beans>
