Merge pull request hornetq#1969 from clebertsuconic/master

https://bugzilla.redhat.com/show_bug.cgi?id=1169781 Improving depaging process

hornetq-server/src/main/java/org/hornetq/core/server/impl/QueueImpl.java

@@ -660,6 +660,8 @@ public void deliverAsync()
             // no-op
             scheduledRunners.decrementAndGet();
          }
+
+         checkDepage();
       }
 
    }
@@ -2184,12 +2186,29 @@ else if (status == HandleStatus.NO_MATCH)
          }
       }
 
-      if (pageIterator != null && messageReferences.size() == 0 && pageSubscription.isPaging() && pageIterator.hasNext() && !depagePending)
+      checkDepage();
+   }
+
+   private void checkDepage()
+   {
+      if (pageIterator != null && pageSubscription.isPaging() && !depagePending && needsDepage() && pageIterator.hasNext())
       {
          scheduleDepage(false);
       }
    }
 
+   /**
+    * This is a common check we do before scheduling depaging.. or while depaging.
+    * Before scheduling a depage runnable we verify if it fits / needs depaging.
+    * We also check for while needsDepage While depaging.
+    * This is just to avoid a copy & paste dependency
+    * @return
+    */
+   private boolean needsDepage()
+   {
+      return queueMemorySize.get() < pageSubscription.getPagingStore().getMaxSize();
+   }
+
    private SimpleString extractGroupID(MessageReference ref)
    {
       if (internalQueue)
@@ -2264,7 +2283,7 @@ private void depage(final boolean scheduleExpiry)
       this.directDeliver = false;
 
       int depaged = 0;
-      while (timeout > System.currentTimeMillis() && queueMemorySize.get() < maxSize && pageIterator.hasNext())
+      while (timeout > System.currentTimeMillis() && needsDepage() && pageIterator.hasNext())
       {
          depaged++;
          PagedReference reference = pageIterator.next();

hornetq-server/src/main/java/org/hornetq/core/server/impl/ServerConsumerImpl.java

@@ -209,7 +209,7 @@ public ServerConsumerImpl(final long id,
 
       if (browseOnly)
       {
-         browserDeliverer = new BrowserDeliverer(messageQueue.iterator());
+         browserDeliverer = new BrowserDeliverer(messageQueue.totalIterator());
       }
       else
       {

tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/PagingTest.java

@@ -5854,6 +5854,119 @@ public void testPageHole() throws Throwable
 
    }
 
+   @Test
+   public void testMultiFiltersBrowsing() throws Throwable
+   {
+      internalTestMultiFilters(true);
+   }
+
+   @Test
+   public void testMultiFiltersRegularConsumer() throws Throwable
+   {
+      internalTestMultiFilters(false);
+   }
+
+   public void internalTestMultiFilters(boolean browsing) throws Throwable
+   {
+      clearDataRecreateServerDirs();
+
+      Configuration config = createDefaultConfig();
+      config.setJournalSyncNonTransactional(false);
+
+      server = createServer(true,
+                            config,
+                            PagingTest.PAGE_SIZE,
+                            PagingTest.PAGE_MAX,
+                            new HashMap<String, AddressSettings>());
+
+      server.start();
+
+      try
+      {
+         ServerLocator locator = createInVMNonHALocator();
+         locator.setBlockOnDurableSend(true);
+         ClientSessionFactory sf = locator.createSessionFactory();
+         ClientSession session = sf.createSession(true, true, 0);
+
+         session.createQueue(ADDRESS.toString(), "Q1", null, true);
+
+         PagingStore store = server.getPagingManager().getPageStore(ADDRESS);
+
+         ClientProducer prod = session.createProducer(ADDRESS);
+
+         ClientMessage msg = null;
+         store.startPaging();
+
+         for (int i = 0; i < 100; i++)
+         {
+            msg = session.createMessage(true);
+            msg.putStringProperty("color", "red");
+            msg.putIntProperty("count", i);
+            prod.send(msg);
+
+            if (i > 0 && i % 10 == 0)
+            {
+               store.startPaging();
+               store.forceAnotherPage();
+            }
+         }
+
+         for (int i = 0; i < 100; i++)
+         {
+            msg = session.createMessage(true);
+            msg.putStringProperty("color", "green");
+            msg.putIntProperty("count", i);
+            prod.send(msg);
+
+            if (i > 0 && i % 10 == 0)
+            {
+               store.startPaging();
+               store.forceAnotherPage();
+            }
+         }
+
+         session.commit();
+
+         session.close();
+
+         session = sf.createSession(false, false, 0);
+         session.start();
+
+
+         ClientConsumer cons1;
+
+         if (browsing)
+         {
+            cons1 = session.createConsumer("Q1", "color='green'", true);
+         }
+         else
+         {
+            cons1 = session.createConsumer("Q1", "color='red'", false);
+         }
+
+         for (int i = 0; i < 100; i++)
+         {
+            msg = cons1.receive(5000);
+
+            System.out.println("Received " + msg);
+            assertNotNull(msg);
+            if (!browsing)
+            {
+               msg.acknowledge();
+            }
+         }
+
+         session.commit();
+
+         session.close();
+      }
+      finally
+      {
+         server.stop();
+      }
+
+   }
+
 
    @Test
    public void testPendingACKOutOfOrder() throws Throwable