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

set the prefetch to something sensible (this is actually a key bit!) 

git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@509566 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Robert Davies 2007-02-20 14:01:55 +00:00
parent d0beb32cb8
commit b0b602121e
1 changed files with 123 additions and 152 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

157
activemq-core/src/test/java/org/apache/activemq/bugs/JmsDurableTopicSlowReceiveTest.java
View File

@ -1,24 +1,21 @@
/** /**
* *
* Licensed to the Apache Software Foundation (ASF) under one or more * Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements. See the NOTICE
* contributor license agreements. See the NOTICE file distributed with * file distributed with this work for additional information regarding copyright ownership. The ASF licenses this file
* this work for additional information regarding copyright ownership. * to You under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the
* The ASF licenses this file to You under the Apache License, Version 2.0 * License. You may obtain a copy of the License at
* (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 * http://www.apache.org/licenses/LICENSE-2.0
* *
* Unless required by applicable law or agreed to in writing, software * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
* distributed under the License is distributed on an "AS IS" BASIS, * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * specific language governing permissions and limitations under the License.
* See the License for the specific language governing permissions and
* limitations under the License.
*/ */
package org.apache.activemq.bugs; package org.apache.activemq.bugs;
import java.io.File; import java.io.File;
import java.util.Properties;
import javax.jms.BytesMessage; import javax.jms.BytesMessage;
import javax.jms.Connection; import javax.jms.Connection;
import javax.jms.Destination; import javax.jms.Destination;
@ -37,32 +34,21 @@ import org.apache.activemq.test.JmsTopicSendReceiveTest;
/** /**
* @version $Revision: 1.5 $ * @version $Revision: 1.5 $
*/ */
public class JmsDurableTopicSlowReceiveTest extends JmsTopicSendReceiveTest { public class JmsDurableTopicSlowReceiveTest extends JmsTopicSendReceiveTest{
private static final org.apache.commons.logging.Log log = org.apache.commons.logging.LogFactory
private static final org.apache.commons.logging.Log log=org.apache.commons.logging.LogFactory
.getLog(JmsDurableTopicSlowReceiveTest.class); .getLog(JmsDurableTopicSlowReceiveTest.class);
protected Connection connection2; protected Connection connection2;
protected Session session2; protected Session session2;
protected Session consumeSession2; protected Session consumeSession2;
protected MessageConsumer consumer2; protected MessageConsumer consumer2;
protected MessageProducer producer2; protected MessageProducer producer2;
protected Destination consumerDestination2; protected Destination consumerDestination2;
BrokerService broker;
final int NMSG=100;
final int NMSG = 100; final int MSIZE=256000;
final int MSIZE = 256000;
private Connection connection3; private Connection connection3;
private Session consumeSession3; private Session consumeSession3;
private TopicSubscriber consumer3; private TopicSubscriber consumer3;
/** /**
@ -70,15 +56,23 @@ public class JmsDurableTopicSlowReceiveTest extends JmsTopicSendReceiveTest {
* *
* @see junit.framework.TestCase#setUp() * @see junit.framework.TestCase#setUp()
*/ */
protected void setUp() throws Exception { protected void setUp() throws Exception{
this.durable = true; this.durable=true;
createBroker(); broker=createBroker();
super.setUp(); super.setUp();
} }
protected ActiveMQConnectionFactory createConnectionFactory() throws Exception { protected void tearDown() throws Exception{
ActiveMQConnectionFactory result = new ActiveMQConnectionFactory("vm://localhost"); super.tearDown();
broker.stop();
}
protected ActiveMQConnectionFactory createConnectionFactory() throws Exception{
ActiveMQConnectionFactory result=new ActiveMQConnectionFactory("vm://localhost");
Properties props=new Properties();
props.put("prefetchPolicy.durableTopicPrefetch","5");
props.put("prefetchPolicy.optimizeDurableTopicPrefetch","5");
result.setProperties(props);
return result; return result;
} }
@ -90,111 +84,88 @@ public class JmsDurableTopicSlowReceiveTest extends JmsTopicSendReceiveTest {
} }
protected void configureBroker(BrokerService answer) throws Exception{ protected void configureBroker(BrokerService answer) throws Exception{
//KahaPersistenceAdapter adapter = new KahaPersistenceAdapter(new File("activemq-data/durableTest")); //KahaPersistenceAdapter adapter=new KahaPersistenceAdapter(new File("activemq-data/durableTest"));
//answer.setPersistenceAdapter(adapter);
//JDBCPersistenceAdapter adapter = new JDBCPersistenceAdapter(); //JDBCPersistenceAdapter adapter = new JDBCPersistenceAdapter();
//answer.setPersistenceAdapter(adapter); // answer.setPersistenceAdapter(adapter);
answer.setDeleteAllMessagesOnStartup(true); answer.setDeleteAllMessagesOnStartup(true);
} }
/** /**
* Test if all the messages sent are being received. * Test if all the messages sent are being received.
* *
* @throws Exception * @throws Exception
*/ */
public void testSlowReceiver() throws Exception { public void testSlowReceiver() throws Exception{
connection2 = createConnection(); connection2=createConnection();
connection2.setClientID("test"); connection2.setClientID("test");
connection2.start(); connection2.start();
consumeSession2=connection2.createSession(false,Session.AUTO_ACKNOWLEDGE);
consumeSession2 = connection2.createSession(false, session2=connection2.createSession(false,Session.AUTO_ACKNOWLEDGE);
Session.AUTO_ACKNOWLEDGE); consumerDestination2=session2.createTopic(getConsumerSubject()+"2");
session2 = connection2.createSession(false, Session.AUTO_ACKNOWLEDGE); consumer2=consumeSession2.createDurableSubscriber((Topic)consumerDestination2,getName());
consumerDestination2 = session2.createTopic(getConsumerSubject() + "2");
consumer2 = consumeSession2.createDurableSubscriber(
(Topic) consumerDestination2, getName());
Thread.sleep(1000); Thread.sleep(1000);
consumer2.close(); consumer2.close();
connection2.close(); connection2.close();
new Thread(new Runnable(){
new Thread(new Runnable() { public void run(){
public void run() { try{
try { for(int loop=0;loop<4;loop++){
for (int loop = 0; loop < 4; loop++) { connection2=createConnection();
connection2 = createConnection();
connection2.start(); connection2.start();
session2 = connection2.createSession(false, session2=connection2.createSession(false,Session.AUTO_ACKNOWLEDGE);
Session.AUTO_ACKNOWLEDGE); producer2=session2.createProducer(null);
producer2 = session2.createProducer(null);
producer2.setDeliveryMode(deliveryMode); producer2.setDeliveryMode(deliveryMode);
Thread.sleep(1000); Thread.sleep(1000);
for(int i=0;i<NMSG/4;i++){
for (int i = 0; i < NMSG / 4; i++) { BytesMessage message=session2.createBytesMessage();
BytesMessage message = session2
.createBytesMessage();
message.writeBytes(new byte[MSIZE]); message.writeBytes(new byte[MSIZE]);
message.setStringProperty("test", "test"); message.setStringProperty("test","test");
message.setJMSType("test"); message.setJMSType("test");
producer2.send(consumerDestination2, message); producer2.send(consumerDestination2,message);
Thread.sleep(50); Thread.sleep(50);
System.err.println("Sent(" + loop +"): " + i); System.err.println("Sent("+loop+"): "+i);
} }
producer2.close(); producer2.close();
connection2.stop(); connection2.stop();
connection2.close(); connection2.close();
} }
} catch (Throwable e) { }catch(Throwable e){
e.printStackTrace(); e.printStackTrace();
} }
} }
},"SENDER Thread").start(); },"SENDER Thread").start();
connection3 = createConnection(); connection3=createConnection();
connection3.setClientID("test"); connection3.setClientID("test");
connection3.start(); connection3.start();
consumeSession3 = connection3.createSession(false, consumeSession3=connection3.createSession(false,Session.CLIENT_ACKNOWLEDGE);
Session.CLIENT_ACKNOWLEDGE); consumer3=consumeSession3.createDurableSubscriber((Topic)consumerDestination2,getName());
consumer3 = consumeSession3.createDurableSubscriber(
(Topic) consumerDestination2, getName());
connection3.close(); connection3.close();
for(int loop=0;loop<4;++loop){
connection3=createConnection();
for (int loop= 0; loop < 4; ++loop) {
connection3 = createConnection();
connection3.setClientID("test"); connection3.setClientID("test");
connection3.start(); connection3.start();
consumeSession3 = connection3.createSession(false, consumeSession3=connection3.createSession(false,Session.CLIENT_ACKNOWLEDGE);
Session.CLIENT_ACKNOWLEDGE); consumer3=consumeSession3.createDurableSubscriber((Topic)consumerDestination2,getName());
consumer3 = consumeSession3.createDurableSubscriber( Message msg=null;
(Topic) consumerDestination2, getName());
Message msg = null;
int i; int i;
for (i = 0; i < NMSG / 4; i++) { for(i=0;i<NMSG/4;i++){
// System.err.println("Receive..."); // System.err.println("Receive...");
msg = consumer3.receive(Integer.MAX_VALUE); msg=consumer3.receive(10000);
if (msg == null) if(msg==null)
break; break;
System.err.println("Received(" + loop + "): " + i ); System.err.println("Received("+loop+"): "+i);
Thread.sleep(500); Thread.sleep(500);
msg.acknowledge(); msg.acknowledge();
} }
consumer3.close(); consumer3.close();
assertEquals("Receiver " + loop , NMSG/4,i); assertEquals("Receiver "+loop,NMSG/4,i);
assertNotNull(msg); assertNotNull(msg);
// assertEquals(((BytesMessage) msg).getText(), "test"); // assertEquals(((BytesMessage) msg).getText(), "test");
assertEquals(msg.getJMSType(), "test"); assertEquals(msg.getJMSType(),"test");
assertEquals(msg.getStringProperty("test"), "test"); assertEquals(msg.getStringProperty("test"),"test");
// connection3.stop();
//connection3.stop();
connection3.close(); connection3.close();
} }
} }
} }