diff options
Diffstat (limited to 'qpid/java')
| -rw-r--r-- | qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolEngine.java | 16 |
1 files changed, 12 insertions, 4 deletions
diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolEngine.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolEngine.java index cf61e135b0..cc7c74057e 100644 --- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolEngine.java +++ b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolEngine.java @@ -219,13 +219,21 @@ public class AMQProtocolEngine implements ServerProtocolEngine, public void setMessageAssignmentSuspended(final boolean messageAssignmentSuspended) { _messageAssignmentSuspended.set(messageAssignmentSuspended ? Thread.currentThread() : null); - if(!messageAssignmentSuspended) + for(AMQSessionModel<?,?> session : getSessionModels()) { - for(AMQSessionModel<?,?> session : getSessionModels()) + for (Consumer<?> consumer : session.getConsumers()) { - for(Consumer<?> consumer : session.getConsumers()) + ConsumerImpl consumerImpl = (ConsumerImpl) consumer; + if (!messageAssignmentSuspended) { - ((ConsumerImpl)consumer).getTarget().notifyCurrentState(); + consumerImpl.getTarget().notifyCurrentState(); + } + else + { + // ensure that by the time the method returns, no consumer can be in the process of + // delivering a message. + consumerImpl.getSendLock(); + consumerImpl.releaseSendLock(); } } } |