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

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

Author: clebert.suconic at jboss.com
Date: 2009-01-18 19:32:23 -0500 (Sun, 18 Jan 2009)
New Revision: 5656

Modified:
   branches/Branch_Failover_Page/src/main/org/jboss/messaging/core/server/impl/ServerSessionImpl.java
Log:
Fix on largeMessage (SendLock at the wrong order)

Modified: branches/Branch_Failover_Page/src/main/org/jboss/messaging/core/server/impl/ServerSessionImpl.java
===================================================================
--- branches/Branch_Failover_Page/src/main/org/jboss/messaging/core/server/impl/ServerSessionImpl.java	2009-01-18 23:28:34 UTC (rev 5655)
+++ branches/Branch_Failover_Page/src/main/org/jboss/messaging/core/server/impl/ServerSessionImpl.java	2009-01-19 00:32:23 UTC (rev 5656)
@@ -109,6 +109,11 @@
    private static final Logger log = Logger.getLogger(ServerSessionImpl.class);
 
    private static final boolean trace = log.isTraceEnabled();
+   
+   private static void trace(String message)
+   {
+      log.trace(message);
+   }
 
    // Static -------------------------------------------------------------------------------
 
@@ -350,6 +355,10 @@
 
       if (result == null)
       {
+         if (trace)
+         {
+            trace("(NoReplication) CreateQueue address =  " + packet.getAddress() + " queueName = " + packet.getQueueName());
+         }
          doHandleCreateQueue(packet);
       }
       else
@@ -359,6 +368,10 @@
          {
             public void run()
             {
+               if (trace)
+               {
+	               trace("(Replication) CreateQueue address =  " + packet.getAddress() + " queueName = " + packet.getQueueName());
+	           }
                doHandleCreateQueue(packet);
 
                lock.unlock();
@@ -387,6 +400,10 @@
 
       if (result == null)
       {
+         if (trace)
+         {
+            trace("(NoReplication) DeleteQueue queueName = " + packet.getQueueName());
+         }
          doHandleDeleteQueue(packet);
       }
       else
@@ -396,6 +413,10 @@
          {
             public void run()
             {
+               if (trace)
+               {
+                  trace("(Replication) DeleteQueue queueName = " + packet.getQueueName());
+               }
                doHandleDeleteQueue(packet);
 
                lock.unlock();
@@ -1039,7 +1060,9 @@
          return;
       }
       
-      
+
+      final ServerMessage msg = packet.getServerMessage();
+
       final SendLock lock;
 
       if (channel.getReplicatingChannel() != null)
@@ -1054,6 +1077,11 @@
       }
 
       
+      if (channel.getReplicatingChannel() != null)
+      {
+         msg.putBooleanProperty(new SimpleString("clustered"), true);
+      }
+
       DelayedResult result = channel.replicatePacket(packet);
 
       // With a send we must make sure it is replicated to backup before being processed on live
@@ -1061,6 +1089,10 @@
 
       if (result == null)
       {
+         if (trace)
+         {
+            trace("(withoutReplication) SendLargeMessage, id=" + msg.getMessageID());
+         }
          doSendLargeMessage(packet);
       }
       else
@@ -1069,6 +1101,10 @@
          {
             public void run()
             {
+               if (trace)
+               {
+	               trace("(Replication) SendLargeMessage, id=" + msg.getMessageID());
+	           }
                doSendLargeMessage(packet);
                lock.afterSend();
             }
@@ -1136,8 +1172,6 @@
 
    public void handleSendContinuations(final SessionSendContinuationMessage packet)
    {
-      DelayedResult result = channel.replicatePacket(packet);
-
       final SendLock lock;
 
       if (channel.getReplicatingChannel() != null)
@@ -1154,8 +1188,17 @@
       }
 
 
+      DelayedResult result = channel.replicatePacket(packet);
+
       if (result == null)
       {
+         if (trace)
+         {
+            if (!packet.isContinues())
+            {
+               trace("(NoReplication) Sending LasChunk MessageID = " + largeMessage.getMessageID());
+            }
+         }
          doSendContinuations(packet);
       }
       else
@@ -1164,6 +1207,13 @@
          {
             public void run()
             {
+               if (trace)
+               {
+                  if (!packet.isContinues())
+                  {
+                     trace("(Replication) Sending LasChunk MessageID = " + largeMessage.getMessageID());
+                  }
+               }
                doSendContinuations(packet);
                lock.afterSend();
             }
@@ -2431,6 +2481,10 @@
 
       try
       {
+         if (this.largeMessage != null)
+         {
+            log.warn("Replacing incomplete LargeMessage with ID=" + largeMessage.getMessageID());
+         }
          this.largeMessage = (LargeServerMessage)packet.getServerMessage();
          if (packet.isRequiresResponse())
          {