[hornetq-commits] JBoss hornetq SVN: r7935 - in trunk: tests/src/org/hornetq/tests/integration/client and 1 other directory.

[do-not-reply at jboss.org]

Author: clebert.suconic at jboss.com
Date: 2009-09-03 13:38:43 -0400 (Thu, 03 Sep 2009)
New Revision: 7935

Modified:
   trunk/src/main/org/hornetq/core/client/impl/LargeMessageBufferImpl.java
   trunk/tests/src/org/hornetq/tests/integration/client/LargeMessageTest.java
Log:
Adding exception check on closed consumers and large message

Modified: trunk/src/main/org/hornetq/core/client/impl/LargeMessageBufferImpl.java
===================================================================
--- trunk/src/main/org/hornetq/core/client/impl/LargeMessageBufferImpl.java	2009-09-02 10:18:43 UTC (rev 7934)
+++ trunk/src/main/org/hornetq/core/client/impl/LargeMessageBufferImpl.java	2009-09-03 17:38:43 UTC (rev 7935)
@@ -63,6 +63,8 @@
    private final long totalSize;
 
    private boolean streamEnded = false;
+   
+   private boolean streamClosed = false;
 
    private final int readTimeout;
 
@@ -213,6 +215,7 @@
    {
       packets.offer(new SessionReceiveContinuationMessage());
       streamEnded = true;
+      streamClosed = true;
 
       notifyAll();
    }
@@ -1262,6 +1265,11 @@
       {
          throw new IndexOutOfBoundsException();
       }
+      
+      if (streamClosed)
+      {
+         throw new IllegalAccessError("The consumer associated with this large message was closed before the body was read");
+      }
 
       if (fileCache == null)
       {

Modified: trunk/tests/src/org/hornetq/tests/integration/client/LargeMessageTest.java
===================================================================
--- trunk/tests/src/org/hornetq/tests/integration/client/LargeMessageTest.java	2009-09-02 10:18:43 UTC (rev 7934)
+++ trunk/tests/src/org/hornetq/tests/integration/client/LargeMessageTest.java	2009-09-03 17:38:43 UTC (rev 7935)
@@ -67,7 +67,83 @@
    // Constructors --------------------------------------------------
 
    // Public --------------------------------------------------------
+   
+   
+   public void testCloseConsumer() throws Exception
+   {
+      final int messageSize = (int)(3.5 * ClientSessionFactoryImpl.DEFAULT_MIN_LARGE_MESSAGE_SIZE);
 
+      ClientSession session = null;           
+
+      try
+      {
+         server = createServer(true);
+
+         server.start();
+         
+         log.info("*********** starting test");
+
+         ClientSessionFactory sf = createInVMFactory();
+
+         session = sf.createSession(false, false, false);
+
+         session.createTemporaryQueue(ADDRESS, ADDRESS);
+
+         ClientProducer producer = session.createProducer(ADDRESS);
+
+         Message clientFile = createLargeClientMessage(session, messageSize, true);
+
+         log.info("*********** sending large message");
+         
+         producer.send(clientFile);
+
+         session.commit();
+
+         session.start();
+
+         ClientConsumer consumer = session.createConsumer(ADDRESS);
+         ClientMessage msg1 = consumer.receive(1000);
+         msg1.acknowledge();
+         session.commit();
+         assertNotNull(msg1);
+         
+         consumer.close();
+
+         try
+         {
+            msg1.getBody().readByte();
+            fail ("Exception was expected");
+         }
+         catch (Throwable ignored)
+         {
+         }
+
+         session.close();
+
+         validateNoFilesOnLargeDir();
+      }
+      finally
+      {
+         try
+         {
+            server.stop();
+         }
+         catch (Throwable ignored)
+         {
+         }
+
+         try
+         {
+            session.close();
+         }
+         catch (Throwable ignored)
+         {
+         }
+      }
+   }
+
+
+
    public void testDLALargeMessage() throws Exception
    {
       final int messageSize = (int)(3.5 * ClientSessionFactoryImpl.DEFAULT_MIN_LARGE_MESSAGE_SIZE);