[hornetq-commits] JBoss hornetq SVN: r11956 - in branches/Branch_2_2_AS7/src/main/org/hornetq: ra/recovery and 1 other directory.
Author: clebert.suconic(a)jboss.com
Date: 2012-01-03 12:49:08 -0500 (Tue, 03 Jan 2012)
New Revision: 11956
Modified:
branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/ClientConsumerImpl.java
branches/Branch_2_2_AS7/src/main/org/hornetq/ra/recovery/RecoveryManager.java
Log:
Merge changes from EAP
Modified:
branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/ClientConsumerImpl.java
===================================================================
---
branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/ClientConsumerImpl.java 2012-01-03
16:46:17 UTC (rev 11955)
+++
branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/ClientConsumerImpl.java 2012-01-03
17:49:08 UTC (rev 11956)
@@ -17,7 +17,9 @@
import java.security.AccessController;
import java.security.PrivilegedAction;
import java.util.Iterator;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executor;
+import java.util.concurrent.TimeUnit;
import org.hornetq.api.core.HornetQException;
import org.hornetq.api.core.SimpleString;
@@ -47,7 +49,7 @@
{
// Constants
//
------------------------------------------------------------------------------------
-
+
private static final Logger log = Logger.getLogger(ClientConsumerImpl.class);
private static final boolean isTrace = log.isTraceEnabled();
@@ -568,7 +570,7 @@
// consumed in, which means that acking all up to won't work
ackIndividually = true;
}
-
+
// Add it to the buffer
buffer.addTail(messageToHandle, messageToHandle.getPriority());
@@ -823,6 +825,25 @@
if (clientWindowSize == 0)
{
sendCredits(0);
+
+ // If resetting a slow consumer, we need to wait the execution
+ final CountDownLatch latch = new CountDownLatch(1);
+ flowControlExecutor.execute(new Runnable()
+ {
+ public void run()
+ {
+ latch.countDown();
+ }
+ });
+
+ try
+ {
+ latch.await(10, TimeUnit.SECONDS);
+ }
+ catch (InterruptedException ignored)
+ {
+ // no big deal
+ }
}
}
Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/ra/recovery/RecoveryManager.java
===================================================================
---
branches/Branch_2_2_AS7/src/main/org/hornetq/ra/recovery/RecoveryManager.java 2012-01-03
16:46:17 UTC (rev 11955)
+++
branches/Branch_2_2_AS7/src/main/org/hornetq/ra/recovery/RecoveryManager.java 2012-01-03
17:49:08 UTC (rev 11956)
@@ -90,7 +90,14 @@
for (int i = 0 ; i < locatorClasses.length; i++)
{
- registry = (RecoveryRegistry) safeInitNewInstance(locatorClasses[i]);
+ try
+ {
+ registry = (RecoveryRegistry) safeInitNewInstance(locatorClasses[i]);
+ }
+ catch (Throwable e)
+ {
+ log.debug("unable to load recovery registry " + locatorClasses[i],
e);
+ }
if (registry != null)
{
break;