activemq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From tab...@apache.org
Subject svn commit: r1441085 - /activemq/trunk/activemq-unit-tests/src/test/java/org/apache/activemq/OptimizedAckTest.java
Date Thu, 31 Jan 2013 17:44:04 GMT
Author: tabish
Date: Thu Jan 31 17:44:04 2013
New Revision: 1441085

URL: http://svn.apache.org/viewvc?rev=1441085&view=rev
Log:
apply fix for: https://issues.apache.org/jira/browse/AMQ-4288

Modified:
    activemq/trunk/activemq-unit-tests/src/test/java/org/apache/activemq/OptimizedAckTest.java

Modified: activemq/trunk/activemq-unit-tests/src/test/java/org/apache/activemq/OptimizedAckTest.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-unit-tests/src/test/java/org/apache/activemq/OptimizedAckTest.java?rev=1441085&r1=1441084&r2=1441085&view=diff
==============================================================================
--- activemq/trunk/activemq-unit-tests/src/test/java/org/apache/activemq/OptimizedAckTest.java
(original)
+++ activemq/trunk/activemq-unit-tests/src/test/java/org/apache/activemq/OptimizedAckTest.java
Thu Jan 31 17:44:04 2013
@@ -33,6 +33,7 @@ public class OptimizedAckTest extends Te
     private static final Logger LOG = LoggerFactory.getLogger(OptimizedAckTest.class);
     private ActiveMQConnection connection;
 
+    @Override
     protected void setUp() throws Exception {
         super.setUp();
         connection = (ActiveMQConnection) createConnection();
@@ -43,130 +44,132 @@ public class OptimizedAckTest extends Te
         connection.setPrefetchPolicy(prefetchPolicy);
     }
 
+    @Override
     protected void tearDown() throws Exception {
         connection.close();
         super.tearDown();
     }
 
