JAVA-635: Added MongoInterruptedException, a runtime exception that wraps an InterruptedException. An instance of this class is thrown in cases where the driver is sleeping or waiting on a condition, and has to catch InterruptedException and do something besides swallowing it. Since InterruptedException is checked and so can't be thrown from methods that don't declare it, best thing to do it throw a runtime exception that can be handled in the application

Author: jyemin

src/main/com/mongodb/DBApiLayer.java

@@ -411,7 +411,7 @@ public boolean hasNext(){
                             try {
                                 Thread.sleep(500);
                             } catch (InterruptedException e) {
-                                // ignore
+                                throw new MongoInterruptedException(e);
                             }
                         }
                     }

src/main/com/mongodb/DBPortPool.java

@@ -23,6 +23,7 @@
 import com.mongodb.util.management.JMException;
 import com.mongodb.util.management.MBeanServerFactory;
 
+import java.io.InterruptedIOException;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
@@ -193,8 +194,9 @@ public DBPort get() {
 
         try {
             port = get( _options.maxWaitTime );
-        }
-        finally {
+        } catch (InterruptedException e) {
+            throw new MongoInterruptedException(e);
+        } finally {
             _waitingSem.release();
         }
 
@@ -207,15 +209,14 @@ public DBPort get() {
 
     // return true if the exception is recoverable
     boolean gotError( Exception e ){
-        if ( e instanceof java.nio.channels.ClosedByInterruptException ||
-             e instanceof InterruptedException ){
+        if (e instanceof java.nio.channels.ClosedByInterruptException){
             // this is probably a request that is taking too long
             // so usually doesn't mean there is a real db problem
             return true;
         }
 
-        if ( e instanceof java.net.SocketTimeoutException ){
-            // we don't want to clear the port pool for a connection timing out
+        if ( e instanceof InterruptedIOException){
+            // we don't want to clear the port pool for a connection timing out or interrupted
             return true;
         }
         Bytes.LOGGER.log( Level.WARNING , "emptying DBPortPool to " + getServerAddress() + " b/c of error" , e );
@@ -224,10 +225,14 @@ boolean gotError( Exception e ){
 
         List<DBPort> all = new ArrayList<DBPort>();
         while ( true ){
-            DBPort temp = get(0);
-            if ( temp == null )
-                break;
-            all.add( temp );
+            try {
+                DBPort temp = get(0);
+                if ( temp == null )
+                    break;
+                all.add( temp );
+            } catch (InterruptedException interruptedException) {
+                throw new MongoInterruptedException(interruptedException);
+            }
         }
 
         for ( DBPort p : all ){

src/main/com/mongodb/DynamicConnectionStatus.java

@@ -183,7 +183,7 @@ private synchronized ConnectionStatus getConnectionStatus() {
             try {
                 wait(_mongoOptions.connectTimeout);
             } catch (InterruptedException e) {
-                throw new MongoException("Interrupted while waiting for next update to dynamic status", e);
+                throw new MongoInterruptedException("Interrupted while waiting for next update to dynamic status", e);
             }
         }
         return connectionStatus;

src/main/com/mongodb/MongoInterruptedException.java

@@ -0,0 +1,34 @@
+/**
+ * Copyright (c) 2008 - 2012 10gen, Inc. <http://10gen.com>
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package com.mongodb;
+
+/**
+ * A non-checked exception indicating that the driver has been interrupted by a call to Thread.interrupt.
+ *
+ * @see Thread#interrupt()
+ * @see InterruptedException
+ */
+public class MongoInterruptedException extends MongoException {
+    public MongoInterruptedException(final InterruptedException e) {
+        super("A driver operation has been interrupted", e);
+    }
+
+    public MongoInterruptedException(final String message, final InterruptedException e) {
+        super(message, e);
+    }
+}

src/main/com/mongodb/MongosStatus.java

@@ -122,7 +122,7 @@ private synchronized Node getPreferred() {
                     wait();
                 }
             } catch (InterruptedException e) {
-                throw new MongoException("Interrupted while waiting for next update to mongos status", e);
+                throw new MongoInterruptedException("Interrupted while waiting for next update to mongos status", e);
             }
         }
         return preferred;

src/main/com/mongodb/ReplicaSetStatus.java

@@ -134,7 +134,7 @@ synchronized ReplicaSet get() {
                    wait(ReplicaSetStatus.mongoOptionsDefaults.socketTimeout);
                }
                catch (InterruptedException e) {
-                   throw new MongoException("Interrupted while waiting for next update to replica set status", e);
+                   throw new MongoInterruptedException("Interrupted while waiting for next update to replica set status", e);
                }
            }
            return members;
@@ -156,7 +156,7 @@ synchronized void waitForNextUpdate() {
                wait(ReplicaSetStatus.mongoOptionsDefaults.socketTimeout);
            }
            catch (InterruptedException e) {
-              throw new MongoException("Interrupted while waiting for next update to replica set status", e);
+              throw new MongoInterruptedException("Interrupted while waiting for next update to replica set status", e);
            }
        }
 

src/main/com/mongodb/util/SimplePool.java

@@ -95,7 +95,7 @@ public void remove( T t ) {
     /** Gets an object from the pool - will block if none are available
      * @return An object from the pool
      */
-    public T get() {
+    public T get() throws InterruptedException {
 	return get(-1);
     }
 
@@ -106,7 +106,7 @@ public T get() {
      *        positive ms to wait
      * @return An object from the pool, or null if can't get one in the given waitTime
      */
-    public T get(long waitTime) {
+    public T get(long waitTime) throws InterruptedException {
         if (!permitAcquired(waitTime)) {
             return null;
         }
@@ -143,18 +143,14 @@ private T createNewAndReleasePermitIfFailure() {
         }
     }
 
-    private boolean permitAcquired(final long waitTime) {
-        try {
-            if (waitTime > 0) {
-                return _sem.tryAcquire(waitTime, TimeUnit.MILLISECONDS);
-            } else if (waitTime < 0) {
-                _sem.acquire();
-                return true;
-            } else {
-                return _sem.tryAcquire();
-            }
-        } catch (InterruptedException e) {
-            return false;
+    private boolean permitAcquired(final long waitTime) throws InterruptedException {
+        if (waitTime > 0) {
+            return _sem.tryAcquire(waitTime, TimeUnit.MILLISECONDS);
+        } else if (waitTime < 0) {
+            _sem.acquire();
+            return true;
+        } else {
+            return _sem.tryAcquire();
         }
     }
 

src/test/com/mongodb/util/SimplePoolTest.java

@@ -16,6 +16,12 @@
 
 package com.mongodb.util;
 
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+
 public class SimplePoolTest extends com.mongodb.util.TestCase {
 
     class MyPool extends SimplePool<Integer> {
@@ -41,7 +47,7 @@ public Integer createNew(){
     }
 
     @org.testng.annotations.Test
-    public void testBasic1(){
+    public void testBasic1() throws InterruptedException {
 	MyPool p = new MyPool( 10 );
 
 	int a = p.get();
@@ -56,7 +62,7 @@ public void testBasic1(){
     }
 
     @org.testng.annotations.Test
-    public void testMax1(){
+    public void testMax1() throws InterruptedException {
 	MyPool p = new MyPool( 10 );
 
 	int a = p.get();
@@ -70,7 +76,7 @@ public void testMax1(){
     }
 
     @org.testng.annotations.Test
-    public void testMax2(){
+    public void testMax2() throws InterruptedException {
 	MyPool p = new MyPool( 10 );
 
 	int a = p.get();
@@ -83,7 +89,7 @@ public void testMax2(){
     }
 
     @org.testng.annotations.Test
-    public void testMax3(){
+    public void testMax3() throws InterruptedException {
 	MyPool p = new MyPool( 10  );
 
 	int a = p.get();
@@ -96,7 +102,7 @@ public void testMax3(){
     }
 
     @org.testng.annotations.Test
-    public void testThrowErrorFromCreate(){
+    public void testThrowErrorFromCreate() throws InterruptedException {
         MyPool p = new MyPool( 1 );
         p._throwError = true;
 
@@ -115,7 +121,7 @@ public void testThrowErrorFromCreate(){
     }
 
     @org.testng.annotations.Test
-    public void testCouldCreate() {
+    public void testCouldCreate() throws InterruptedException {
         SimplePool<Integer> p = new SimplePool<Integer>("pool", 2) {
             @Override
             protected Integer createNew() {
@@ -145,7 +151,7 @@ protected int pick(int recommended, boolean couldCreate) {
     }
 
     @org.testng.annotations.Test
-    public void testReturnNullFromCreate(){
+    public void testReturnNullFromCreate() throws InterruptedException {
         MyPool p = new MyPool( 1 );
         p._returnNull = true;
 
@@ -163,6 +169,41 @@ public void testReturnNullFromCreate(){
         assertEquals( Integer.valueOf(0) , a );
     }
 
+    @org.testng.annotations.Test()
+    public void testThrowsInterruptedException() {
+        final MyPool p = new MyPool(1);
+        try {
+            p.get();
+        } catch (InterruptedException e) {
+            fail("Should not throw InterruptedException here");
+        }
+
+        ExecutorService executor = Executors.newSingleThreadExecutor();
+        Callable<Boolean> callable = new Callable<Boolean>() {
+            @Override
+            public Boolean call() {
+                try {
+                    p.get();
+                    return false;
+                } catch (InterruptedException e) {
+                    // return true if interrupted
+                    return true;
+                }
+            }
+        };
+        Future<Boolean> future = executor.submit(callable);
+
+        // Interrupt the thread
+        executor.shutdownNow();
+
+        try {
+            assertEquals(true, future.get());
+        } catch (InterruptedException e) {
+            fail("Should not happen, since this thread was not interrupted");
+        } catch (ExecutionException e) {
+            fail("Should not happen");
+        }
+    }
 
     public static void main( String args[] ){
 	SimplePoolTest t = new SimplePoolTest();