[jboss-cvs] JBoss Messaging SVN: r5631 - branches/Branch_Failover_Page/src/main/org/jboss/messaging/core/client/impl.

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

Author: clebert.suconic at jboss.com
Date: 2009-01-13 21:46:48 -0500 (Tue, 13 Jan 2009)
New Revision: 5631

Modified:
   branches/Branch_Failover_Page/src/main/org/jboss/messaging/core/client/impl/ClientConsumerImpl.java
Log:
Tweak for flow control on largeMessage

Modified: branches/Branch_Failover_Page/src/main/org/jboss/messaging/core/client/impl/ClientConsumerImpl.java
===================================================================
--- branches/Branch_Failover_Page/src/main/org/jboss/messaging/core/client/impl/ClientConsumerImpl.java	2009-01-14 02:28:21 UTC (rev 5630)
+++ branches/Branch_Failover_Page/src/main/org/jboss/messaging/core/client/impl/ClientConsumerImpl.java	2009-01-14 02:46:48 UTC (rev 5631)
@@ -344,7 +344,7 @@
       }
 
       // Flow control for the first packet, we will have others
-      flowControl(packet.getPacketSize());
+      flowControl(packet.getPacketSize(), true);
 
       currentChunkMessage = createFileMessage(packet.getLargeMessageHeader());
    }
@@ -360,7 +360,7 @@
 
       if (chunk.isContinues())
       {
-         flowControl(chunk.getPacketSize());
+         flowControl(chunk.getPacketSize(), true);
       }
 
       if (isFileConsumer())
@@ -468,7 +468,7 @@
       sessionExecutor.execute(runner);
    }
 
-   private void flowControl(final int messageBytes) throws MessagingException
+   private void flowControl(final int messageBytes, final boolean useExecutor) throws MessagingException
    {
       if (clientWindowSize > 0)
       {
@@ -480,14 +480,21 @@
 
             creditsToSend = 0;
 
-            sessionExecutor.execute(new Runnable(){
-
-               public void run()
-               {
-                  channel.send(new SessionConsumerFlowCreditMessage(id, credits));
-               }
-               
-            });
+            if (useExecutor)
+            {
+               sessionExecutor.execute(new Runnable(){
+   
+                  public void run()
+                  {
+                     channel.send(new SessionConsumerFlowCreditMessage(id, credits));
+                  }
+                  
+               });
+            }
+            else
+            {
+               channel.send(new SessionConsumerFlowCreditMessage(id, credits));
+            }
          }
       }
    }
@@ -576,7 +583,7 @@
    private void flowControlBeforeConsumption(final ClientMessageInternal message) throws MessagingException
    {
       // Chunk messages will execute the flow control while receiving the chunks
-      flowControl(message.getFlowControlSize());
+      flowControl(message.getFlowControlSize(), false);
    }
 
    private void doCleanUp(final boolean sendCloseMessage) throws MessagingException