JAVA-660: Took call to OutMessage.doneWithMessage out of the recursively-called method, to avoid having it called more than once. Protected OutMessage by setting the buffer to null and checking for null everywhere it's used.

Author: jyemin

src/main/com/mongodb/DBTCPConnector.java

@@ -252,64 +252,67 @@ public Response call( DB db , DBCollection coll , OutMessage m , ServerAddress h
      */
     @Override
     public Response call( DB db, DBCollection coll, OutMessage m, ServerAddress hostNeeded, int retries, ReadPreference readPref, DBDecoder decoder ){
-
         try {
-            if (readPref == null)
-                readPref = ReadPreference.primary();
+            return innerCall(db, coll, m, hostNeeded, retries, readPref, decoder);
+        } finally {
+            m.doneWithMessage();
+        }
+    }
 
-            if (readPref == ReadPreference.primary() && m.hasOption( Bytes.QUERYOPTION_SLAVEOK ))
-               readPref = ReadPreference.secondaryPreferred();
+    private Response innerCall(final DB db, final DBCollection coll, final OutMessage m, final ServerAddress hostNeeded, final int retries, ReadPreference readPref, final DBDecoder decoder) {
+        if (readPref == null)
+            readPref = ReadPreference.primary();
 
-            boolean secondaryOk = !(readPref == ReadPreference.primary());
+        if (readPref == ReadPreference.primary() && m.hasOption( Bytes.QUERYOPTION_SLAVEOK ))
+           readPref = ReadPreference.secondaryPreferred();
 
-            _checkClosed();
-            if (!secondaryOk)
-                checkMaster( false, !secondaryOk );
+        boolean secondaryOk = !(readPref == ReadPreference.primary());
 
-            final MyPort mp = _myPort.get();
-            final DBPort port = mp.get( false , readPref, hostNeeded );
+        _checkClosed();
+        if (!secondaryOk)
+            checkMaster( false, !secondaryOk );
 
-            Response res = null;
-            boolean retry = false;
-            try {
-                port.checkAuth( db );
-                res = port.call( m , coll, decoder );
-                if ( res._responseTo != m.getId() )
-                    throw new MongoException( "ids don't match" );
-            }
-            catch ( IOException ioe ){
-                mp.error( port , ioe );
-                retry = retries > 0 && !coll._name.equals( "$cmd" )
-                        && !(ioe instanceof SocketTimeoutException) && _error( ioe, secondaryOk );
-                if ( !retry ){
-                    throw new MongoException.Network( "can't call something : " + port.host() + "/" + db,
-                                                      ioe );
-                }
-            }
-            catch ( RuntimeException re ){
-                mp.error( port , re );
-                throw re;
-            } finally {
-                mp.done( port );
+        final MyPort mp = _myPort.get();
+        final DBPort port = mp.get( false , readPref, hostNeeded );
+
+        Response res = null;
+        boolean retry = false;
+        try {
+            port.checkAuth( db );
+            res = port.call( m , coll, decoder );
+            if ( res._responseTo != m.getId() )
+                throw new MongoException( "ids don't match" );
+        }
+        catch ( IOException ioe ){
+            mp.error( port , ioe );
+            retry = retries > 0 && !coll._name.equals( "$cmd" )
+                    && !(ioe instanceof SocketTimeoutException) && _error( ioe, secondaryOk );
+            if ( !retry ){
+                throw new MongoException.Network( "can't call something : " + port.host() + "/" + db,
+                                                  ioe );
             }
+        }
+        catch ( RuntimeException re ){
+            mp.error( port , re );
+            throw re;
+        } finally {
+            mp.done( port );
+        }
 
-            if (retry)
-                return call( db , coll , m , hostNeeded , retries - 1 , readPref, decoder );
+        if (retry)
+            return innerCall( db , coll , m , hostNeeded , retries - 1 , readPref, decoder );
 
-            ServerError err = res.getError();
+        ServerError err = res.getError();
 
-            if ( err != null && err.isNotMasterError() ){
-                checkMaster( true , true );
-                if ( retries <= 0 ){
-                    throw new MongoException( "not talking to master and retries used up" );
-                }
-                return call( db , coll , m , hostNeeded , retries -1, readPref, decoder );
+        if ( err != null && err.isNotMasterError() ){
+            checkMaster( true , true );
+            if ( retries <= 0 ){
+                throw new MongoException( "not talking to master and retries used up" );
             }
-
-            return res;
-        } finally {
-            m.doneWithMessage();
+            return innerCall( db , coll , m , hostNeeded , retries -1, readPref, decoder );
         }
+
+        return res;
     }
 
     public ServerAddress getAddress(){

src/main/com/mongodb/OutMessage.java

@@ -221,20 +221,38 @@ private void writeMessagePrologue(final OpCode opCode) {
     }
 
     void prepare(){
+        if (_buffer == null) {
+            throw new IllegalStateException("Already closed");
+        }
+
         _buffer.writeInt( 0 , _buffer.size() );
     }
 
     void pipe( OutputStream out ) throws IOException {
+        if (_buffer == null) {
+            throw new IllegalStateException("Already closed");
+        }
+
         _buffer.pipe( out );
     }
 
-    int size(){
+    int size() {
+        if (_buffer == null) {
+            throw new IllegalStateException("Already closed");
+        }
+
         return _buffer.size();
     }
 
-    void doneWithMessage(){
+    void doneWithMessage() {
+        if (_buffer == null) {
+            throw new IllegalStateException("Only call this once per instance");
+        }
+
         _buffer.reset();
         _mongo._bufferPool.done(_buffer);
+        _buffer = null;
+        done();
     }
 
     boolean hasOption( int option ){
@@ -263,6 +281,10 @@ int getNumDocuments() {
 
     @Override
     public int putObject(BSONObject o) {
+        if (_buffer == null) {
+            throw new IllegalStateException("Already closed");
+        }
+
         // check max size
         int objectSize = _encoder.writeObject(_buf, o);
         if (objectSize > Math.max(_mongo.getConnector().getMaxBsonObjectSize(), Bytes.MAX_OBJECT_SIZE)) {
@@ -274,7 +296,7 @@ public int putObject(BSONObject o) {
 
     private final Mongo _mongo;
     private final DBCollection _collection;
-    private final PoolOutputBuffer _buffer;
+    private PoolOutputBuffer _buffer;
     private final int _id;
     private final OpCode _opCode;
     private final int _queryOptions;