Merged revisions 573738-573739,573741-574077,574079-574236,574238-574265,574267-574503,574505-574554,574556-574584,574586-574873,574875-574901,574903-575737,575739-575787,575789-575810,575812-577772,577774-577940,577942-578057,578059-578732,578734,578736-578744,578746-578827,578829-578844,578846-579114,579116-579146,579148-579197,579199-579228,579230-579573,579575-579576,579579-579601,579603-579613,579615-579708,579710-580021,580023-580039,580042-580060,580062-580065,580067-580080,580082-580257,580259-580264,580266-580350,580352-580984,580986-580991,580994-581001,581003-581170,581172-581206,581208-581245,581247-581292,581294-581539,581541-581565,581567-581620,581622-581628 via svnmerge from

https://svn.apache.org/repos/asf/incubator/qpid/branches/M2.1

........
  r581189 | rgodfrey | 2007-10-02 12:08:29 +0100 (Tue, 02 Oct 2007) | 1 line
  
  QPID-614 : Applied patch supplied by Aidan Skinner
........
  r581627 | ritchiem | 2007-10-03 16:26:10 +0100 (Wed, 03 Oct 2007) | 1 line
  
  QPID-614 : Applied supplementary patch from Aidan Skinner.
........


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

1 files changed, 37 insertions, 12 deletions

diff --git a/java/client/src/main/java/org/apache/qpid/client/protocol/BlockingMethodFrameListener.java b/java/client/src/main/java/org/apache/qpid/client/protocol/BlockingMethodFrameListener.java
index 86db9d5859..c64f46ba23 100644
--- a/java/client/src/main/java/org/apache/qpid/client/protocol/BlockingMethodFrameListener.java
+++ b/java/client/src/main/java/org/apache/qpid/client/protocol/BlockingMethodFrameListener.java
@@ -20,6 +20,10 @@
  */
 package org.apache.qpid.client.protocol;
 
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.ReentrantLock;
+
 import org.apache.qpid.AMQException;
 import org.apache.qpid.AMQTimeoutException;
 import org.apache.qpid.client.failover.FailoverException;
@@ -75,7 +79,12 @@ public abstract class BlockingMethodFrameListener implements AMQMethodListener
     private volatile boolean _ready = false;
 
     /** Used to protect the shared event and ready flag between the producer and consumer. */
-    private final Object _lock = new Object();
+    private final ReentrantLock _lock = new ReentrantLock();
+        
+    /**
+     * Used to signal that a method has been received
+     */
+    private final Condition _receivedCondition = _lock.newCondition();
 
     /** Used to hold the most recent exception that is passed to the {@link #error(Exception)} method. */
     private volatile Exception _error;
@@ -126,11 +135,16 @@ public abstract class BlockingMethodFrameListener implements AMQMethodListener
             // we only update the flag from inside the synchronized block
             // so that the blockForFrame method cannot "miss" an update - it
             // will only ever read the flag from within the synchronized block
-            synchronized (_lock)
+            _lock.lock();
+            try
             {
                 _doneEvt = evt;
                 _ready = ready;
-                _lock.notify();
+                _receivedCondition.signal();
+            }
+            finally 
+            {
+                _lock.unlock();
             }
         }
 
@@ -159,27 +173,29 @@ public abstract class BlockingMethodFrameListener implements AMQMethodListener
      */
     public AMQMethodEvent blockForFrame(long timeout) throws AMQException, FailoverException
     {
-        synchronized (_lock)
+        long nanoTimeout = TimeUnit.MILLISECONDS.toNanos(timeout);
+        
+        _lock.lock();
+        try
         {
             while (!_ready)
             {
-                try
-                {
+                try {
                     if (timeout == -1)
                     {
-                        _lock.wait();
+                        _receivedCondition.await();
                     }
                     else
                     {
+                        nanoTimeout = _receivedCondition.awaitNanos(nanoTimeout);
 
-                        _lock.wait(timeout);
-                        if (!_ready)
+                        if (nanoTimeout <= 0 && !_ready && _error == null)
                         {
                             _error = new AMQTimeoutException("Server did not respond in a timely fashion");
                             _ready = true;
                         }
                     }
-                }
+                } 
                 catch (InterruptedException e)
                 {
                     // IGNORE    -- //fixme this isn't ideal as being interrupted isn't equivellant to sucess
@@ -191,6 +207,10 @@ public abstract class BlockingMethodFrameListener implements AMQMethodListener
                 }
             }
         }
+        finally
+        {
+            _lock.unlock();
+        }
 
         if (_error != null)
         {
@@ -224,10 +244,15 @@ public abstract class BlockingMethodFrameListener implements AMQMethodListener
         // can pick up the exception and rethrow to the caller
         _error = e;
 
-        synchronized (_lock)
+        _lock.lock();
+        try
         {
             _ready = true;
-            _lock.notify();
+            _receivedCondition.signal();
+        }
+        finally
+        {
+            _lock.unlock();
         }
     }
 }