[jboss-cvs] JBoss Messaging SVN: r2783 - in trunk: src/main/org/jboss/messaging/core/plugin/postoffice/cluster and 1 other directories.

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

Author: sergeypk
Date: 2007-06-15 03:56:43 -0400 (Fri, 15 Jun 2007)
New Revision: 2783

Modified:
   trunk/src/main/org/jboss/jms/server/endpoint/ServerSessionEndpoint.java
   trunk/src/main/org/jboss/messaging/core/plugin/postoffice/cluster/DefaultClusteredPostOffice.java
   trunk/tests/src/org/jboss/test/messaging/jms/message/ExpiredMessageTest.java
Log:
http://jira.jboss.com/jira/browse/JBMESSAGING-984

Modified: trunk/src/main/org/jboss/jms/server/endpoint/ServerSessionEndpoint.java
===================================================================
--- trunk/src/main/org/jboss/jms/server/endpoint/ServerSessionEndpoint.java	2007-06-14 12:16:17 UTC (rev 2782)
+++ trunk/src/main/org/jboss/jms/server/endpoint/ServerSessionEndpoint.java	2007-06-15 07:56:43 UTC (rev 2783)
@@ -86,6 +86,7 @@
 import org.jboss.util.id.GUID;
 
 import EDU.oswego.cs.dl.util.concurrent.ConcurrentHashMap;
+import EDU.oswego.cs.dl.util.concurrent.LinkedQueue;
 import EDU.oswego.cs.dl.util.concurrent.QueuedExecutor;
 import EDU.oswego.cs.dl.util.concurrent.SynchronizedLong;
 
@@ -153,7 +154,7 @@
    private SynchronizedLong deliveryIdSequence;
    
    //Temporary until we have our own NIO transport   
-   QueuedExecutor executor = new QueuedExecutor();
+   QueuedExecutor executor = new QueuedExecutor(new LinkedQueue());
    
    // Constructors ---------------------------------------------------------------------------------
 
@@ -953,7 +954,7 @@
          }
       }      
    } 
-   
+
    void promptDelivery(final Channel channel)
    {
    	if (trace) { log.trace("Prompting delivery on " + channel); }

Modified: trunk/src/main/org/jboss/messaging/core/plugin/postoffice/cluster/DefaultClusteredPostOffice.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/plugin/postoffice/cluster/DefaultClusteredPostOffice.java	2007-06-14 12:16:17 UTC (rev 2782)
+++ trunk/src/main/org/jboss/messaging/core/plugin/postoffice/cluster/DefaultClusteredPostOffice.java	2007-06-15 07:56:43 UTC (rev 2783)
@@ -274,7 +274,7 @@
 
       nbSupport = new NotificationBroadcasterSupport();
 
-      viewExecutor = new QueuedExecutor();
+      viewExecutor = new QueuedExecutor(new LinkedQueue());
 
       this.jChannelFactory = JChannelFactory;
       

Modified: trunk/tests/src/org/jboss/test/messaging/jms/message/ExpiredMessageTest.java
===================================================================
--- trunk/tests/src/org/jboss/test/messaging/jms/message/ExpiredMessageTest.java	2007-06-14 12:16:17 UTC (rev 2782)
+++ trunk/tests/src/org/jboss/test/messaging/jms/message/ExpiredMessageTest.java	2007-06-15 07:56:43 UTC (rev 2783)
@@ -35,6 +35,8 @@
    // Attributes -----------------------------------------------------------------------------------
 
    private InitialContext ic;
+   private ConnectionFactory cf;
+   private Queue queue;
 
    // Constructors ---------------------------------------------------------------------------------
 
@@ -47,9 +49,6 @@
 
    public void testSimpleExpiration() throws Exception
    {
-      ConnectionFactory cf = (ConnectionFactory)ic.lookup("/ConnectionFactory");
-      Queue queue = (Queue)ic.lookup("/queue/expiredMessageTestQueue");
-
       Connection conn = cf.createConnection();
 
       Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
@@ -70,8 +69,51 @@
       conn.start();
 
       assertNull(cons.receive(3000));
+      
+      conn.close();
    }
+   
+   public void testManyExpiredMessagesAtOnce() throws Exception
+   {
+      Connection conn = cf.createConnection();
+      
+      Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
+      
+      MessageProducer prod = session.createProducer(queue);
+      prod.setTimeToLive(1);
+      
+      Message m = session.createTextMessage("This message will die");
+      
+      final int MESSAGE_COUNT = 2000;
 
+      log.info("Going to send " + MESSAGE_COUNT + " messages");
+
+      for (int i = 0; i < MESSAGE_COUNT; i++)
+      {
+         prod.send(m);
+         if ((i + 1) % 1000 == 0)
+         {
+            log.info("Sent " + (i + 1) + " messages out of " + MESSAGE_COUNT);
+         }
+      }
+      
+      Thread.sleep(1000);
+      
+      log.info("Creating consumer");
+
+      MessageConsumer cons = session.createConsumer(queue);
+      conn.start();
+      
+      final int TIMEOUT = 3000;
+      log.info("Trying to receive a message, timeout is " + TIMEOUT + " ms");
+
+      assertNull(cons.receive(TIMEOUT));
+      
+      log.info("Done");
+      
+      conn.close();
+   }
+
   
 
    // Package protected ----------------------------------------------------------------------------
@@ -87,13 +129,21 @@
       ic = new InitialContext(ServerManagement.getJNDIEnvironment());
 
       ServerManagement.deployQueue("expiredMessageTestQueue");
+      ServerManagement.deployQueue("ExpiryQueue");
 
+      cf = (ConnectionFactory)ic.lookup("/ConnectionFactory");
+
+      queue = (Queue)ic.lookup("/queue/expiredMessageTestQueue");
+
       log.debug("setup done");
    }
 
    protected void tearDown() throws Exception
    {
       ServerManagement.undeployQueue("expiredMessageTestQueue");
+      ServerManagement.undeployQueue("ExpiryQueue");
+      
+      ic.close();
 
       ServerManagement.stop();