Repository: activemq Updated Branches: refs/heads/activemq-5.10.x 22f2f3dde -> e3d218a97
https://issues.apache.org/jira/browse/AMQ-5352 Applied and tested, all tests still passing after this change. Project: http://git-wip-us.apache.org/repos/asf/activemq/repo Commit: http://git-wip-us.apache.org/repos/asf/activemq/commit/d7e65a3c Tree: http://git-wip-us.apache.org/repos/asf/activemq/tree/d7e65a3c Diff: http://git-wip-us.apache.org/repos/asf/activemq/diff/d7e65a3c Branch: refs/heads/activemq-5.10.x Commit: d7e65a3c1962cf29292c38f9e43b6c6b0399e17d Parents: 22f2f3d Author: Timothy Bish <[email protected]> Authored: Tue Sep 9 14:19:39 2014 -0400 Committer: Hadrian Zbarcea <[email protected]> Committed: Wed Dec 17 21:42:50 2014 -0500 ---------------------------------------------------------------------- .../activemq/transport/amqp/AmqpProtocolConverter.java | 13 +++++++++++-- 1 file changed, 11 insertions(+), 2 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/activemq/blob/d7e65a3c/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpProtocolConverter.java ---------------------------------------------------------------------- diff --git a/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpProtocolConverter.java b/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpProtocolConverter.java index df8509d..69bf856 100644 --- a/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpProtocolConverter.java +++ b/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpProtocolConverter.java @@ -610,7 +610,7 @@ class AmqpProtocolConverter implements IAmqpProtocolConverter { LOG.trace("Inbound Message:{} from Producer:{}", message.getMessageId(), producerId + ":" + messageId.getProducerSequenceId()); - DeliveryState remoteState = delivery.getRemoteState(); + final DeliveryState remoteState = delivery.getRemoteState(); if (remoteState != null && remoteState instanceof TransactionalState) { TransactionalState s = (TransactionalState) remoteState; long txid = toLong(s.getTxnId()); @@ -648,7 +648,16 @@ class AmqpProtocolConverter implements IAmqpProtocolConverter { receiver.flow(prefetch - receiver.getCredit()); } - delivery.disposition(Accepted.getInstance()); + if (remoteState != null && remoteState instanceof TransactionalState) { + TransactionalState txAccepted = new TransactionalState(); + txAccepted.setOutcome(Accepted.getInstance()); + txAccepted.setTxnId(((TransactionalState) remoteState).getTxnId()); + + delivery.disposition(txAccepted); + } else { + delivery.disposition(Accepted.getInstance()); + } + delivery.settle(); }
