JAVA-370: in mongo.close(), the sockets used for checking replica servers status are left open

Map Reduce: ability to add extra options, and always use "out" as object to be consistent

Author: agirbal

src/main/com/mongodb/MapReduceCommand.java

@@ -225,57 +225,67 @@ public void setOutputDB(String outputDB) {
         this._outputDB = outputDB;
     }
 
+    
 
     public DBObject toDBObject() {
-        BasicDBObjectBuilder builder = BasicDBObjectBuilder.start();
-
-        builder.add("mapreduce", _input)
-               .add("map", _map)
-               .add("reduce", _reduce)
-               .add("verbose", _verbose);
-
-        if (_outputType == OutputType.REPLACE && _outputDB == null) {
-            builder.add("out", _outputTarget);
-        } else {
-            BasicDBObject out = new BasicDBObject();
-            switch(_outputType) {
-                case INLINE:
-                    out.put("inline", 1);
-                    break;
-                case REPLACE:
-                    out.put("replace", _outputTarget);
-                    break;
-                case MERGE:
-                    out.put("merge", _outputTarget);
-                    break;
-                case REDUCE:
-                    out.put("reduce", _outputTarget);
-                    break;
-            }
-            if (_outputDB != null)
-                out.put("db", _outputDB);
-            builder.add("out", out);
+        BasicDBObject cmd = new BasicDBObject();
+
+        cmd.put("mapreduce", _input);
+        cmd.put("map", _map);
+        cmd.put("reduce", _reduce);
+        cmd.put("verbose", _verbose);
+
+        BasicDBObject out = new BasicDBObject();
+        switch(_outputType) {
+            case INLINE:
+                out.put("inline", 1);
+                break;
+            case REPLACE:
+                out.put("replace", _outputTarget);
+                break;
+            case MERGE:
+                out.put("merge", _outputTarget);
+                break;
+            case REDUCE:
+                out.put("reduce", _outputTarget);
+                break;
         }
+        if (_outputDB != null)
+            out.put("db", _outputDB);
+        cmd.put("out", out);
 
         if (_query != null)
-            builder.add("query", _query);
+            cmd.put("query", _query);
 
         if (_finalize != null) 
-            builder.add( "finalize", _finalize );
+            cmd.put( "finalize", _finalize );
 
         if (_sort != null)
-            builder.add("sort", _sort);
+            cmd.put("sort", _sort);
 
         if (_limit > 0)
-            builder.add("limit", _limit);
+            cmd.put("limit", _limit);
 
         if (_scope != null)
-            builder.add("scope", _scope);
+            cmd.put("scope", _scope);
 
-
-        return builder.get();
+        if (_extra != null) {
+            cmd.putAll(_extra);
+        }
+        
+        return cmd;
     }
 
+    public void addExtraOption(String name, Object value) {
+        if (_extra == null)
+            _extra = new BasicDBObject();
+        _extra.put(name, value);
+    }
+    
+    public DBObject getExtraOptions() {
+        return _extra;
+    }
+    
     public String toString() { 
         return toDBObject().toString();
     }
@@ -292,4 +302,5 @@ public String toString() {
     int _limit;
     Map<String, Object> _scope;
     Boolean _verbose = true;
+    DBObject _extra;
 }

src/main/com/mongodb/MapReduceOutput.java

@@ -9,7 +9,7 @@
 public class MapReduceOutput {
 
     @SuppressWarnings("unchecked")
-    MapReduceOutput( DBCollection from , DBObject cmd, BasicDBObject raw ){
+    public MapReduceOutput( DBCollection from , DBObject cmd, BasicDBObject raw ){
         _raw = raw;
         _cmd = cmd;
 

src/main/com/mongodb/ReplicaSetStatus.java

@@ -249,6 +249,11 @@ public String toString(){
             return buf.toString();
         }
 
+        public void close() {
+            _port.close();
+            _port = null;
+        }
+        
         final ServerAddress _addr;
         final Set<String> _names = Collections.synchronizedSet( new HashSet<String>() );
         DBPort _port; // we have our own port so we can set different socket options and don't have to owrry about the pool
@@ -388,7 +393,12 @@ void printStatus(){
     }
 
     void close(){
-        _closed = true;
+        if (!_closed) {
+            _closed = true;
+            for (int i = 0; i < _all.size(); i++) {
+                _all.get(i).close();
+            }
+        }
     }
 
     /**