Apache
/
activemq
mirror of https://github.com/apache/activemq.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

https://issues.apache.org/activemq/browse/AMQ-2973 - removing composite subscription 

git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@1021768 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Bosanac Dejan 2010-10-12 13:29:37 +00:00
parent 1e900b300a
commit 3a2d93958d
3 changed files with 9 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
activemq-core/src/main/java/org/apache/activemq/broker/region/PrefetchSubscription.java
View File

@ -547,14 +547,16 @@ public abstract class PrefetchSubscription extends AbstractSubscription {
// Synchronized to DispatchLock // Synchronized to DispatchLock
synchronized(dispatchLock) { synchronized(dispatchLock) {
ArrayList<MessageReference> references = new ArrayList<MessageReference>();
for (MessageReference r : dispatched) { for (MessageReference r : dispatched) {
if( r.getRegionDestination() == destination) { if( r.getRegionDestination() == destination) {
rc.add(r); references.add(r);
} }
} }
destination.getDestinationStatistics().getDispatched().subtract(dispatched.size()); rc.addAll(references);
destination.getDestinationStatistics().getInflight().subtract(dispatched.size()); destination.getDestinationStatistics().getDispatched().subtract(references.size());
dispatched.clear(); destination.getDestinationStatistics().getInflight().subtract(references.size());
dispatched.removeAll(references);
} }
} }
return rc; return rc;

4
activemq-spring/src/test/java/org/apache/activemq/spring/ListenerTest.java
View File

@ -51,7 +51,7 @@ public class ListenerTest {
Thread.sleep(3000); Thread.sleep(3000);
LOG.info("messages received= " + listener.messages.size()); LOG.info("messages received= " + listener.messages.size());
Assert.assertEquals(listener.messages.size(), msgNum); Assert.assertEquals(msgNum, listener.messages.size());
} }
@ -63,7 +63,7 @@ public class ListenerTest {
Thread.sleep(3000); Thread.sleep(3000);
LOG.info("messages received= " + listener.messages.size()); LOG.info("messages received= " + listener.messages.size());
Assert.assertEquals(listener.messages.size(), 6 * msgNum); Assert.assertEquals(6 * msgNum, listener.messages.size());
} }
public void sendMessages(String destName, int msgNum) throws Exception { public void sendMessages(String destName, int msgNum) throws Exception {

3
activemq-spring/src/test/resources/spring/spring.xml
View File

@ -58,9 +58,8 @@
<bean id="compositeContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer"> <bean id="compositeContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="connectionFactory" ref="connectionFactory"/> <property name="connectionFactory" ref="connectionFactory"/>
<property name="messageListener" ref="messageListener"/> <property name="messageListener" ref="messageListener"/>
<property name="destinationName" value="TEST.>"/> <property name="destinationName" value="TEST.>?consumer.prefetchSize=1"/>
<property name="transactionManager" ref="transactionManager"/> <property name="transactionManager" ref="transactionManager"/>
<property name="cacheLevelName" value="CACHE_CONSUMER"/>
</bean> </bean>
</beans> </beans>