diff --git a/activemq-client/src/main/java/org/apache/activemq/ActiveMQSession.java b/activemq-client/src/main/java/org/apache/activemq/ActiveMQSession.java index 1d2ae836fb..2e0f6468fa 100755 --- a/activemq-client/src/main/java/org/apache/activemq/ActiveMQSession.java +++ b/activemq-client/src/main/java/org/apache/activemq/ActiveMQSession.java @@ -717,7 +717,7 @@ public class ActiveMQSession implements Session, QueueSession, TopicSession, Sta if (!closed) { try { - executor.stop(); + executor.close(); for (Iterator iter = consumers.iterator(); iter.hasNext();) { ActiveMQMessageConsumer consumer = iter.next(); @@ -978,11 +978,24 @@ public class ActiveMQSession implements Session, QueueSession, TopicSession, Sta for (int i = 0; i < redeliveryCounter; i++) { redeliveryDelay = redeliveryPolicy.getNextRedeliveryDelay(redeliveryDelay); } + + if ( connection.isNonBlockingRedelivery() == false) { + LOG.debug("Blocking session until re-delivery..."); + executor.stop(); + } + connection.getScheduler().executeAfterDelay(new Runnable() { @Override public void run() { - ((ActiveMQDispatcher)md.getConsumer()).dispatch(md); + + if (connection.isNonBlockingRedelivery()) { + ((ActiveMQDispatcher)md.getConsumer()).dispatch(md); + } else { + LOG.debug("Session released, issuing re-delivery..."); + executor.executeFirst(md); + executor.start(); + } } }, redeliveryDelay); } @@ -1016,6 +1029,12 @@ public class ActiveMQSession implements Session, QueueSession, TopicSession, Sta if (deliveryListener != null) { deliveryListener.afterDelivery(this, message); } + + try { + executor.waitForQueueRestart(); + } catch (InterruptedException ex) { + connection.onClientInternalException(ex); + } } } diff --git a/activemq-client/src/main/java/org/apache/activemq/ActiveMQSessionExecutor.java b/activemq-client/src/main/java/org/apache/activemq/ActiveMQSessionExecutor.java index caa1ca9884..357815561c 100755 --- a/activemq-client/src/main/java/org/apache/activemq/ActiveMQSessionExecutor.java +++ b/activemq-client/src/main/java/org/apache/activemq/ActiveMQSessionExecutor.java @@ -207,4 +207,15 @@ public class ActiveMQSessionExecutor implements Task { List getUnconsumedMessages() { return messageQueue.removeAll(); } + + void waitForQueueRestart() throws InterruptedException { + synchronized (messageQueue.getMutex()) { + while (messageQueue.isRunning() == false) { + if (messageQueue.isClosed()) { + break; + } + messageQueue.getMutex().wait(); + } + } + } }