[jboss-svn-commits] JBL Code SVN: r10474 - labs/jbossesb/branches/JBESB_4_0_MP1/product/core/listeners/src/org/jboss/soa/esb/listeners/message.

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

Author: kevin.conner at jboss.com
Date: 2007-03-23 09:52:54 -0400 (Fri, 23 Mar 2007)
New Revision: 10474

Modified:
   labs/jbossesb/branches/JBESB_4_0_MP1/product/core/listeners/src/org/jboss/soa/esb/listeners/message/MessageAwareListener.java
Log:
Added notifications between threads

Modified: labs/jbossesb/branches/JBESB_4_0_MP1/product/core/listeners/src/org/jboss/soa/esb/listeners/message/MessageAwareListener.java
===================================================================
--- labs/jbossesb/branches/JBESB_4_0_MP1/product/core/listeners/src/org/jboss/soa/esb/listeners/message/MessageAwareListener.java	2007-03-23 13:51:55 UTC (rev 10473)
+++ labs/jbossesb/branches/JBESB_4_0_MP1/product/core/listeners/src/org/jboss/soa/esb/listeners/message/MessageAwareListener.java	2007-03-23 13:52:54 UTC (rev 10474)
@@ -43,7 +43,6 @@
 import org.jboss.soa.esb.listeners.RegistryUtil;
 import org.jboss.soa.esb.listeners.lifecycle.AbstractThreadedManagedLifecycle;
 import org.jboss.soa.esb.listeners.lifecycle.ManagedLifecycleException;
-import org.jboss.soa.esb.listeners.lifecycle.ManagedLifecycleThreadState;
 import org.jboss.soa.esb.message.Message;
 import org.jboss.soa.esb.services.registry.RegistryException;
 import org.jboss.soa.esb.util.Util;
@@ -232,12 +231,8 @@
             while (isRunning())
             {
                 // Only pickup a message when a thread is available
-                if (getActiveThreadCount() >= _maxThreads)
+                if (waitForThread(_pauseLapseInMillis))
                 {
-                    waitForRunningStateChange(ManagedLifecycleThreadState.STOPPING, _pauseLapseInMillis) ;
-                }
-                else
-                {
                     waitForEventAndProcess(100) ;
                 }
             }
@@ -342,12 +337,22 @@
             }
         }
 
-        private int getActiveThreadCount()
+        private boolean waitForThread(final long delay)
         {
-            synchronized (_synchThreads)
+            boolean result = true ;
+            synchronized(_synchThreads)
             {
-                return _qRunningThreads ;
+                if (_qRunningThreads >= _maxThreads)
+                {
+                    try
+                    {
+                        _synchThreads.wait(delay) ;
+                    }
+                    catch (final InterruptedException ie) {}
+                    result = _qRunningThreads < _maxThreads ;
+                }
             }
+            return result ;
         }
 
         private void updateThreadCount(Integer i)
@@ -355,6 +360,10 @@
             synchronized (_synchThreads)
             {
                 _qRunningThreads += i.intValue();
+                if (_qRunningThreads < _maxThreads)
+                {
+                    _synchThreads.notifyAll() ;
+                }
             }
         }
 
@@ -368,7 +377,7 @@
 
         private int _maxThreads;
 
-        private int _defaultMaxThreads;
+        private int _defaultMaxThreads = 1;
         
         private long _latencySecs;