Author: clebert.suconic(a)jboss.com Date: 2012-02-06 21:46:48 -0500 (Mon, 06 Feb 2012) New Revision: 12088 Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/impl/QueueImpl.java branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/cluster/failover/FailoverTestBase.java branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/cluster/failover/PagingFailoverTest.java Log: HORNETQ-843/JBPAPP-8067 - fixing expiry on Paging Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/impl/QueueImpl.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/impl/QueueImpl.java 2012-02-07 02:46:06 UTC (rev 12087) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/impl/QueueImpl.java 2012-02-07 02:46:48 UTC (rev 12088) @@ -1223,8 +1223,10 @@ try { boolean expired = false; + boolean hasElements = false; while (iter.hasNext()) { + hasElements = true; MessageReference ref = iter.next(); try { @@ -1243,7 +1245,8 @@ } } - if (expired && pageIterator != null && pageIterator.hasNext()) + // If empty we need to schedule depaging to make sure we would depage expired messages as well + if ((!hasElements || expired && pageIterator != null) && pageIterator.hasNext()) { scheduleDepage(); } Modified: branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/cluster/failover/FailoverTestBase.java =================================================================== --- branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/cluster/failover/FailoverTestBase.java 2012-02-07 02:46:06 UTC (rev 12087) +++ branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/cluster/failover/FailoverTestBase.java 2012-02-07 02:46:48 UTC (rev 12088) @@ -138,6 +138,7 @@ backupConfig.setSharedStore(true); backupConfig.setBackup(true); backupConfig.setClustered(true); + backupConfig.setMessageExpiryScanPeriod(100); TransportConfiguration liveConnector = getConnectorTransportConfiguration(true); TransportConfiguration backupConnector = getConnectorTransportConfiguration(false); backupConfig.getConnectorConfigurations().put(liveConnector.getName(), liveConnector); @@ -156,6 +157,7 @@ liveConfig.setSecurityEnabled(false); liveConfig.setSharedStore(true); liveConfig.setClustered(true); + liveConfig.setMessageExpiryScanPeriod(100); List<String> pairs = null; ClusterConnectionConfiguration ccc0 = new ClusterConnectionConfiguration("cluster1", "jms", liveConnector.getName(), -1, false, false, 1, 1, pairs, false); Modified: branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/cluster/failover/PagingFailoverTest.java =================================================================== --- branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/cluster/failover/PagingFailoverTest.java 2012-02-07 02:46:06 UTC (rev 12087) +++ branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/cluster/failover/PagingFailoverTest.java 2012-02-07 02:46:48 UTC (rev 12088) @@ -31,6 +31,7 @@ import org.hornetq.core.client.impl.ClientSessionInternal; import org.hornetq.core.config.Configuration; import org.hornetq.core.server.HornetQServer; +import org.hornetq.core.server.Queue; import org.hornetq.core.settings.impl.AddressSettings; import org.hornetq.spi.core.protocol.RemotingConnection; import org.hornetq.tests.integration.cluster.util.SameProcessHornetQServer; @@ -202,6 +203,68 @@ } } + public void testExpireMessage() throws Exception + { + locator.setBlockOnNonDurableSend(true); + locator.setBlockOnDurableSend(true); + locator.setReconnectAttempts(-1); + + ClientSessionFactoryInternal sf = createSessionFactoryAndWaitForTopology(locator, 2); + ClientSession session = sf.createSession(true, true, 0); + + try + { + + session.createQueue(PagingFailoverTest.ADDRESS, PagingFailoverTest.ADDRESS, true); + + ClientProducer prod = session.createProducer(PagingFailoverTest.ADDRESS); + + final int TOTAL_MESSAGES = 1000; + + for (int i = 0; i < TOTAL_MESSAGES; i++) + { + ClientMessage msg = session.createMessage(true); + msg.putIntProperty(new SimpleString("key"), i); + msg.setExpiration(System.currentTimeMillis() + 1000); + prod.send(msg); + } + + crash(session); + + session.close(); + + Queue queue = backupServer.getServer().locateQueue(ADDRESS); + + long timeout = System.currentTimeMillis() + 60000; + System.out.println("Starting now"); + while (timeout > System.currentTimeMillis() && queue.getPageSubscription().isPaging()) + { + Thread.sleep(100); + } + + try + { + assertFalse(queue.getPageSubscription().isPaging()); + } + catch (Throwable e) + { + e.printStackTrace(); + System.exit(-1); + } + + } + finally + { + try + { + session.close(); + } + catch (Exception ignored) + { + } + } + } + /** * @param session * @param latch