There is an issue when we do client side selectors for sync receive.

when a message is released (bcos it didn't match the selector) we need to request more credit as in credit mode, a release doesn't automatically increase credit.


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

1 files changed, 22 insertions, 1 deletions

diff --git a/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_10.java b/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_10.java
index 8761a08317..af259080a3 100644
--- a/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_10.java
+++ b/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_10.java
@@ -133,7 +133,6 @@ public class BasicMessageConsumer_0_10 extends BasicMessageConsumer<Struct[], By
         }
     }
 
-
     public void onMessage(Message message)
     {
         int channelId = getSession().getChannelId();
@@ -272,6 +271,7 @@ public class BasicMessageConsumer_0_10 extends BasicMessageConsumer<Struct[], By
                 _logger.debug("filterMessage - trying to ack message");
             }
             acknowledgeMessage(message);
+            requestCreditIfCreditMode();
         }
         else if (!messageOk)
         {
@@ -280,6 +280,7 @@ public class BasicMessageConsumer_0_10 extends BasicMessageConsumer<Struct[], By
                 _logger.debug("Message not OK, releasing");
             }
             releaseMessage(message);
+            requestCreditIfCreditMode();
         }
         // now we need to acquire this message if needed
         // this is the case of queue with a message selector set
@@ -294,6 +295,26 @@ public class BasicMessageConsumer_0_10 extends BasicMessageConsumer<Struct[], By
         return messageOk;
     }
 
+    private void requestCreditIfCreditMode()
+    {
+        try
+        {
+            // the current message received is not good, so we need to get a message.
+            if (getMessageListener() == null)
+            {
+                _0_10session.getQpidSession().messageFlow(getConsumerTag().toString(),
+                    org.apache.qpidity.nclient.Session.MESSAGE_FLOW_UNIT_MESSAGE,1);
+                _0_10session.getQpidSession().messageFlush(getConsumerTag().toString());
+                _0_10session.getQpidSession().sync();
+                _0_10session.getQpidSession().messageFlow(getConsumerTag().toString(),Session.MESSAGE_FLOW_UNIT_BYTE, 0xFFFFFFFF);
+            }
+        }
+        catch(Exception e)
+        {
+            _logger.error("Error getting message listener, couldn't request credit after releasing a message that failed the selector test",e);
+        }
+    }
+
     /**
      * Acknowledge a message
      *