-     public void testReceivedMessageStillInflight() throws Exception {
-         connection.start();
-         Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
-         Queue queue = session.createQueue("test");
-         MessageProducer producer = session.createProducer(queue);
-         for (int i = 0; i < 10; i++) {
-             producer.send(session.createTextMessage("Hello" + i));
-         }
-
-         final RegionBroker regionBroker = (RegionBroker) BrokerRegistry.getInstance().findFirst().getRegionBroker();
-         MessageConsumer consumer = session.createConsumer(queue);
-
-         assertTrue("prefetch full", Wait.waitFor(new Wait.Condition() {
-             @Override
-             public boolean isSatisified() throws Exception {
-                 LOG.info("inflight count: " + regionBroker.getDestinationStatistics().getInflight().getCount());
-                 return 10 == regionBroker.getDestinationStatistics().getInflight().getCount();
-             }
-         }));
+    public void testReceivedMessageStillInflight() throws Exception {
+        connection.start();
+        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        Queue queue = session.createQueue("test");
+        MessageProducer producer = session.createProducer(queue);
+        for (int i = 0; i < 10; i++) {
+            producer.send(session.createTextMessage("Hello" + i));
+        }
+
+        final RegionBroker regionBroker = (RegionBroker) BrokerRegistry.getInstance().findFirst().getRegionBroker();
+        MessageConsumer consumer = session.createConsumer(queue);
+
+        assertTrue("prefetch full", Wait.waitFor(new Wait.Condition() {
+            @Override
+            public boolean isSatisified() throws Exception {
+                LOG.info("inflight count: " + regionBroker.getDestinationStatistics().getInflight().getCount());
+                return 10 == regionBroker.getDestinationStatistics().getInflight().getCount();
+            }
+        }));
 
-         for (int i=0; i<10; i++) {
+        for (int i = 0; i < 10; i++) {
             javax.jms.Message msg = consumer.receive(4000);
+            long inFlightCount = regionBroker.getDestinationStatistics().getInflight().getCount();
             assertNotNull(msg);
-             if (i<7) {
-                 assertEquals("all prefetch is still in flight", 10, regionBroker.getDestinationStatistics().getInflight().getCount());
-             } else {
-                 assertTrue("most are acked but 3 remain", Wait.waitFor(new Wait.Condition(){
-                     @Override
-                     public boolean isSatisified() throws Exception {
-                         LOG.info("inflight count: " + regionBroker.getDestinationStatistics().getInflight().getCount());
-                         return 3 == regionBroker.getDestinationStatistics().getInflight().getCount();
-                     }
-                 }));
-             }
-         }
-     }
-
-     public void testVerySlowReceivedMessageStillInflight() throws Exception {
-         connection.start();
-         Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
-         Queue queue = session.createQueue("test");
-         MessageProducer producer = session.createProducer(queue);
-         for (int i = 0; i < 10; i++) {
-             producer.send(session.createTextMessage("Hello" + i));
-         }
-
-         final RegionBroker regionBroker = (RegionBroker) BrokerRegistry.getInstance().findFirst().getRegionBroker();
-         MessageConsumer consumer = session.createConsumer(queue);
-
-         assertTrue("prefetch full", Wait.waitFor(new Wait.Condition() {
-             @Override
-             public boolean isSatisified() throws Exception {
-                 LOG.info("inflight count: " + regionBroker.getDestinationStatistics().getInflight().getCount());
-                 return 10 == regionBroker.getDestinationStatistics().getInflight().getCount();
-             }
-         }));
+            if (i < 7) {
+                assertEquals("all prefetch is still in flight", 10, inFlightCount);
+            } else {
+                assertTrue("most are acked but 3 remain", Wait.waitFor(new Wait.Condition()
{
+                    @Override
+                    public boolean isSatisified() throws Exception {
+                        LOG.info("inflight count: " + regionBroker.getDestinationStatistics().getInflight().getCount());
+                        return 3 == regionBroker.getDestinationStatistics().getInflight().getCount();
+                    }
+                }));
+            }
+        }
+    }
+
+    public void testVerySlowReceivedMessageStillInflight() throws Exception {
+        connection.start();
+        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        Queue queue = session.createQueue("test");
+        MessageProducer producer = session.createProducer(queue);
+        for (int i = 0; i < 10; i++) {
+            producer.send(session.createTextMessage("Hello" + i));
+        }
+
+        final RegionBroker regionBroker = (RegionBroker) BrokerRegistry.getInstance().findFirst().getRegionBroker();
+        MessageConsumer consumer = session.createConsumer(queue);
+
+        assertTrue("prefetch full", Wait.waitFor(new Wait.Condition() {
+            @Override
+            public boolean isSatisified() throws Exception {
+                LOG.info("inflight count: " + regionBroker.getDestinationStatistics().getInflight().getCount());
+                return 10 == regionBroker.getDestinationStatistics().getInflight().getCount();
+            }
+        }));
 
-         for (int i=0; i<10; i++) {
-             Thread.sleep(400);
+        for (int i = 0; i < 10; i++) {
+            Thread.sleep(400);
             javax.jms.Message msg = consumer.receive(4000);
-             long inFlightCount = regionBroker.getDestinationStatistics().getInflight().getCount();
+            long inFlightCount = regionBroker.getDestinationStatistics().getInflight().getCount();
             assertNotNull(msg);
-             if (i<7) {
-                 assertEquals("all prefetch is still in flight: " + i, 10, inFlightCount);
-             } else {
-                 assertTrue("most are acked but 3 remain", Wait.waitFor(new Wait.Condition(){
-                     @Override
-                     public boolean isSatisified() throws Exception {
-                         return 3 == regionBroker.getDestinationStatistics().getInflight().getCount();
-                     }
-                 }));
-             }
-         }
-     }
-
-     public void testReceivedMessageNotInFlightAfterScheduledAckFires() throws Exception
{
-         connection.setOptimizedAckScheduledAckInterval(TimeUnit.SECONDS.toMillis(10));
-         connection.start();
-         Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
-         Queue queue = session.createQueue("test");
-         MessageProducer producer = session.createProducer(queue);
-         for (int i = 0; i < 10; i++) {
-             producer.send(session.createTextMessage("Hello" + i));
-         }
-
-         final RegionBroker regionBroker = (RegionBroker) BrokerRegistry.getInstance().findFirst().getRegionBroker();
-         MessageConsumer consumer = session.createConsumer(queue);
-
-         assertTrue("prefetch full", Wait.waitFor(new Wait.Condition() {
-             @Override
-             public boolean isSatisified() throws Exception {
-                 LOG.info("inflight count: " + regionBroker.getDestinationStatistics().getInflight().getCount());
-                 return 10 == regionBroker.getDestinationStatistics().getInflight().getCount();
-             }
-         }));
+            if (i < 7) {
+                assertEquals("all prefetch is still in flight: " + i, 10, inFlightCount);
+            } else {
+                assertTrue("most are acked but 3 remain", Wait.waitFor(new Wait.Condition()
{
+                    @Override
+                    public boolean isSatisified() throws Exception {
+                        return 3 == regionBroker.getDestinationStatistics().getInflight().getCount();
+                    }
+                }));
+            }
+        }
+    }
 
-         for (int i=0; i<10; i++) {
+    public void testReceivedMessageNotInFlightAfterScheduledAckFires() throws Exception {
+        connection.setOptimizedAckScheduledAckInterval(TimeUnit.SECONDS.toMillis(10));
+        connection.start();
+        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        Queue queue = session.createQueue("test");
+        MessageProducer producer = session.createProducer(queue);
+        for (int i = 0; i < 10; i++) {
+            producer.send(session.createTextMessage("Hello" + i));
+        }
+
+        final RegionBroker regionBroker = (RegionBroker) BrokerRegistry.getInstance().findFirst().getRegionBroker();
+        MessageConsumer consumer = session.createConsumer(queue);
+
+        assertTrue("prefetch full", Wait.waitFor(new Wait.Condition() {
+            @Override
+            public boolean isSatisified() throws Exception {
+                LOG.info("inflight count: " + regionBroker.getDestinationStatistics().getInflight().getCount());
+                return 10 == regionBroker.getDestinationStatistics().getInflight().getCount();
+            }
+        }));
+
+        for (int i = 0; i < 10; i++) {
             javax.jms.Message msg = consumer.receive(4000);
-             long inFlightCount = regionBroker.getDestinationStatistics().getInflight().getCount();
+            long inFlightCount = regionBroker.getDestinationStatistics().getInflight().getCount();
             assertNotNull(msg);
-             if (i<7) {
-                 assertEquals("all prefetch is still in flight", 10, inFlightCount);
-             } else {
-                 assertTrue("most are acked but 3 remain", Wait.waitFor(new Wait.Condition(){
-                     @Override
-                     public boolean isSatisified() throws Exception {
-                         LOG.info("inflight count: " + regionBroker.getDestinationStatistics().getInflight().getCount());
-                         return 3 == regionBroker.getDestinationStatistics().getInflight().getCount();
-                     }
-                 }));
-             }
-         }
-
-         assertTrue("After delay the scheduled ack should ack all inflight.", Wait.waitFor(new
Wait.Condition(){
-             @Override
-             public boolean isSatisified() throws Exception {
-                 LOG.info("inflight count: " + regionBroker.getDestinationStatistics().getInflight().getCount());
-                 return 0 == regionBroker.getDestinationStatistics().getInflight().getCount();
-             }
-         }));
-     }
+            if (i < 7) {
+                assertEquals("all prefetch is still in flight", 10, inFlightCount);
+            } else {
+                assertTrue("most are acked but 3 remain", Wait.waitFor(new Wait.Condition()
{
+                    @Override
+                    public boolean isSatisified() throws Exception {
+                        LOG.info("inflight count: " + regionBroker.getDestinationStatistics().getInflight().getCount());
+                        return 3 == regionBroker.getDestinationStatistics().getInflight().getCount();
+                    }
+                }));
+            }
+        }
+
+        assertTrue("After delay the scheduled ack should ack all inflight.", Wait.waitFor(new
Wait.Condition() {
+            @Override
+            public boolean isSatisified() throws Exception {
+                LOG.info("inflight count: " + regionBroker.getDestinationStatistics().getInflight().getCount());
+                return 0 == regionBroker.getDestinationStatistics().getInflight().getCount();
+            }
+        }));
+    }
 }



Mime
View raw message