QueueSubscriptionTest


1   /**
2    *
3    * Licensed to the Apache Software Foundation (ASF) under one or more
4    * contributor license agreements.  See the NOTICE file distributed with
5    * this work for additional information regarding copyright ownership.
6    * The ASF licenses this file to You under the Apache License, Version 2.0
7    * (the "License"); you may not use this file except in compliance with
8    * the License.  You may obtain a copy of the License at
9    *
10   * http://www.apache.org/licenses/LICENSE-2.0
11   *
12   * Unless required by applicable law or agreed to in writing, software
13   * distributed under the License is distributed on an "AS IS" BASIS,
14   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15   * See the License for the specific language governing permissions and
16   * limitations under the License.
17   */
18  package org.apache.activemq.broker;
19  
20  import org.apache.activemq.JmsMultipleClientsTestSupport;
21  import org.apache.activemq.ActiveMQConnectionFactory;
22  import org.apache.activemq.command.ActiveMQDestination;
23  
24  public class QueueSubscriptionTest extends JmsMultipleClientsTestSupport {
25      protected int messageCount  = 1000; // 1000 Messages per producer
26      protected int prefetchCount = 10;
27  
28      protected void setUp() throws Exception   {
29          super.setUp();
30          durable = false;
31          topic = false;
32      }
33  
34      public void testManyProducersOneConsumer() throws Exception   {
35          consumerCount = 1;
36          producerCount = 10;
37          messageCount  = 100;
38          messageSize   = 1; // 1 byte
39          prefetchCount = 10;
40  
41          doMultipleClientsTest();
42  
43          assertTotalMessagesReceived(messageCount * producerCount);
44      }
45  
46      public void testOneProducerTwoConsumersSmallMessagesOnePrefetch() throws Exception   {
47          consumerCount = 2;
48          producerCount = 1;
49          messageCount  = 1000;
50          messageSize   = 1024; // 1 Kb
51          configurePrefetchOfOne();
52  
53          doMultipleClientsTest();
54  
55          assertTotalMessagesReceived(messageCount * producerCount);
56      }
57  
58      public void testOneProducerTwoConsumersSmallMessagesLargePrefetch() throws Exception   {
59          consumerCount = 2;
60          producerCount = 1;
61          messageCount  = 1000;
62          prefetchCount = messageCount * 2;
63          messageSize   = 1024; // 1 Kb
64  
65          doMultipleClientsTest();
66  
67          assertTotalMessagesReceived(messageCount * producerCount);
68      }
69  
70      public void testOneProducerTwoConsumersLargeMessagesOnePrefetch() throws Exception   {
71          consumerCount = 2;
72          producerCount = 1;
73          messageCount  = 10;
74          messageSize   = 1024 * 1024 * 1; // 2 MB
75          configurePrefetchOfOne();
76  
77          doMultipleClientsTest();
78  
79          assertTotalMessagesReceived(messageCount * producerCount);
80      }
81  
82      public void testOneProducerTwoConsumersLargeMessagesLargePrefetch() throws Exception   {
83          consumerCount = 2;
84          producerCount = 1;
85          messageCount  = 10;
86          prefetchCount = messageCount * 2;
87          messageSize   = 1024 * 1024 * 1; // 2 MB
88  
89          doMultipleClientsTest();
90  
91          assertTotalMessagesReceived(messageCount * producerCount);
92      }
93  
94      public void testOneProducerManyConsumersFewMessages() throws Exception   {
95          consumerCount = 50;
96          producerCount = 1;
97          messageCount  = 10;
98          messageSize   = 1; // 1 byte
99          prefetchCount = 10;
100 
101         doMultipleClientsTest();
102 
103         assertTotalMessagesReceived(messageCount * producerCount);
104     }
105 
106     public void testOneProducerManyConsumersManyMessages() throws Exception   {
107         consumerCount = 50;
108         producerCount = 1;
109         messageCount  = 1000;
110         messageSize   = 1; // 1 byte
111         prefetchCount = 10;
112 
113         doMultipleClientsTest();
114 
115         assertTotalMessagesReceived(messageCount * producerCount);
116     }
117 
118     public void testManyProducersManyConsumers() throws Exception   {
119         consumerCount = 50;
120         producerCount = 50;
121         messageCount  = 100;
122         messageSize   = 1; // 1 byte
123         prefetchCount = 100;
124 
125         doMultipleClientsTest();
126 
127         assertTotalMessagesReceived(messageCount * producerCount);
128     }
129 
130     protected void configurePrefetchOfOne() {
131         prefetchCount = 1;
132 
133         // this is gonna be a bit slow what with the low prefetch so bump up the wait time
134         allMessagesList.setMaximumDuration(allMessagesList.getMaximumDuration() * 20);
135     }
136 
137     public void doMultipleClientsTest() throws Exception   {
138         // Create destination
139         final ActiveMQDestination dest = createDestination();
140 
141         // Create consumers
142         ActiveMQConnectionFactory consumerFactory = (ActiveMQConnectionFactory)createConnectionFactory();
143         consumerFactory.getPrefetchPolicy().setAll(prefetchCount);
144 
145         startConsumers(consumerFactory, dest);
146 
147         startProducers(dest, messageCount);
148 
149         // Wait for messages to be received. Make it proportional to the messages delivered.
150         int totalMessageCount = messageCount * producerCount;
151         if( dest.isTopic() )
152             totalMessageCount *= consumerCount;       
153         waitForAllMessagesToBeReceived(totalMessageCount);
154     }
155 }
156
A to Z: JavaDoc & Examples Daily Java News & Articles Open Source Projects Open Source Codes Free Computer Books Remove Frame
Popular Tags