[jboss-cvs] JBoss Messaging SVN: r6091 - trunk/tests/src/org/jboss/messaging/tests/integration/client.

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

Author: ataylor
Date: 2009-03-16 09:25:50 -0400 (Mon, 16 Mar 2009)
New Revision: 6091

Modified:
   trunk/tests/src/org/jboss/messaging/tests/integration/client/ClientEndToEndTest.java
Log:
more end to end tests

Modified: trunk/tests/src/org/jboss/messaging/tests/integration/client/ClientEndToEndTest.java
===================================================================
--- trunk/tests/src/org/jboss/messaging/tests/integration/client/ClientEndToEndTest.java	2009-03-16 11:07:00 UTC (rev 6090)
+++ trunk/tests/src/org/jboss/messaging/tests/integration/client/ClientEndToEndTest.java	2009-03-16 13:25:50 UTC (rev 6091)
@@ -22,11 +22,13 @@
 package org.jboss.messaging.tests.integration.client;
 
 import org.jboss.messaging.core.client.ClientConsumer;
+import org.jboss.messaging.core.client.ClientFileMessage;
 import org.jboss.messaging.core.client.ClientMessage;
 import org.jboss.messaging.core.client.ClientProducer;
 import org.jboss.messaging.core.client.ClientSession;
 import org.jboss.messaging.core.client.ClientSessionFactory;
 import org.jboss.messaging.core.client.MessageHandler;
+import org.jboss.messaging.core.client.impl.ClientFileMessageImpl;
 import org.jboss.messaging.core.exception.MessagingException;
 import org.jboss.messaging.core.server.MessagingService;
 import org.jboss.messaging.core.server.Queue;
@@ -35,6 +37,10 @@
 import org.jboss.messaging.tests.util.ServiceTestBase;
 import org.jboss.messaging.utils.SimpleString;
 
+import java.io.File;
+import java.io.FileOutputStream;
+import java.nio.ByteBuffer;
+import java.nio.channels.FileChannel;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -1263,7 +1269,125 @@
       }
    }
 
+   public void testSendConsumeLargeMessage() throws Exception
+   {
+      MessagingService messagingService = createService(false);
+      try
+      {
+         messagingService.start();
+         ClientSessionFactory cf = createInVMFactory();
+         cf.setMinLargeMessageSize(1000);
+         ClientSession sendSession = cf.createSession(false, true, true);
+         ClientSession recSession = cf.createSession(false, true, true);
+         sendSession.createQueue(addressA, queueA, false);
+         ClientProducer cp = sendSession.createProducer(addressA);
+         ClientConsumer cc = recSession.createConsumer(queueA);
+         recSession.start();
+         ClientMessage message = recSession.createClientMessage(false);
+         byte[] bytes = new byte[3000];
+         message.getBody().writeBytes(bytes);
+         cp.send(message);
+         ClientMessage m = cc.receive(5000);
+         assertNotNull(m);
+         byte[] recBytes = new byte[3000];
+         m.getBody().readBytes(recBytes);
+         assertEqualsByteArrays(bytes, recBytes);
+      }
+      finally
+      {
+         if(messagingService.isStarted())
+         {
+            messagingService.stop();
+         }
+      }
+   }
 
+   public void testConsumeFileMessage() throws Exception
+   {
+      String testDir = System.getProperty("java.io.tmpdir", "/tmp") + "/jbm-unit-test";
+
+      MessagingService messagingService = createService(false);
+      try
+      {
+         messagingService.start();
+         ClientSessionFactory cf = createInVMFactory();
+         cf.setMinLargeMessageSize(1000);
+         ClientSession sendSession = cf.createSession(false, true, true);
+         ClientSession recSession = cf.createSession(false, true, true);
+         sendSession.createQueue(addressA, queueA, false);
+         ClientProducer cp = sendSession.createProducer(addressA);
+         File directory = new File(testDir);
+         directory.mkdirs();
+         ClientConsumer cc = recSession.createFileConsumer(directory, queueA);
+         recSession.start();
+         ClientMessage message = recSession.createClientMessage(false);
+         byte[] bytes = new byte[3000];
+         message.getBody().writeBytes(bytes);
+         cp.send(message);
+         ClientFileMessageImpl m = (ClientFileMessageImpl) cc.receive(5000);
+         assertNotNull(m);
+         FileChannel channel = m.getChannel();
+         ByteBuffer dst = ByteBuffer.allocate(3000);
+         channel.read(dst);
+         assertEqualsByteArrays(bytes,dst.array());
+         sendSession.close();
+         recSession.close();
+      }
+      finally
+      {
+         if(messagingService.isStarted())
+         {
+            messagingService.stop();
+         }
+      }
+   }
+
+   public void testProduceFileMessage() throws Exception
+   {
+      String testDir = System.getProperty("java.io.tmpdir", "/tmp") + "/jbm-unit-test";
+
+      MessagingService messagingService = createService(false);
+      try
+      {
+         messagingService.start();
+         ClientSessionFactory cf = createInVMFactory();
+         cf.setMinLargeMessageSize(1000);
+         ClientSession sendSession = cf.createSession(false, true, true);
+         ClientSession recSession = cf.createSession(false, true, true);
+         sendSession.createQueue(addressA, queueA, false);
+         ClientProducer cp = sendSession.createProducer(addressA);
+         File directory = new File(testDir);
+         directory.delete();
+         directory.mkdirs();
+         ClientConsumer cc = recSession.createConsumer(queueA);
+         recSession.start();
+         ClientFileMessage message = sendSession.createFileMessage(false);
+         byte[] bytes = new byte[3000];
+         File src = new File(directory, "test.jbm");
+         src.createNewFile();
+         FileOutputStream fos = new FileOutputStream(src);
+         fos.write(bytes);
+         fos.close();
+         message.setFile(src);
+         cp.send(message);
+         ClientMessage m =  cc.receive(5000);
+         assertNotNull(m);
+         byte[] recBytes = new byte[3000];
+         m.getBody().readBytes(recBytes);
+         assertEqualsByteArrays(bytes,recBytes);
+         sendSession.close();
+         recSession.close();
+      }
+      finally
+      {
+         if(messagingService.isStarted())
+         {
+            messagingService.stop();
+         }
+      }
+   }
+
+
    class Receiver implements MessageHandler
    {
       final CountDownLatch latch;