JAVA-428 - Support new ReadPreference semantics, deprecate SlaveOK

Brendan W. McAdams · Brendan W. McAdams · commit fe574ad80225 · 2011-10-10T12:35:24.000+02:00
* Pass basic readPreference information down to the API layer; doesn't support dealing w/ tags yet just "Primary or Secondary"
* We are not yet implementing the proposed "bucketing" behavior for pings, but using the existing ping selection of secondary code from the existing library
diff --git a/src/main/com/mongodb/DBApiLayer.java b/src/main/com/mongodb/DBApiLayer.java
@@ -298,7 +298,7 @@ Iterator<DBObject> __find( DBObject ref , DBObject fields , int numToSkip , int
 
             if ( willTrace() ) trace( "find: " + _fullNameSpace + " " + JSON.serialize( ref ) );
 
-            OutMessage query = OutMessage.query( _mongo , options , _fullNameSpace , numToSkip , chooseBatchSize(batchSize, limit, 0) , ref , fields );
+            OutMessage query = OutMessage.query( _mongo , options , _fullNameSpace , numToSkip , chooseBatchSize(batchSize, limit, 0) , ref , fields, readPref);
 
             Response res = _connector.call( _db , this , query , null , 2 );
 
diff --git a/src/main/com/mongodb/DBTCPConnector.java b/src/main/com/mongodb/DBTCPConnector.java
@@ -139,7 +139,7 @@ public WriteResult say( DB db , OutMessage m , WriteConcern concern , ServerAddr
         checkMaster( false , true );
 
         MyPort mp = _myPort.get();
-        DBPort port = mp.get( true , false , hostNeeded );
+        DBPort port = mp.get( true , ReadPreference.PRIMARY, hostNeeded );
 
         try {
             port.checkAuth( db );
@@ -188,12 +188,17 @@ public Response call( DB db , DBCollection coll , OutMessage m , ServerAddress h
 
     public Response call( DB db , DBCollection coll , OutMessage m , ServerAddress hostNeeded , int retries )
         throws MongoException {
-        boolean slaveOk = m.hasOption( Bytes.QUERYOPTION_SLAVEOK );
+        ReadPreference readPref = m.getReadPreference();
+        if (readPref == ReadPreference.PRIMARY && m.hasOption( Bytes.QUERYOPTION_SLAVEOK ))
+           readPref = ReadPreference.SECONDARY;
+
+        boolean secondaryOk = !(readPref == ReadPreference.PRIMARY);
+
         _checkClosed();
-        checkMaster( false , !slaveOk );
+        checkMaster( false, !secondaryOk );
         
         final MyPort mp = _myPort.get();
-        final DBPort port = mp.get( false , slaveOk, hostNeeded );
+        final DBPort port = mp.get( false , readPref, hostNeeded );
                 
         Response res = null;
         boolean retry = false;
@@ -206,7 +211,7 @@ public Response call( DB db , DBCollection coll , OutMessage m , ServerAddress h
         catch ( IOException ioe ){
             mp.error( port , ioe );
             retry = retries > 0 && !coll._name.equals( "$cmd" )
-                    && !(ioe instanceof SocketTimeoutException) && _error( ioe, slaveOk );
+                    && !(ioe instanceof SocketTimeoutException) && _error( ioe, secondaryOk );
             if ( !retry ){
                 throw new MongoException.Network( "can't call something : " + port.host() + "/" + db, 
                                                   ioe ); 
@@ -282,11 +287,11 @@ public String getConnectPoint(){
      * This method is called in case of an IOException.
      * It will potentially trigger a checkMaster() to check the status of all servers.
      * @param t the exception thrown
-     * @param slaveOk slaveOk flag
+     * @param secondaryOk secondaryOk flag
      * @return true if the request should be retried, false otherwise
      * @throws MongoException
      */
-    boolean _error( Throwable t, boolean slaveOk )
+    boolean _error( Throwable t, boolean secondaryOk )
         throws MongoException {
         if (_rsStatus == null) {
             // single server, no need to retry
@@ -297,14 +302,14 @@ boolean _error( Throwable t, boolean slaveOk )
         // if no server is up, we wont retry until the updater thread finds one
         // this is to cut down the volume of requests/errors when all servers are down
         if ( _rsStatus.hasServerUp() ){
-            checkMaster( true , !slaveOk );
+            checkMaster( true , !secondaryOk );
         }
         return _rsStatus.hasServerUp();
     }
 
     class MyPort {
 
-        DBPort get( boolean keep , boolean slaveOk , ServerAddress hostNeeded ){
+        DBPort get( boolean keep , ReadPreference readPref, ServerAddress hostNeeded ){
             
             if ( hostNeeded != null ){
                 // asked for a specific host
@@ -325,8 +330,8 @@ DBPort get( boolean keep , boolean slaveOk , ServerAddress hostNeeded ){
                 _requestPort = null;
             }
             
-            if ( slaveOk && _rsStatus != null ){
-                // if slaveOk, try to use a secondary
+            if ( !(readPref == ReadPreference.PRIMARY) && _rsStatus != null ){
+                // if not a primary read set, try to use a secondary
                 ServerAddress slave = _rsStatus.getASecondary();
                 if ( slave != null ){
                     return _portHolder.get( slave ).get();
diff --git a/src/main/com/mongodb/OutMessage.java b/src/main/com/mongodb/OutMessage.java
@@ -26,18 +26,22 @@
 import java.io.OutputStream;
 import java.util.concurrent.atomic.AtomicInteger;
 
-import org.bson.BSONEncoder;
-import org.bson.BSONObject;
+import org.bson.*;
 import org.bson.io.PoolOutputBuffer;
 import org.bson.types.ObjectId;
 
-class OutMessage extends BSONEncoder {
+class OutMessage extends BasicBSONEncoder {
 
     static AtomicInteger ID = new AtomicInteger(1);
-    
+
     static OutMessage query( Mongo m , int options , String ns , int numToSkip , int batchSize , DBObject query , DBObject fields ){
+        return query( m, options, ns, numToSkip, batchSize, query, fields, ReadPreference.PRIMARY );
+    }
+
+    static OutMessage query( Mongo m , int options , String ns , int numToSkip , int batchSize , DBObject query , DBObject fields, ReadPreference readPref ){
         OutMessage out = new OutMessage( m , 2004 );
         out._appendQuery( options , ns , numToSkip , batchSize , query , fields );
+        out.setReadPreference( readPref );
         return out;
     }
 
@@ -198,9 +202,19 @@ public int putObject(BSONObject o) {
         return sz;
     }
 
+
+    public ReadPreference getReadPreference(){
+        return _readPref;
+    }
+
+    public void setReadPreference( ReadPreference readPref ){
+        _readPref = readPref;
+    }
+
     private Mongo _mongo;
     private PoolOutputBuffer _buffer;
     private int _id;
     private int _queryOptions = 0;
+    private ReadPreference _readPref = ReadPreference.PRIMARY;
 
 }
