[jboss-cvs] JBoss Messaging SVN: r5224 - in branches/Branch_Chunk_CRS2/src/main/org/jboss/messaging/core: server/impl and 1 other directory.

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

Author: clebert.suconic at jboss.com
Date: 2008-10-30 12:25:55 -0400 (Thu, 30 Oct 2008)
New Revision: 5224

Modified:
   branches/Branch_Chunk_CRS2/src/main/org/jboss/messaging/core/client/impl/ClientConsumerImpl.java
   branches/Branch_Chunk_CRS2/src/main/org/jboss/messaging/core/server/impl/ServerConsumerImpl.java
Log:
tweaks..

Modified: branches/Branch_Chunk_CRS2/src/main/org/jboss/messaging/core/client/impl/ClientConsumerImpl.java
===================================================================
--- branches/Branch_Chunk_CRS2/src/main/org/jboss/messaging/core/client/impl/ClientConsumerImpl.java	2008-10-30 16:17:27 UTC (rev 5223)
+++ branches/Branch_Chunk_CRS2/src/main/org/jboss/messaging/core/client/impl/ClientConsumerImpl.java	2008-10-30 16:25:55 UTC (rev 5224)
@@ -70,7 +70,7 @@
 
    private final Runner runner = new Runner();
    
-   private File largeMessagesDir;
+   private File directory;
 
    private ClientMessage currentChunkMessage;
 
@@ -108,7 +108,7 @@
 
       this.direct = direct;
       
-      this.largeMessagesDir = directory;
+      this.directory = directory;
    }
 
    // ClientConsumer implementation
@@ -203,14 +203,14 @@
    {
       if (isFileConsumer())
       {
-         if (!this.largeMessagesDir.exists())
+         if (!this.directory.exists())
          {
-            largeMessagesDir.mkdirs();
+            directory.mkdirs();
          }
          
          FileClientMessageImpl message = new FileClientMessageImpl();
          message.decodeProperties(propertiesBuffer);
-         message.setFile(new File(this.largeMessagesDir, message.getMessageID() + "-" + this.session.getName() + "-" + this.getID() + ".jbm"));
+         message.setFile(new File(this.directory, message.getMessageID() + "-" + this.session.getName() + "-" + this.getID() + ".jbm"));
          message.setLargeMessage(true);
          return message;
       }
@@ -296,7 +296,7 @@
     */
    public boolean isFileConsumer()
    {
-      return largeMessagesDir != null;
+      return directory != null;
    }
    
    // ClientConsumerInternal implementation
@@ -367,7 +367,7 @@
          return;
       }
       
-      flowControl(chunk.getPacketSize());
+      flowControl(chunk.getBody().length);
       
 
       if (chunk.getHeader() != null)

Modified: branches/Branch_Chunk_CRS2/src/main/org/jboss/messaging/core/server/impl/ServerConsumerImpl.java
===================================================================
--- branches/Branch_Chunk_CRS2/src/main/org/jboss/messaging/core/server/impl/ServerConsumerImpl.java	2008-10-30 16:17:27 UTC (rev 5223)
+++ branches/Branch_Chunk_CRS2/src/main/org/jboss/messaging/core/server/impl/ServerConsumerImpl.java	2008-10-30 16:25:55 UTC (rev 5224)
@@ -491,20 +491,28 @@
             return true;
          }
 
+         if (availableCredits != null && availableCredits.get() <= 0)
+         {
+            return false;
+         }
+
          if (readAheadChunk != null)
          {
-            positionLargeMessage += readAheadChunk.getBody().length;
+            int chunkLen = readAheadChunk.getBody().length;
+            positionLargeMessage += chunkLen;
             channel.send(readAheadChunk);
             readAheadChunk = null;
-            
+            if (availableCredits != null)
+            {
+               availableCredits.addAndGet(-chunkLen);
+            }
          }
          
          while (positionLargeMessage < sizeLargeMessage)
          {
             
-            if (availableCredits.get() <= 0)
+            if (availableCredits != null && availableCredits.get() <= 0)
             {
-               
                if (readAheadChunk == null)
                {
                   readAheadChunk = createChunkSend();