[hornetq-commits] JBoss hornetq SVN: r12084 - trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover.

[do-not-reply at jboss.org]

Author: borges
Date: 2012-02-03 11:05:31 -0500 (Fri, 03 Feb 2012)
New Revision: 12084

Modified:
   trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/BackupSyncLargeMessageTest.java
Log:
HORNETQ-720 Add test for sync start while the live is receiving largeMessage data.

Modified: trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/BackupSyncLargeMessageTest.java
===================================================================
--- trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/BackupSyncLargeMessageTest.java	2012-02-03 13:28:27 UTC (rev 12083)
+++ trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/BackupSyncLargeMessageTest.java	2012-02-03 16:05:31 UTC (rev 12084)
@@ -3,10 +3,17 @@
 import java.io.File;
 import java.util.HashSet;
 import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicBoolean;
 
+import org.hornetq.api.core.HornetQException;
 import org.hornetq.api.core.client.ClientMessage;
+import org.hornetq.api.core.client.ClientProducer;
+import org.hornetq.api.core.client.ClientSession;
 import org.hornetq.api.core.client.ServerLocator;
 import org.hornetq.core.client.impl.ServerLocatorInternal;
+import org.hornetq.tests.util.UnitTestCase;
 
 public class BackupSyncLargeMessageTest extends BackupSyncJournalTest
 {
@@ -41,6 +48,7 @@
       createProducerSendSomeMessages();
       startBackupFinishSyncing();
       receiveMsgsInRange(0, n_msgs / 2);
+      backupServer.stop();
       assertEquals("we really ought to delete these after delivery", n_msgs / 2, getAllMessageFileIds(dir).size());
    }
 
@@ -61,6 +69,54 @@
       assertEquals("we really ought to delete these after delivery", n_msgs / 2, getAllMessageFileIds(dir).size());
    }
 
+   /**
+    * LargeMessages are passed from the client to the server in chunks. Here we test the backup
+    * starting the data synchronization with the live in the middle of a multiple chunks large
+    * message upload from the client to the live server.
+    * @throws Exception
+    */
+   public void testBackupStartsWhenLiveIsReceivingLargeMessage() throws Exception
+   {
+      final ClientSession session = addClientSession(sessionFactory.createSession(true, true));
+      session.createQueue(FailoverTestBase.ADDRESS, FailoverTestBase.ADDRESS, null, true);
+      final ClientProducer producer = session.createProducer(FailoverTestBase.ADDRESS);
+      final ClientMessage message = session.createMessage(true);
+      message.setBodyInputStream(UnitTestCase.createFakeLargeStream(1000 * MIN_LARGE_MESSAGE));
+
+      final AtomicBoolean caughtException = new AtomicBoolean(false);
+      final CountDownLatch latch = new CountDownLatch(1);
+      final CountDownLatch latch2 = new CountDownLatch(1);
+
+      Runnable r = new Runnable()
+      {
+         @Override
+         public void run()
+         {
+            try
+            {
+               latch.countDown();
+               producer.send(message);
+               session.commit();
+            }
+            catch (HornetQException e)
+            {
+               e.printStackTrace();
+               caughtException.set(true);
+            }
+            finally
+            {
+               latch2.countDown();
+            }
+         }
+      };
+      Executors.defaultThreadFactory().newThread(r).start();
+      latch.await();
+      startBackupFinishSyncing();
+      latch2.await();
+      crash(session);
+      assertFalse("no exceptions while sending message", caughtException.get());
+   }
+
    private Set<Long> getAllMessageFileIds(File dir)
    {
       Set<Long> idsOnBkp = new HashSet<Long>();