Apache
/
activemq
mirror of https://github.com/apache/activemq.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

On consumer close, wait for it's async threads to finish before returning. Was getting errors when async acks were sent and the subscription was allready closed. 

git-svn-id: https://svn.apache.org/repos/asf/incubator/activemq/trunk@394705 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Hiram R. Chirino 2006-04-17 15:29:43 +00:00
parent 969eefeef0
commit 91720daee7
1 changed files with 8 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
activemq-core/src/main/java/org/apache/activemq/ActiveMQMessageConsumer.java
View File

@ -45,8 +45,10 @@ import org.apache.activemq.util.JMSExceptionSupport;
import org.apache.commons.logging.Log; import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory; import org.apache.commons.logging.LogFactory;
import edu.emory.mathcs.backport.java.util.concurrent.ExecutorService;
import edu.emory.mathcs.backport.java.util.concurrent.Executors;
import edu.emory.mathcs.backport.java.util.concurrent.TimeUnit;
import edu.emory.mathcs.backport.java.util.concurrent.atomic.AtomicBoolean; import edu.emory.mathcs.backport.java.util.concurrent.atomic.AtomicBoolean;
import edu.emory.mathcs.backport.java.util.concurrent.*;
/** /**
* A client uses a <CODE>MessageConsumer</CODE> object to receive messages * A client uses a <CODE>MessageConsumer</CODE> object to receive messages
@ -560,6 +562,11 @@ public class ActiveMQMessageConsumer implements MessageAvailableConsumer, StatsC
deliverAcks();//only processes optimized acknowledgements deliverAcks();//only processes optimized acknowledgements
if (executorService!=null){ if (executorService!=null){
executorService.shutdown(); executorService.shutdown();
try {
executorService.awaitTermination(60, TimeUnit.SECONDS);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
} }
if ((session.isTransacted() || session.isDupsOkAcknowledge())) { if ((session.isTransacted() || session.isDupsOkAcknowledge())) {
acknowledge(); acknowledge();