QPID-3951 : [Java AMQP 1.0] Correctly set default outcome on the broker

git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@1326997 13f79535-47bb-0310-9956-ffa450edef68

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();
+                        }
                     }
                 });
             }