Merge branch '4.0.x-stable'

Author: acogoluegnes

src/main/java/com/rabbitmq/utility/BlockingCell.java

@@ -63,12 +63,13 @@ public synchronized T get() throws InterruptedException {
     public synchronized T get(long timeout) throws InterruptedException, TimeoutException {
         if (timeout == INFINITY) return get();
 
-        if (timeout < 0)
+        if (timeout < 0) {
             throw new AssertionError("Timeout cannot be less than zero");
+        }
 
-        long maxTime = System.currentTimeMillis() + timeout;
-        long now;
-        while (!_filled && (now = System.currentTimeMillis()) < maxTime) {
+        long now = System.nanoTime() / NANOS_IN_MILLI;
+        long maxTime = now + timeout;
+        while (!_filled && (now = (System.nanoTime() / NANOS_IN_MILLI)) < maxTime) {
             wait(maxTime - now);
         }
 
@@ -83,11 +84,19 @@ public synchronized T get(long timeout) throws InterruptedException, TimeoutExce
      * @return the waited-for value
      */
     public synchronized T uninterruptibleGet() {
-        while (true) {
-            try {
-                return get();
-            } catch (InterruptedException ex) {
-                // no special handling necessary
+        boolean wasInterrupted = false;
+        try {
+            while (true) {
+                try {
+                    return get();
+                } catch (InterruptedException ex) {
+                    // no special handling necessary
+                    wasInterrupted = true;
+                }
+            }
+        } finally {
+            if (wasInterrupted) {
+                Thread.currentThread().interrupt();
             }
         }
     }
@@ -104,14 +113,21 @@ public synchronized T uninterruptibleGet() {
     public synchronized T uninterruptibleGet(int timeout) throws TimeoutException {
         long now = System.nanoTime() / NANOS_IN_MILLI;
         long runTime = now + timeout;
-
-        do {
-            try {
-                return get(runTime - now);
-            } catch (InterruptedException e) {
-                // Ignore.
+        boolean wasInterrupted = false;
+        try {
+            do {
+                try {
+                    return get(runTime - now);
+                } catch (InterruptedException e) {
+                    // Ignore.
+                    wasInterrupted = true;
+                }
+            } while ((timeout == INFINITY) || ((now = System.nanoTime() / NANOS_IN_MILLI) < runTime));
+        } finally {
+            if (wasInterrupted) {
+                Thread.currentThread().interrupt();
             }
-        } while ((timeout == INFINITY) || ((now = System.nanoTime() / NANOS_IN_MILLI) < runTime));
+        }
 
         throw new TimeoutException();
     }

src/main/java/com/rabbitmq/utility/SingleShotLinearTimer.java

@@ -31,7 +31,10 @@
  * package-private.
  * 
  * We currently just use this to time the quiescing RPC in AMQChannel.
- * 
+ *
+ * Will be removed in next major release.
+ *
+ * @deprecated
  * @see AMQChannel
  */
 
@@ -72,21 +75,27 @@ public TimerThread(long timeoutMillisec) {
         public void run() {
             try {
                 long now;
-                while ((now = System.nanoTime() / NANOS_IN_MILLI) < _runTime) {
-                    if (_task == null) break;
+                boolean wasInterrupted = false;
+                try {
+                    while ((now = System.nanoTime() / NANOS_IN_MILLI) < _runTime) {
+                        if (_task == null) break;
 
-                    try {
-                        synchronized(this) {
-                            wait(_runTime - now);
+                        try {
+                            synchronized(this) {
+                                wait(_runTime - now);
+                            }
+                        } catch (InterruptedException e) {
+                            wasInterrupted = true;
                         }
-                    } catch (InterruptedException e) {
-                        Thread.currentThread().interrupt();
                     }
+                } finally {
+                    if (wasInterrupted)
+                        Thread.currentThread().interrupt();
                 }
 
                 Runnable task = _task;
                 if (task != null) {
-                    task.run();                    
+                    task.run();
                 }
 
             } finally {