[hornetq-commits] JBoss hornetq SVN: r8332 - in branches/ClebertTemporary: src/main/org/hornetq/core/persistence/impl/journal and 1 other directories.

[do-not-reply at jboss.org]

Author: clebert.suconic at jboss.com
Date: 2009-11-19 23:00:56 -0500 (Thu, 19 Nov 2009)
New Revision: 8332

Modified:
   branches/ClebertTemporary/src/main/org/hornetq/core/paging/impl/PagingStoreImpl.java
   branches/ClebertTemporary/src/main/org/hornetq/core/persistence/impl/journal/OperationContextImpl.java
   branches/ClebertTemporary/tests/src/org/hornetq/tests/integration/client/PagingTest.java
Log:
Fixing ordering on paging

Modified: branches/ClebertTemporary/src/main/org/hornetq/core/paging/impl/PagingStoreImpl.java
===================================================================
--- branches/ClebertTemporary/src/main/org/hornetq/core/paging/impl/PagingStoreImpl.java	2009-11-20 03:01:51 UTC (rev 8331)
+++ branches/ClebertTemporary/src/main/org/hornetq/core/paging/impl/PagingStoreImpl.java	2009-11-20 04:00:56 UTC (rev 8332)
@@ -919,6 +919,8 @@
       for (PagedMessage pagedMessage : pagedMessages)
       {
          ServerMessage message = pagedMessage.getMessage(storageManager);
+         
+         System.out.println("Depaged id = " + message.getIntProperty("id"));
 
          if (message.isLargeMessage())
          {
@@ -1008,9 +1010,10 @@
       }
 
       depageTransaction.commit();
+      
+      // TODO: If we implement ordering on AIO, we won't need to block here
+      storageManager.waitOnOperations();
 
-      storageManager.completeOperations();
-
       if (isTrace)
       {
          trace("Depage committed, running = " + running);

Modified: branches/ClebertTemporary/src/main/org/hornetq/core/persistence/impl/journal/OperationContextImpl.java
===================================================================
--- branches/ClebertTemporary/src/main/org/hornetq/core/persistence/impl/journal/OperationContextImpl.java	2009-11-20 03:01:51 UTC (rev 8331)
+++ branches/ClebertTemporary/src/main/org/hornetq/core/persistence/impl/journal/OperationContextImpl.java	2009-11-20 04:00:56 UTC (rev 8332)
@@ -44,9 +44,9 @@
 
    private List<IOAsyncTask> tasks;
    
-   private int linedup = 0;
+   private int storeLinedUp = 0;
 
-   private int replicated = 0;
+   private int stored = 0;
 
    private boolean empty = false;
 
@@ -63,12 +63,12 @@
    /** To be called by the replication manager, when new replication is added to the queue */
    public void linedUp()
    {
-      linedup++;
+      storeLinedUp++;
    }
 
    public boolean hasData()
    {
-      return linedup > 0;
+      return storeLinedUp > 0;
    }
 
    /** You may have several actions to be done after a replication operation is completed. */
@@ -90,10 +90,10 @@
       tasks.add(completion);
    }
 
-   /** To be called by the replication manager, when data is confirmed on the channel */
+   /** To be called by the storage manager, when data is confirmed on the channel */
    public synchronized void done()
    {
-      if (++replicated == linedup && complete)
+      if (++stored == storeLinedUp && complete)
       {
          flush();
       }
@@ -106,7 +106,7 @@
    {
       tlContext.set(null);
       complete = true;
-      if (replicated == linedup && complete)
+      if (stored == storeLinedUp && complete)
       {
          flush();
       }

Modified: branches/ClebertTemporary/tests/src/org/hornetq/tests/integration/client/PagingTest.java
===================================================================
--- branches/ClebertTemporary/tests/src/org/hornetq/tests/integration/client/PagingTest.java	2009-11-20 03:01:51 UTC (rev 8331)
+++ branches/ClebertTemporary/tests/src/org/hornetq/tests/integration/client/PagingTest.java	2009-11-20 04:00:56 UTC (rev 8332)
@@ -163,8 +163,7 @@
 
             assertNotNull(message2);
 
-             // TODO: AIO doesn't support ordering ATM
-//            assertEquals(i, ((Integer)message2.getObjectProperty(new SimpleString("id"))).intValue());
+            assertEquals(i, message2.getIntProperty("id").intValue());
 
             message2.acknowledge();