[jboss-cvs] JBoss Messaging SVN: r5549 - in trunk: src/main/org/jboss/messaging/core/postoffice/impl and 1 other directories.

[jboss-cvs-commits at lists.jboss.org]

Author: clebert.suconic at jboss.com
Date: 2008-12-19 12:11:39 -0500 (Fri, 19 Dec 2008)
New Revision: 5549

Modified:
   trunk/src/main/org/jboss/messaging/core/paging/impl/PagingManagerImpl.java
   trunk/src/main/org/jboss/messaging/core/postoffice/impl/PostOfficeImpl.java
   trunk/tests/src/org/jboss/messaging/tests/integration/paging/PagingServiceIntegrationTest.java
Log:
Fixing testsuite on paging after PostOffice changes

Modified: trunk/src/main/org/jboss/messaging/core/paging/impl/PagingManagerImpl.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/paging/impl/PagingManagerImpl.java	2008-12-19 16:08:08 UTC (rev 5548)
+++ trunk/src/main/org/jboss/messaging/core/paging/impl/PagingManagerImpl.java	2008-12-19 17:11:39 UTC (rev 5549)
@@ -159,7 +159,7 @@
 
       if (store == null)
       {
-         throw new IllegalStateException("Store " + storeName + " not found on paging");
+         store = createPageStore(storeName);
       }
 
       return store;

Modified: trunk/src/main/org/jboss/messaging/core/postoffice/impl/PostOfficeImpl.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/postoffice/impl/PostOfficeImpl.java	2008-12-19 16:08:08 UTC (rev 5548)
+++ trunk/src/main/org/jboss/messaging/core/postoffice/impl/PostOfficeImpl.java	2008-12-19 17:11:39 UTC (rev 5549)
@@ -38,6 +38,7 @@
 import org.jboss.messaging.core.logging.Logger;
 import org.jboss.messaging.core.management.ManagementService;
 import org.jboss.messaging.core.paging.PagingManager;
+import org.jboss.messaging.core.paging.PagingStore;
 import org.jboss.messaging.core.persistence.StorageManager;
 import org.jboss.messaging.core.postoffice.AddressManager;
 import org.jboss.messaging.core.postoffice.Binding;
@@ -319,39 +320,31 @@
 
    public List<MessageReference> route(final ServerMessage message) throws Exception
    {
-     // long size = pagingManager.addSize(message);
-      
-      //FIXME - paging
-      long size = Long.MAX_VALUE;
+      SimpleString address = message.getDestination();
 
-      if (size < 0)
+      if (checkAllowable)
       {
-         return emptyList;
-      }
-      else
-      {
-         SimpleString address = message.getDestination();
-
-         if (checkAllowable)
+         if (!addressManager.containsDestination(address))
          {
-            if (!addressManager.containsDestination(address))
-            {
-               throw new MessagingException(MessagingException.ADDRESS_DOES_NOT_EXIST,
-                                            "Cannot route to address " + address);
-            }
+            throw new MessagingException(MessagingException.ADDRESS_DOES_NOT_EXIST,
+                                         "Cannot route to address " + address);
          }
+      }
 
-         Bindings bindings = addressManager.getBindings(address);
+      Bindings bindings = addressManager.getBindings(address);
 
-         if (bindings != null)
-         {
-            return bindings.route(message);
-         }
-         else
-         {
-            return emptyList;
-         }
+      if (bindings != null)
+      {
+         List<MessageReference> references = bindings.route(message);
+         
+         computePaging(address, message, references);
+         
+         return references;
       }
+      else
+      {
+         return emptyList;
+      }
    }
 
    public PagingManager getPagingManager()
@@ -448,7 +441,30 @@
    }
 
    // Private -----------------------------------------------------------------
+   
+   /**
+    * Add sizes on Paging
+    * @param address
+    * @param message
+    * @param references
+    * @throws Exception
+    */
+   private void computePaging(SimpleString address, final ServerMessage message, List<MessageReference> references) throws Exception
+   {
+      if (references.size() > 0)
+      {
+         PagingStore store = pagingManager.getPageStore(address);
+         
+         store.addSize(message.getMemoryEstimate());
+         
+         for (MessageReference ref: references)
+         {
+            store.addSize(ref.getMemoryEstimate());
+         }
+      }
+   }
 
+
    private Binding createBinding(final SimpleString address,
                                  final SimpleString name,
                                  final Filter filter,
@@ -517,6 +533,9 @@
          queues.put(binding.getQueue().getPersistenceID(), binding.getQueue());
       }
 
+      preInitPageDestinations();
+
+      
       Map<SimpleString, List<Pair<SimpleString, Long>>> duplicateIDMap = new HashMap<SimpleString, List<Pair<SimpleString, Long>>>();
 
       storageManager.loadMessageJournal(this, queues, resourceManager, duplicateIDMap);
@@ -538,6 +557,19 @@
       pagingManager.startGlobalDepage();
    }
 
+   /**
+    * We need to pre-initialize already existent destinations on loading, or resuming Depage after restart won't work
+    * @throws Exception
+    */
+   private void preInitPageDestinations() throws Exception
+   {
+      Set<SimpleString> destinations = addressManager.getDestinations();
+      for (SimpleString destination : destinations)
+      {
+         pagingManager.createPageStore(destination);
+      }
+   }
+
    private class MessageExpiryRunner extends Thread
    {
       @Override

Modified: trunk/tests/src/org/jboss/messaging/tests/integration/paging/PagingServiceIntegrationTest.java
===================================================================
--- trunk/tests/src/org/jboss/messaging/tests/integration/paging/PagingServiceIntegrationTest.java	2008-12-19 16:08:08 UTC (rev 5548)
+++ trunk/tests/src/org/jboss/messaging/tests/integration/paging/PagingServiceIntegrationTest.java	2008-12-19 17:11:39 UTC (rev 5549)
@@ -258,6 +258,8 @@
          
 
          
+         assertTrue(messagingService.getServer().getPostOffice().getPagingManager().isPaging(ADDRESS));
+         
          session.start();
          
          ClientSession sessionTransacted = sf.createSession(null, null, false, false, false, false, 0);