QPID-637 Patch submitted by Aidan Skinner to address receive not waiting for full timeout.

git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/branches/M2.1@583882 13f79535-47bb-0310-9956-ffa450edef68

1 files changed, 31 insertions, 10 deletions

diff --git a/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java b/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java
index 0da55c9857..9c5c344275 100644
--- a/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java
+++ b/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java
@@ -357,28 +357,49 @@ public class BasicMessageConsumer extends Closeable implements MessageConsumer
             Object o = null;
             if (l > 0)
             {
-                o = _synchronousQueue.poll(l, TimeUnit.MILLISECONDS);
+                long endtime = System.currentTimeMillis() + l;
+                while (System.currentTimeMillis() < endtime && o == null)
+                {
+                    try 
+                    {
+                        o = _synchronousQueue.poll(endtime - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
+                    }
+                    catch (InterruptedException e)
+                    {
+                        _logger.warn("Interrupted: " + e);
+                        if (isClosed()) 
+                        {
+                            return null;
+                        }
+                    }
+                }
             }
             else
             {
-                o = _synchronousQueue.take();
+                while (o == null)
+                {
+                    try
+                    {
+                        o = _synchronousQueue.take();
+                    } 
+                    catch (InterruptedException e)
+                    {
+                        _logger.warn("Interrupted: " + e);
+                        if (isClosed()) 
+                        {
+                            return null;
+                        }
+                    }
+                }
             }
-
             final AbstractJMSMessage m = returnMessageOrThrow(o);
             if (m != null)
             {
                 preApplicationProcessing(m);
                 postDeliver(m);
             }
-
             return m;
         }
-        catch (InterruptedException e)
-        {
-            _logger.warn("Interrupted: " + e);
-
-            return null;
-        }
         finally
         {
             releaseReceiving();