diff options
Diffstat (limited to 'qpid/java')
3 files changed, 25 insertions, 8 deletions
diff --git a/qpid/java/amqp-1-0-client/src/main/java/org/apache/qpid/amqp_1_0/client/Receiver.java b/qpid/java/amqp-1-0-client/src/main/java/org/apache/qpid/amqp_1_0/client/Receiver.java index c5b3cd35ab..e7e0fc78b2 100644 --- a/qpid/java/amqp-1-0-client/src/main/java/org/apache/qpid/amqp_1_0/client/Receiver.java +++ b/qpid/java/amqp-1-0-client/src/main/java/org/apache/qpid/amqp_1_0/client/Receiver.java @@ -348,7 +348,10 @@ public class Receiver implements DeliveryStateHandler public void modified(Binary tag)
{
- update(new Modified(), tag, null, null);
+ final Modified outcome = new Modified();
+ outcome.setDeliveryFailed(true);
+
+ update(outcome, tag, null, null);
}
public void acknowledge(final Message m)
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/v1_0/SendingLink_1_0.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/v1_0/SendingLink_1_0.java index edd3bb6248..6c0f2c0c58 100644 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/v1_0/SendingLink_1_0.java +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/v1_0/SendingLink_1_0.java @@ -44,6 +44,7 @@ import org.apache.qpid.amqp_1_0.type.messaging.Accepted; import org.apache.qpid.amqp_1_0.type.messaging.ExactSubjectFilter; import org.apache.qpid.amqp_1_0.type.messaging.Filter; import org.apache.qpid.amqp_1_0.type.messaging.MatchingSubjectFilter; +import org.apache.qpid.amqp_1_0.type.messaging.Modified; import org.apache.qpid.amqp_1_0.type.messaging.NoLocalFilter; import org.apache.qpid.amqp_1_0.type.messaging.Released; import org.apache.qpid.amqp_1_0.type.messaging.Source; @@ -361,7 +362,7 @@ public class SendingLink_1_0 implements SendingLinkListener, Link_1_0, DeliveryS e.printStackTrace(); //TODO } - DeliveryState state = new Released(); + DeliveryState state = new Modified(); for(UnsettledAction action : _unsettledActionMap.values()) { diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/v1_0/Subscription_1_0.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/v1_0/Subscription_1_0.java index 425f63dd90..08926d000d 100644 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/v1_0/Subscription_1_0.java +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/v1_0/Subscription_1_0.java @@ -21,14 +21,12 @@ package org.apache.qpid.server.protocol.v1_0; import java.nio.ByteBuffer; -import java.util.ArrayList; import java.util.List; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicReference; import java.util.concurrent.locks.ReentrantLock; import org.apache.qpid.AMQException; -import org.apache.qpid.amqp_1_0.codec.DescribedTypeConstructorRegistry; import org.apache.qpid.amqp_1_0.codec.ValueHandler; import org.apache.qpid.amqp_1_0.messaging.SectionEncoder; import org.apache.qpid.amqp_1_0.messaging.SectionEncoderImpl; @@ -37,7 +35,6 @@ import org.apache.qpid.amqp_1_0.type.AmqpErrorException; import org.apache.qpid.amqp_1_0.type.Binary; import org.apache.qpid.amqp_1_0.type.DeliveryState; import org.apache.qpid.amqp_1_0.type.Outcome; -import org.apache.qpid.amqp_1_0.type.Section; import org.apache.qpid.amqp_1_0.type.UnsignedInteger; import org.apache.qpid.amqp_1_0.type.codec.AMQPDescribedTypeRegistry; import org.apache.qpid.amqp_1_0.type.messaging.Accepted; @@ -487,7 +484,7 @@ class Subscription_1_0 implements Subscription _queueEntry = queueEntry; } - public boolean process(DeliveryState state, Boolean settled) + public boolean process(DeliveryState state, final Boolean settled) { Binary transactionId = null; @@ -540,6 +537,13 @@ class Subscription_1_0 implements Subscription public void onRollback() { + if(Boolean.TRUE.equals(settled)) + { + final Modified modified = new Modified(); + modified.setDeliveryFailed(true); + _link.getEndpoint().updateDisposition(_deliveryTag, modified, true); + _link.getEndpoint().sendFlowConditional(); + } } }); } @@ -569,13 +573,22 @@ class Subscription_1_0 implements Subscription { _queueEntry.release(); - _queueEntry.incrementDeliveryCount(); + if(Boolean.TRUE.equals(((Modified)outcome).getDeliveryFailed())) + { + _queueEntry.incrementDeliveryCount(); + } _link.getEndpoint().settle(_deliveryTag); } public void onRollback() { - _link.getEndpoint().settle(_deliveryTag); + if(Boolean.TRUE.equals(settled)) + { + final Modified modified = new Modified(); + modified.setDeliveryFailed(true); + _link.getEndpoint().updateDisposition(_deliveryTag, modified, true); + _link.getEndpoint().sendFlowConditional(); + } } }); } |