From b4d6577049bd648fd1407c0283a63d66b5d6bbdf Mon Sep 17 00:00:00 2001
From: Rajith Muditha Attapattu <rajith@apache.org>
Date: Wed, 17 Oct 2007 00:11:58 +0000
Subject: 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@585294 13f79535-47bb-0310-9956-ffa450edef68
---
 .../qpid/client/BasicMessageConsumer_0_10.java     | 23 +++++++++++++++++++++-
 1 file changed, 22 insertions(+), 1 deletion(-)

(limited to 'qpid/java')

diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_10.java b/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_10.java
index 8761a08317..af259080a3 100644
--- a/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_10.java
+++ b/qpid/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
      *
-- 
cgit v1.2.1