JAVA-613: Made OutMessage mostly immutable (except for the buffer) so that it can safely be examined via JMX

Author: jyemin

src/main/com/mongodb/OutMessage.java

@@ -1,5 +1,3 @@
-// OutMessage.java
-
 /**
  *      Copyright (C) 2008 10gen Inc.
  *
@@ -28,10 +26,6 @@
 import java.util.Collection;
 import java.util.concurrent.atomic.AtomicInteger;
 
-import java.io.IOException;
-import java.io.OutputStream;
-import java.util.concurrent.atomic.AtomicInteger;
-
 class OutMessage extends BasicBSONEncoder {
 
     enum OpCode {
@@ -57,58 +51,26 @@ public int getValue() {
 
     public static OutMessage insert(final DBCollection collection, final DBEncoder encoder, WriteConcern concern) {
         OutMessage om = new OutMessage(collection, OpCode.OP_INSERT, encoder);
-
-        int flags = 0;
-        if (concern.getContinueOnErrorForInsert()) {
-            flags |= 1;
-        }
-        om.writeInt( flags );
-        om.writeCString( collection.getFullName() );
+        om.writeInsertPrologue(concern);
 
         return om;
     }
 
     public static OutMessage update(final DBCollection collection, final DBEncoder encoder,
                               final boolean upsert, final boolean multi, final DBObject query, final DBObject o) {
-        OutMessage om = new OutMessage(collection, OpCode.OP_UPDATE, encoder);
-
-        om.writeInt( 0 ); // reserved
-        om.writeCString(collection.getFullName());
-
-        int flags = 0;
-        if ( upsert ) flags |= 1;
-        if ( multi ) flags |= 2;
-        om.writeInt( flags );
-
-        om.putObject( query );
-        om.putObject( o );
-
-        om._query = query;
+        OutMessage om = new OutMessage(collection, OpCode.OP_UPDATE, encoder, query);
+        om.writeUpdate(upsert, multi, query, o);
 
         return om;
     }
 
     public static OutMessage remove(final DBCollection collection, final DBEncoder encoder, final DBObject query) {
-        OutMessage om = new OutMessage(collection, OpCode.OP_DELETE, encoder);
-
-        om.writeInt( 0 ); // reserved
-        om.writeCString( collection.getFullName() );
-
-        Collection<String> keys = query.keySet();
-
-        if ( keys.size() == 1 && keys.iterator().next().equals( "_id" ) && query.get( keys.iterator().next() ) instanceof ObjectId)
-            om.writeInt( 1 );
-        else
-            om.writeInt( 0 );
-
-        om.putObject( query );
-
-        om._query = query;
+        OutMessage om = new OutMessage(collection, OpCode.OP_DELETE, encoder, query);
+        om.writeRemove();
 
         return om;
     }
 
-
     static OutMessage query( DBCollection collection , int options , int numToSkip , int batchSize , DBObject query , DBObject fields ){
         return query( collection, options, numToSkip, batchSize, query, fields, ReadPreference.primary() );
     }
@@ -118,95 +80,140 @@ static OutMessage query( DBCollection collection , int options , int numToSkip ,
     }
 
     static OutMessage query( DBCollection collection , int options , int numToSkip , int batchSize , DBObject query , DBObject fields, ReadPreference readPref, DBEncoder enc ){
-        OutMessage out = new OutMessage(collection, OpCode.OP_QUERY, enc);
+        OutMessage om =  new OutMessage(collection, enc, query, options, readPref);
+        om.writeQuery(fields, numToSkip, batchSize);
 
-        out._appendQuery(options, collection, numToSkip, batchSize, query, fields, readPref);
-
-        return out;
+        return om;
     }
 
     static OutMessage getMore(DBCollection collection, long cursorId, int batchSize) {
         OutMessage om = new OutMessage(collection, OpCode.OP_GETMORE);
-
-        om.writeInt(0);
-        om.writeCString(collection.getFullName());
-        om.writeInt(batchSize);
-        om.writeLong(cursorId);
+        om.writeGetMore(cursorId, batchSize);
 
         return om;
     }
 
     static OutMessage killCursors(Mongo mongo, int numCursors) {
         OutMessage om = new OutMessage(mongo , OpCode.OP_KILL_CURSORS);
-
-        om.writeInt(0); // reserved
-        om.writeInt(numCursors);
-
+        om.writeKillCursorsPrologue(numCursors);
 
         return om;
     }
 
-    OutMessage(final DBCollection collection, final OpCode opQuery, final DBEncoder enc) {
-        this(collection.getDB().getMongo(), opQuery, enc);
-        this._collection = collection;
+    private OutMessage( Mongo m , OpCode opCode ){
+        this(null, m, opCode, null);
     }
 
-    OutMessage(final DBCollection collection, final OpCode opQuery) {
-        this(collection.getDB().getMongo(), opQuery);
-        this._collection = collection;
+    private OutMessage(final DBCollection collection, final OpCode opCode) {
+        this(collection, opCode, null);
     }
 
+    private OutMessage(final DBCollection collection, final OpCode opCode, final DBEncoder enc) {
+        this(collection, collection.getDB().getMongo(), opCode, enc);
+    }
 
-    OutMessage( Mongo m , OpCode opCode ){
-        this( m, DefaultDBEncoder.FACTORY.create() );
-        reset(opCode);
+    private OutMessage(final DBCollection collection, final Mongo m, final OpCode opCode, final DBEncoder enc) {
+        this(collection, m, opCode, enc, null, -1, null);
     }
 
-    OutMessage( Mongo m , OpCode opCode , DBEncoder enc ) {
-        this( m , enc );
-        reset( opCode );
+    private OutMessage(final DBCollection collection, final OpCode opCode, final DBEncoder enc, final DBObject query) {
+        this(collection, collection.getDB().getMongo(), opCode, enc, query, 0, null);
     }
 
-    private OutMessage( Mongo m , DBEncoder encoder ) {
-        _encoder = encoder;
+    private OutMessage(final DBCollection collection, final DBEncoder enc, final DBObject query, final int options, final ReadPreference readPref) {
+        this(collection, collection.getDB().getMongo(), OpCode.OP_QUERY, enc, query, options, readPref);
+    }
+
+    private OutMessage(final DBCollection collection, final Mongo m, OpCode opCode, final DBEncoder enc, final DBObject query, final int options, final ReadPreference readPref) {
+        _collection = collection;
         _mongo = m;
-        _buffer = _mongo == null ? new PoolOutputBuffer() : _mongo._bufferPool.get();
+        _encoder = enc;
+
+        _buffer = _mongo._bufferPool.get();
         _buffer.reset();
+        set(_buffer);
 
-        set( _buffer );
-    }
+        _id = REQUEST_ID.getAndIncrement();
+        _opCode = opCode;
+
+        writeMessagePrologue(opCode);
 
-    private void _appendQuery( int options , DBCollection collection , int numToSkip , int batchSize , DBObject query , DBObject fields, ReadPreference readPref){
-        _queryOptions = options;
-        _readPref = readPref;
+        if (query == null) {
+            _query = null;
+            _queryOptions = 0;
+        } else {
+            _query = query;
 
-        //If the readPrefs are non-null and non-primary, set slaveOk query option
-        if (_readPref != null && _readPref.isSlaveOk()) {
-            _queryOptions |= Bytes.QUERYOPTION_SLAVEOK;
+            int allOptions = options;
+            if (readPref != null && readPref.isSlaveOk()) {
+                allOptions |= Bytes.QUERYOPTION_SLAVEOK;
+            }
+
+            _queryOptions = allOptions;
         }
+    }
+
+    private void writeInsertPrologue(final WriteConcern concern) {
+        int flags = 0;
+        if (concern.getContinueOnErrorForInsert()) {
+            flags |= 1;
+        }
+        writeInt(flags);
+        writeCString(_collection.getFullName());
+    }
 
-        writeInt( _queryOptions );
-        writeCString( collection.getFullName() );
+    private void writeUpdate(final boolean upsert, final boolean multi, final DBObject query, final DBObject o) {
+        writeInt(0); // reserved
+        writeCString(_collection.getFullName());
 
-        writeInt( numToSkip );
-        writeInt( batchSize );
+        int flags = 0;
+        if ( upsert ) flags |= 1;
+        if ( multi ) flags |= 2;
+        writeInt(flags);
+
+        putObject(query);
+        putObject(o);
+    }
 
-        putObject( query );
-        if ( fields != null )
-            putObject( fields );
+    private void writeRemove() {
+        writeInt(0); // reserved
+        writeCString(_collection.getFullName());
 
-        this._query = query;
+        Collection<String> keys = _query.keySet();
 
+        if ( keys.size() == 1 && keys.iterator().next().equals( "_id" ) && _query.get( keys.iterator().next() ) instanceof ObjectId)
+            writeInt( 1 );
+        else
+            writeInt( 0 );
+
+        putObject(_query);
     }
 
-    private void reset( OpCode opCode ){
-        done();
-        _buffer.reset();
-        set( _buffer );
+    private void writeGetMore(final long cursorId, final int batchSize) {
+        writeInt(0);
+        writeCString(_collection.getFullName());
+        writeInt(batchSize);
+        writeLong(cursorId);
+    }
 
-        _id = REQUEST_ID.getAndIncrement();
-        _opCode = opCode;
+    private void writeKillCursorsPrologue(final int numCursors) {
+        writeInt(0); // reserved
+        writeInt(numCursors);
+    }
+
+    private void writeQuery(final DBObject fields, final int numToSkip, final int batchSize) {
+        writeInt(_queryOptions);
+        writeCString(_collection.getFullName());
+
+        writeInt(numToSkip);
+        writeInt(batchSize);
 
+        putObject(_query);
+        if (fields != null)
+            putObject(fields);
+    }
+
+    private void writeMessagePrologue(final OpCode opCode) {
         writeInt( 0 ); // length: will set this later
         writeInt( _id );
         writeInt( 0 ); // response to
@@ -217,28 +224,17 @@ void prepare(){
         _buffer.writeInt( 0 , _buffer.size() );
     }
 
-
-    void pipe( OutputStream out )
-        throws IOException {
+    void pipe( OutputStream out ) throws IOException {
         _buffer.pipe( out );
     }
 
     int size(){
         return _buffer.size();
     }
 
-    byte[] toByteArray(){
-        return _buffer.toByteArray();
-    }
-
     void doneWithMessage(){
-        if ( _buffer != null && _mongo != null ) {
-            _buffer.reset();
-            _mongo._bufferPool.done( _buffer );
-        }
-
-        _buffer = null;
-        _mongo = null;
+        _buffer.reset();
+        _mongo._bufferPool.done(_buffer);
     }
 
     boolean hasOption( int option ){
@@ -264,30 +260,19 @@ String getNamespace() {
     @Override
     public int putObject(BSONObject o) {
         // check max size
-        int sz = _encoder.writeObject(_buf, o);
-        if (_mongo != null) {
-            int maxsize = _mongo.getConnector().getMaxBsonObjectSize();
-            maxsize = Math.max(maxsize, Bytes.MAX_OBJECT_SIZE);
-            if (sz > maxsize) {
-                throw new MongoInternalException("DBObject of size " + sz + " is over Max BSON size " + _mongo.getMaxBsonObjectSize());
-            }
+        int objectSize = _encoder.writeObject(_buf, o);
+        if (objectSize > Math.max(_mongo.getConnector().getMaxBsonObjectSize(), Bytes.MAX_OBJECT_SIZE)) {
+            throw new MongoInternalException("DBObject of size " + objectSize + " is over Max BSON size " + _mongo.getMaxBsonObjectSize());
         }
-        return sz;
+        return objectSize;
     }
 
-
-    public ReadPreference getReadPreference(){
-        return _readPref;
-    }
-
-    private Mongo _mongo;
-    private PoolOutputBuffer _buffer;
-    private int _id;
-    private OpCode _opCode;
-    private DBCollection _collection;
-    private int _queryOptions = 0;
-    private DBObject _query;
-    private ReadPreference _readPref = ReadPreference.primary();
-    private DBEncoder _encoder;
-
+    private final Mongo _mongo;
+    private final DBCollection _collection;
+    private final PoolOutputBuffer _buffer;
+    private final int _id;
+    private final OpCode _opCode;
+    private final int _queryOptions;
+    private final DBObject _query;
+    private final DBEncoder _encoder;
 }