https://issues.apache.org/jira/browse/AMQ-3986 - respect prefetch values for MDBs

git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@1379433 13f79535-47bb-0310-9956-ffa450edef68

activemq-ra/src/main/java/org/apache/activemq/ra/ActiveMQEndpointWorker.java

@@ -21,6 +21,7 @@ import java.lang.reflect.Method;
 import java.util.concurrent.atomic.AtomicBoolean;
 import javax.jms.Connection;
 import javax.jms.ConnectionConsumer;
+import javax.jms.Destination;
 import javax.jms.ExceptionListener;
 import javax.jms.JMSException;
 import javax.jms.Message;
@@ -125,21 +126,20 @@ public class ActiveMQEndpointWorker {
                         });
                         connection.start();
 
-                        int prefetchSize = activationSpec.getMaxMessagesPerSessionsIntValue() * activationSpec.getMaxSessionsIntValue();
                         if (activationSpec.isDurableSubscription()) {
                             consumer = connection.createDurableConnectionConsumer(
                                     (Topic) dest,
                                     activationSpec.getSubscriptionName(),
                                     emptyToNull(activationSpec.getMessageSelector()),
                                     serverSessionPool,
-                                    prefetchSize,
+                                    connection.getPrefetchPolicy().getDurableTopicPrefetch(),
                                     activationSpec.getNoLocalBooleanValue());
                         } else {
                             consumer = connection.createConnectionConsumer(
                                     dest,
                                     emptyToNull(activationSpec.getMessageSelector()),
                                     serverSessionPool,
-                                    prefetchSize,
+                                    getPrefetch(activationSpec, connection, dest),
                                     activationSpec.getNoLocalBooleanValue());
                         }
 
@@ -161,6 +161,16 @@ public class ActiveMQEndpointWorker {
                 }
             }
 
+            private int getPrefetch(MessageActivationSpec activationSpec, ActiveMQConnection connection, ActiveMQDestination destination) {
+                if (destination.isTopic()) {
+                    return connection.getPrefetchPolicy().getTopicPrefetch();
+                } else if (destination.isQueue()) {
+                    return connection.getPrefetchPolicy().getQueuePrefetch();
+                } else {
+                    return activationSpec.getMaxMessagesPerSessionsIntValue() * activationSpec.getMaxSessionsIntValue();
+                }
+            }
+            
             private void pause(JMSException error) {
                 if (currentReconnectDelay == MAX_RECONNECT_DELAY) {
                     LOG.error("Failed to connect to broker [" + adapter.getInfo().getServerUrl() + "]: " 

activemq-ra/src/test/java/org/apache/activemq/ra/MDBTest.java

@@ -48,7 +48,12 @@ import javax.transaction.xa.Xid;
 
 import junit.framework.TestCase;
 import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.activemq.advisory.AdvisorySupport;
+import org.apache.activemq.broker.BrokerRegistry;
+import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.command.ActiveMQMessage;
 import org.apache.activemq.command.ActiveMQQueue;
+import org.apache.activemq.command.ConsumerInfo;
 
 public class MDBTest extends TestCase {
 
@@ -133,10 +138,14 @@ public class MDBTest extends TestCase {
 
         ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("vm://localhost?broker.persistent=false");
         Connection connection = factory.createConnection();
+        connection.start();
         Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
 
+        MessageConsumer advisory = session.createConsumer(AdvisorySupport.getConsumerAdvisoryTopic(new ActiveMQQueue("TEST")));
+
         ActiveMQResourceAdapter adapter = new ActiveMQResourceAdapter();
         adapter.setServerUrl("vm://localhost?broker.persistent=false");
+        adapter.setQueuePrefetch(1);
         adapter.start(new StubBootstrapContext());
 
         final CountDownLatch messageDelivered = new CountDownLatch(1);
@@ -168,16 +177,17 @@ public class MDBTest extends TestCase {
         // Activate an Endpoint
         adapter.endpointActivation(messageEndpointFactory, activationSpec);
 
-        // Give endpoint a chance to setup and register its listeners
+        ActiveMQMessage msg = (ActiveMQMessage)advisory.receive(1000);
-        try {
+        if (msg != null) {
-            Thread.sleep(1000);
+            assertEquals("Prefetch size hasn't been set", 1, ((ConsumerInfo)msg.getDataStructure()).getPrefetchSize());
-        } catch (Exception e) {
+        } else {
-
+            fail("Consumer hasn't been created");
         }
 
         // Send the broker a message to that endpoint
         MessageProducer producer = session.createProducer(new ActiveMQQueue("TEST"));
         producer.send(session.createTextMessage("Hello!"));
+
         connection.close();
 
         // Wait for the message to be delivered.