/*
    * #%L
    * AbstractMongoOperations.java - mongodb-async-driver - Allanbank Consulting, Inc.
    * %%
    * Copyright (C) 2011 - 2014 Allanbank Consulting, Inc.
    * %%
    * 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.
   * #L%
   */
  package com.allanbank.mongodb.client;
  
  import java.util.ArrayList;
  import java.util.Collection;
  import java.util.List;
  
  import com.allanbank.mongodb.AsyncMongoCollection;
  import com.allanbank.mongodb.Callback;
  import com.allanbank.mongodb.Durability;
  import com.allanbank.mongodb.MongoCursorControl;
  import com.allanbank.mongodb.MongoDatabase;
  import com.allanbank.mongodb.MongoDbException;
  import com.allanbank.mongodb.MongoIterator;
  import com.allanbank.mongodb.ReadPreference;
  import com.allanbank.mongodb.StreamCallback;
  import com.allanbank.mongodb.Version;
  import com.allanbank.mongodb.bson.Document;
  import com.allanbank.mongodb.bson.DocumentAssignable;
  import com.allanbank.mongodb.bson.Element;
  import com.allanbank.mongodb.bson.builder.ArrayBuilder;
  import com.allanbank.mongodb.bson.builder.BuilderFactory;
  import com.allanbank.mongodb.bson.builder.DocumentBuilder;
  import com.allanbank.mongodb.bson.impl.EmptyDocument;
  import com.allanbank.mongodb.bson.impl.ImmutableDocument;
  import com.allanbank.mongodb.bson.impl.RootDocument;
  import com.allanbank.mongodb.builder.Aggregate;
  import com.allanbank.mongodb.builder.BatchedWrite;
  import com.allanbank.mongodb.builder.ConditionBuilder;
  import com.allanbank.mongodb.builder.Count;
  import com.allanbank.mongodb.builder.Distinct;
  import com.allanbank.mongodb.builder.Find;
  import com.allanbank.mongodb.builder.FindAndModify;
  import com.allanbank.mongodb.builder.GroupBy;
  import com.allanbank.mongodb.builder.MapReduce;
  import com.allanbank.mongodb.builder.ParallelScan;
  import com.allanbank.mongodb.builder.write.WriteOperation;
  import com.allanbank.mongodb.client.callback.BatchedNativeWriteCallback;
  import com.allanbank.mongodb.client.callback.BatchedWriteCallback;
  import com.allanbank.mongodb.client.callback.CursorCallback;
  import com.allanbank.mongodb.client.callback.CursorStreamingCallback;
  import com.allanbank.mongodb.client.callback.LongToIntCallback;
  import com.allanbank.mongodb.client.callback.MultipleCursorCallback;
  import com.allanbank.mongodb.client.callback.ReplyArrayCallback;
  import com.allanbank.mongodb.client.callback.ReplyDocumentCallback;
  import com.allanbank.mongodb.client.callback.ReplyIntegerCallback;
  import com.allanbank.mongodb.client.callback.ReplyLongCallback;
  import com.allanbank.mongodb.client.callback.ReplyResultCallback;
  import com.allanbank.mongodb.client.callback.SingleDocumentCallback;
  import com.allanbank.mongodb.client.message.AggregateCommand;
  import com.allanbank.mongodb.client.message.Command;
  import com.allanbank.mongodb.client.message.Delete;
  import com.allanbank.mongodb.client.message.GetLastError;
  import com.allanbank.mongodb.client.message.Insert;
  import com.allanbank.mongodb.client.message.ParallelScanCommand;
  import com.allanbank.mongodb.client.message.Query;
  import com.allanbank.mongodb.client.message.Update;
  
  /**
   * AbstractMongoOperations provides the core functionality for the operations on
   * a MongoDB collection.
   * 
   * @api.no This class is <b>NOT</b> part of the drivers API. This class may be
   *         mutated in incompatible ways between any two releases of the driver.
   * @copyright 2014, Allanbank Consulting, Inc., All Rights Reserved
   */
  public abstract class AbstractMongoOperations {
  
      /**
       * The default for if a delete should only delete the first document it
       * matches.
       */
      public static final boolean DELETE_SINGLE_DELETE_DEFAULT = false;
  
      /** The default empty index options. */
      public static final Document EMPTY_INDEX_OPTIONS = EmptyDocument.INSTANCE;
  
      /** The name of the canonical id field for MongoDB. */
      public static final String ID_FIELD_NAME = "_id";
  
      /** The default for if an insert should continue on an error. */
     public static final boolean INSERT_CONTINUE_ON_ERROR_DEFAULT = false;
 
     /** The default for a UNIQUE index options. */
     public static final Document UNIQUE_INDEX_OPTIONS = new ImmutableDocument(
             BuilderFactory.start().add("unique", true));
 
     /** The default for doing a multiple-update on an update. */
     public static final boolean UPDATE_MULTIUPDATE_DEFAULT = false;
 
     /** The default for doing an upsert on an update. */
     public static final boolean UPDATE_UPSERT_DEFAULT = false;
 
     /** The client for interacting with MongoDB. */
     protected final Client myClient;
 
     /** The name of the database we interact with. */
     protected final MongoDatabase myDatabase;
 
     /** The name of the collection we interact with. */
     protected final String myName;
 
     /** The {@link Durability} for writes from this database instance. */
     private Durability myDurability;
 
     /** The {@link ReadPreference} for reads from this database instance. */
     private ReadPreference myReadPreference;
 
     /**
      * Create a new AbstractAsyncMongoCollection.
      * 
      * @param client
      *            The client for interacting with MongoDB.
      * @param database
      *            The database we interact with.
      * @param name
      *            The name of the collection we interact with.
      */
     public AbstractMongoOperations(final Client client,
             final MongoDatabase database, final String name) {
         super();
 
         myClient = client;
         myDatabase = database;
         myName = name;
         myDurability = null;
         myReadPreference = null;
     }
 
     /**
      * Constructs a {@code aggregate} command and sends it to the server via the
      * {@link Client}.
      * 
      * @param results
      *            Callback for the aggregation results returned.
      * @param command
      *            The details of the aggregation request.
      * @throws MongoDbException
      *             On an error executing the aggregate command.
      * @see AsyncMongoCollection#aggregateAsync(Callback, Aggregate)
      */
     public void aggregateAsync(final Callback<MongoIterator<Document>> results,
             final Aggregate command) throws MongoDbException {
 
         final AggregateCommand commandMsg = toCommand(command, false);
 
         final CursorCallback callback = new CursorCallback(myClient,
                 commandMsg, true, results);
 
         myClient.send(commandMsg, callback);
     }
 
     /**
      * Constructs a {@code count} command and sends it to the server via the
      * {@link Client}.
      * 
      * @param results
      *            The callback to notify of the results.
      * @param count
      *            The count command.
      * @throws MongoDbException
      *             On an error counting the documents.
      * @see AsyncMongoCollection#countAsync(Callback, Count)
      */
     public void countAsync(final Callback<Long> results, final Count count)
             throws MongoDbException {
         Version minVersion = null;
         final DocumentBuilder builder = BuilderFactory.start();
 
         builder.addString("count", getName());
         builder.addDocument("query", count.getQuery());
         if (count.getMaximumTimeMilliseconds() > 0) {
             minVersion = Count.MAX_TIMEOUT_VERSION;
             builder.add("maxTimeMS", count.getMaximumTimeMilliseconds());
         }
 
         // Should be last since might wrap command in a $query element.
         final ReadPreference finalPreference = updateReadPreference(builder,
                 count.getReadPreference(), true);
 
         final Command commandMsg = new Command(getDatabaseName(), getName(),
                 builder.build(), count.getQuery(), finalPreference,
                 VersionRange.minimum(minVersion));
 
         myClient.send(commandMsg, new ReplyLongCallback(results));
     }
 
     /**
      * Constructs a {@link Delete} message and sends it to the server via the
      * {@link Client}.
      * 
      * @param results
      *            Callback that will be notified of the results of the query. If
      *            the durability of the operation is NONE then this will be -1.
      * @param query
      *            Query to locate the documents to be deleted.
      * @param singleDelete
      *            If true then only a single document will be deleted. If
      *            running in a sharded environment then this field must be false
      *            or the query must contain the shard key.
      * @param durability
      *            The durability for the delete.
      * @throws MongoDbException
      *             On an error deleting the documents.
      * @see AsyncMongoCollection#deleteAsync(Callback, DocumentAssignable,
      *      boolean, Durability)
      */
     public void deleteAsync(final Callback<Long> results,
             final DocumentAssignable query, final boolean singleDelete,
             final Durability durability) throws MongoDbException {
 
         if ((durability != Durability.NONE) && useWriteCommand()
                 && isWriteCommandsSupported(null)) {
 
             final BatchedWrite write = BatchedWrite.delete(query, singleDelete,
                     durability);
 
             writeAsync(results, write);
         }
         else {
             final Delete deleteMessage = new Delete(getDatabaseName(), myName,
                     query.asDocument(), singleDelete);
 
             if (Durability.NONE.equals(durability)) {
                 myClient.send(deleteMessage, null);
                 results.callback(Long.valueOf(-1));
             }
             else {
                 myClient.send(deleteMessage, asGetLastError(durability),
                         new ReplyLongCallback(results));
             }
         }
     }
 
     /**
      * Constructs a {@code distinct} command and sends it to the server via the
      * {@link Client}.
      * 
      * @param results
      *            Callback for the distinct results returned.
      * @param command
      *            The details of the distinct request.
      * @throws MongoDbException
      *             On an error finding the documents.
      * @see AsyncMongoCollection#distinctAsync(Callback, Distinct)
      */
     public void distinctAsync(final Callback<MongoIterator<Element>> results,
             final Distinct command) throws MongoDbException {
 
         Version minVersion = null;
 
         final DocumentBuilder builder = BuilderFactory.start();
 
         builder.addString("distinct", getName());
         builder.addString("key", command.getKey());
         if (command.getQuery() != null) {
             builder.addDocument("query", command.getQuery());
         }
         if (command.getMaximumTimeMilliseconds() > 0) {
             minVersion = Distinct.MAX_TIMEOUT_VERSION;
             builder.add("maxTimeMS", command.getMaximumTimeMilliseconds());
         }
 
         // Should be last since might wrap command in a $query element.
         final ReadPreference readPreference = updateReadPreference(builder,
                 command.getReadPreference(), true);
 
         final Command commandMsg = new Command(getDatabaseName(), getName(),
                 builder.build(), readPreference,
                 VersionRange.minimum(minVersion));
 
         myClient.send(commandMsg, new ReplyArrayCallback(results));
 
     }
 
     /**
      * Constructs a {@link AggregateCommand} and sends it to the server via the
      * {@link Client}.
      * 
      * @param aggregation
      *            The aggregation details.
      * @param results
      *            Callback that will be notified of the results of the explain.
      * @throws MongoDbException
      *             On an error finding the documents.
      * @since MongoDB 2.6
      * @see AsyncMongoCollection#explainAsync(Callback, Aggregate)
      */
     public void explainAsync(final Callback<Document> results,
             final Aggregate aggregation) throws MongoDbException {
         final AggregateCommand commandMsg = toCommand(aggregation, true);
 
         myClient.send(commandMsg, new SingleDocumentCallback(results));
     }
 
     /**
      * Constructs a {@link Query} message and sends it to the server via the
      * {@link Client}.
      * 
      * @param query
      *            The query details.
      * @param results
      *            Callback that will be notified of the results of the explain.
      * @throws MongoDbException
      *             On an error finding the documents.
      * @see AsyncMongoCollection#explainAsync(Callback, Find)
      */
     public void explainAsync(final Callback<Document> results, final Find query)
             throws MongoDbException {
 
         ReadPreference readPreference = query.getReadPreference();
         if (readPreference == null) {
             readPreference = getReadPreference();
         }
 
         Document queryDoc;
         if (!readPreference.isLegacy()
                 && (myClient.getClusterType() == ClusterType.SHARDED)) {
             queryDoc = query.toQueryRequest(true, readPreference);
         }
         else {
             queryDoc = query.toQueryRequest(true);
         }
 
         final Query queryMessage = new Query(getDatabaseName(), myName,
                 queryDoc, query.getProjection(), query.getBatchSize(),
                 query.getLimit(), query.getNumberToSkip(),
                 false /* tailable */, readPreference,
                 false /* noCursorTimeout */, false /* awaitData */,
                 false /* exhaust */, query.isPartialOk());
 
         myClient.send(queryMessage, new SingleDocumentCallback(results));
     }
 
     /**
      * Constructs a {@code findAndModify} command and sends it to the server via
      * the {@link Client}.
      * 
      * @param results
      *            Callback for the the found document.
      * @param command
      *            The details of the find and modify request.
      * @throws MongoDbException
      *             On an error finding the documents.
      * @see AsyncMongoCollection#findAndModifyAsync(Callback, FindAndModify)
      */
     public void findAndModifyAsync(final Callback<Document> results,
             final FindAndModify command) throws MongoDbException {
         Version minVersion = null;
 
         final DocumentBuilder builder = BuilderFactory.start();
 
         builder.addString("findAndModify", getName());
         builder.addDocument("query", command.getQuery());
         if (command.getUpdate() != null) {
             builder.addDocument("update", command.getUpdate());
         }
         if (command.getSort() != null) {
             builder.addDocument("sort", command.getSort());
         }
         if (command.getFields() != null) {
             builder.addDocument("fields", command.getFields());
         }
         if (command.isRemove()) {
             builder.addBoolean("remove", true);
         }
         if (command.isReturnNew()) {
             builder.addBoolean("new", true);
         }
         if (command.isUpsert()) {
             builder.addBoolean("upsert", true);
         }
         if (command.getMaximumTimeMilliseconds() > 0) {
             minVersion = FindAndModify.MAX_TIMEOUT_VERSION;
             builder.add("maxTimeMS", command.getMaximumTimeMilliseconds());
         }
 
         // Must be the primary since this is a write.
         final Command commandMsg = new Command(getDatabaseName(), getName(),
                 builder.build(), command.getQuery(), ReadPreference.PRIMARY,
                 VersionRange.minimum(minVersion));
         myClient.send(commandMsg, new ReplyDocumentCallback(results));
     }
 
     /**
      * Constructs a {@link Query} message and sends it to the server via the
      * {@link Client}.
      * 
      * @param query
      *            The query details.
      * @param results
      *            Callback that will be notified of the results of the find.
      * @throws MongoDbException
      *             On an error finding the documents.
      * @see AsyncMongoCollection#findAsync(Callback, Find)
      */
     public void findAsync(final Callback<MongoIterator<Document>> results,
             final Find query) throws MongoDbException {
 
         final Query queryMessage = createQuery(query, query.getLimit(),
                 query.getBatchSize(), query.isTailable(), query.isAwaitData(),
                 query.isImmortalCursor());
 
         final CursorCallback callback = new CursorCallback(myClient,
                 queryMessage, false, results);
 
         myClient.send(queryMessage, callback);
     }
 
     /**
      * Constructs a {@link Query} message and sends it to the server via the
      * {@link Client}.
      * 
      * @param query
      *            The query details.
      * @param results
      *            Callback that will be notified of the results of the find.
      * @throws MongoDbException
      *             On an error finding the documents.
      * @see AsyncMongoCollection#findOneAsync(Callback, Find)
      */
     public void findOneAsync(final Callback<Document> results, final Find query)
             throws MongoDbException {
         final Query queryMessage = createQuery(query, 1, 1, false, false, false);
 
         myClient.send(queryMessage, new SingleDocumentCallback(results));
     }
 
     /**
      * Returns the name of the database.
      * 
      * @return The name of the database.
      */
     public String getDatabaseName() {
         return myDatabase.getName();
     }
 
     /**
      * Returns the durability to use when no durability is specified for the
      * write operation.
      * 
      * @return The durability to use when no durability is specified for the
      *         write operation.
      */
     public Durability getDurability() {
         Durability result = myDurability;
         if (result == null) {
             result = myDatabase.getDurability();
         }
         return result;
     }
 
     /**
      * Returns the name of the collection.
      * 
      * @return The name of the collection.
      */
     public String getName() {
         return myName;
     }
 
     /**
      * Returns the read preference to use when no read preference is specified
      * for the read operation.
      * 
      * @return The read preference to use when no read preference is specified
      *         for the read operation.
      */
     public ReadPreference getReadPreference() {
         ReadPreference result = myReadPreference;
         if (result == null) {
             result = myDatabase.getReadPreference();
         }
         return result;
     }
 
     /**
      * Constructs a {@code group} command and sends it to the server via the
      * {@link Client}.
      * 
      * @param results
      *            Callback for the group results returned.
      * @param command
      *            The details of the group request.
      * @throws MongoDbException
      *             On an error finding the documents.
      * @see AsyncMongoCollection#groupByAsync(Callback, GroupBy)
      */
     public void groupByAsync(final Callback<MongoIterator<Element>> results,
             final GroupBy command) throws MongoDbException {
         Version minVersion = null;
 
         final DocumentBuilder builder = BuilderFactory.start();
 
         final DocumentBuilder groupDocBuilder = builder.push("group");
 
         groupDocBuilder.addString("ns", getName());
         if (!command.getKeys().isEmpty()) {
             final DocumentBuilder keysBuilder = groupDocBuilder.push("key");
             for (final String key : command.getKeys()) {
                 keysBuilder.addBoolean(key, true);
             }
         }
         if (command.getKeyFunction() != null) {
             groupDocBuilder.addJavaScript("$keyf", command.getKeyFunction());
         }
         if (command.getInitialValue() != null) {
             groupDocBuilder.addDocument("initial", command.getInitialValue());
         }
         if (command.getReduceFunction() != null) {
             groupDocBuilder.addJavaScript("$reduce",
                     command.getReduceFunction());
         }
         if (command.getFinalizeFunction() != null) {
             groupDocBuilder.addJavaScript("finalize",
                     command.getFinalizeFunction());
         }
         if (command.getQuery() != null) {
             groupDocBuilder.addDocument("cond", command.getQuery());
         }
         if (command.getMaximumTimeMilliseconds() > 0) {
             minVersion = GroupBy.MAX_TIMEOUT_VERSION;
             // maxTimeMS is not in the "group" sub-doc.
             // See SERVER-12595 commands.
             builder.add("maxTimeMS", command.getMaximumTimeMilliseconds());
         }
 
         // Should be last since might wrap command in a $query element.
         final ReadPreference readPreference = updateReadPreference(
                 groupDocBuilder, command.getReadPreference(), false);
 
         final Command commandMsg = new Command(getDatabaseName(), getName(),
                 builder.build(), readPreference,
                 VersionRange.minimum(minVersion));
         myClient.send(commandMsg, new ReplyArrayCallback("retval", results));
     }
 
     /**
      * Constructs a {@link Insert} message and sends it to the server via the
      * {@link Client}.
      * 
      * @param results
      *            {@link Callback} that will be notified with the results of the
      *            insert. Currently, the value is always zero. Once <a
      *            href="http://jira.mongodb.org/browse/SERVER-4381"
      *            >SERVER-4381</a> is fixed then expected to be the number of
      *            documents inserted. If the durability is NONE then returns
      *            <code>-1</code>.
      * @param continueOnError
      *            If the insert should continue if one of the documents causes
      *            an error.
      * @param durability
      *            The durability for the insert.
      * @param documents
      *            The documents to add to the collection.
      * @throws MongoDbException
      *             On an error inserting the documents.
      * @see AsyncMongoCollection#insertAsync(Callback, boolean, Durability,
      *      DocumentAssignable...)
      */
     public void insertAsync(final Callback<Integer> results,
             final boolean continueOnError, final Durability durability,
             final DocumentAssignable... documents) throws MongoDbException {
 
         doInsertAsync(results, continueOnError, durability, null, documents);
     }
 
     /**
      * Constructs a {@code mapreduce} command and sends it to the server via the
      * {@link Client}.
      * 
      * @param results
      *            Callback for the map/reduce results returned. Note this might
      *            be empty if the output type is not inline.
      * @param command
      *            The details of the map/reduce request.
      * @throws MongoDbException
      *             On an error finding the documents.
      * @see AsyncMongoCollection#mapReduceAsync(Callback, MapReduce)
      */
     public void mapReduceAsync(final Callback<MongoIterator<Document>> results,
             final MapReduce command) throws MongoDbException {
         Version minVersion = null;
 
         final DocumentBuilder builder = BuilderFactory.start();
 
         builder.addString("mapreduce", getName());
         builder.addJavaScript("map", command.getMapFunction());
         builder.addJavaScript("reduce", command.getReduceFunction());
         if (command.getFinalizeFunction() != null) {
             builder.addJavaScript("finalize", command.getFinalizeFunction());
         }
         if (command.getQuery() != null) {
             builder.addDocument("query", command.getQuery());
         }
         if (command.getSort() != null) {
             builder.addDocument("sort", command.getSort());
         }
         if (command.getScope() != null) {
             builder.addDocument("scope", command.getScope());
         }
         if (command.getLimit() != 0) {
             builder.addInteger("limit", command.getLimit());
         }
         if (command.isKeepTemp()) {
             builder.addBoolean("keeptemp", true);
         }
         if (command.isJsMode()) {
             builder.addBoolean("jsMode", true);
         }
         if (command.isVerbose()) {
             builder.addBoolean("verbose", true);
         }
         if (command.getMaximumTimeMilliseconds() > 0) {
             minVersion = MapReduce.MAX_TIMEOUT_VERSION;
             builder.add("maxTimeMS", command.getMaximumTimeMilliseconds());
         }
 
         final DocumentBuilder outputBuilder = builder.push("out");
         switch (command.getOutputType()) {
         case INLINE: {
             outputBuilder.addInteger("inline", 1);
             break;
         }
         case REPLACE: {
             outputBuilder.addString("replace", command.getOutputName());
             if (command.getOutputDatabase() != null) {
                 outputBuilder.addString("db", command.getOutputDatabase());
             }
             break;
         }
         case MERGE: {
             outputBuilder.addString("merge", command.getOutputName());
             if (command.getOutputDatabase() != null) {
                 outputBuilder.addString("db", command.getOutputDatabase());
             }
             break;
         }
         case REDUCE: {
             outputBuilder.addString("reduce", command.getOutputName());
             if (command.getOutputDatabase() != null) {
                 outputBuilder.addString("db", command.getOutputDatabase());
             }
             break;
         }
         }
 
         // Should be last since might wrap command in a $query element.
         final ReadPreference readPreference = updateReadPreference(builder,
                 command.getReadPreference(), true);
 
         final Command commandMsg = new Command(getDatabaseName(), getName(),
                 builder.build(), readPreference,
                 VersionRange.minimum(minVersion));
         myClient.send(commandMsg, new ReplyResultCallback(results));
     }
 
     /**
      * Constructs a {@code parallelCollectionScan} command and sends it to the
      * server via the {@link Client}.
      * 
      * @param results
      *            Callback for the collection of iterators.
      * @param parallelScan
      *            The details on the scan.
      * @throws MongoDbException
      *             On an error initializing the parallel scan.
      * @see AsyncMongoCollection#parallelScanAsync(Callback, ParallelScan)
      * @see <a
      *      href="http://docs.mongodb.org/manual/reference/command/parallelCollectionScan/">parallelCollectionScan
      *      Command</a>
      */
     public void parallelScanAsync(
             final Callback<Collection<MongoIterator<Document>>> results,
             final ParallelScan parallelScan) throws MongoDbException {
         final DocumentBuilder builder = BuilderFactory.start();
 
         builder.add("parallelCollectionScan", getName());
         builder.add("numCursors", parallelScan.getRequestedIteratorCount());
 
         // Should be last since might wrap command in a $query element.
         final ReadPreference readPreference = updateReadPreference(builder,
                 parallelScan.getReadPreference(), true);
 
         final ParallelScanCommand commandMsg = new ParallelScanCommand(
                 parallelScan, getDatabaseName(), getName(), builder.build(),
                 readPreference);
 
         myClient.send(commandMsg, new MultipleCursorCallback(myClient,
                 commandMsg, results));
 
     }
 
     /**
      * Constructs a {@link Insert} of {@link Update} message based on if the
      * document contains a {@link #ID_FIELD_NAME} and sends it to the server via
      * the {@link Client}.
      * 
      * @param results
      *            {@link Callback} that will be notified with the results of the
      *            insert. If the durability of the operation is NONE then this
      *            will be -1.
      * @param document
      *            The document to save to the collection.
      * @param durability
      *            The durability for the save.
      * @throws MongoDbException
      *             On an error saving the documents.
      * @see AsyncMongoCollection#saveAsync(Callback, DocumentAssignable,
      *      Durability)
      */
     public void saveAsync(final Callback<Integer> results,
             final DocumentAssignable document, final Durability durability)
             throws MongoDbException {
         final Document doc = document.asDocument();
 
         if (doc.contains(ID_FIELD_NAME)) {
             updateAsync(new LongToIntCallback(results), BuilderFactory.start()
                     .add(doc.get(ID_FIELD_NAME)), doc, false, true, durability);
         }
         else {
             insertAsync(results, INSERT_CONTINUE_ON_ERROR_DEFAULT, durability,
                     doc);
         }
     }
 
     /**
      * Sets the durability to use when no durability is specified for the write
      * operation.
      * 
      * @param durability
      *            The durability to use when no durability is specified for the
      *            write operation.
      */
     public void setDurability(final Durability durability) {
         myDurability = durability;
     }
 
     /**
      * Sets the read preference to use when no read preference is specified for
      * the read operation.
      * 
      * @param readPreference
      *            The read preference to use when no read preference is
      *            specified for the read operation.
      */
     public void setReadPreference(final ReadPreference readPreference) {
         myReadPreference = readPreference;
     }
 
     /**
      * Constructs a {@code aggregate} command and sends it to the server via the
      * {@link Client}.
      * 
      * @param results
      *            Callback that will be notified of the results of the query.
      * @param aggregation
      *            The aggregation details.
      * @return A {@link MongoCursorControl} to control the cursor streaming
      *         documents to the caller. This includes the ability to stop the
      *         cursor and persist its state.
      * @throws MongoDbException
      *             On an error finding the documents.
      * @see AsyncMongoCollection#stream(StreamCallback, Aggregate)
      */
     public MongoCursorControl stream(final StreamCallback<Document> results,
             final Aggregate aggregation) throws MongoDbException {
         final AggregateCommand commandMsg = toCommand(aggregation, false);
 
         final CursorStreamingCallback callback = new CursorStreamingCallback(
                 myClient, commandMsg, true, results);
 
         myClient.send(commandMsg, callback);
 
         return callback;
     }
 
     /**
      * Constructs a {@link Query} message and sends it to the server via the
      * {@link Client}.
      * 
      * @param results
      *            Callback that will be notified of the results of the query.
      * @param query
      *            The query details.
      * @return A {@link MongoCursorControl} to control the cursor streaming
      *         documents to the caller. This includes the ability to stop the
      *         cursor and persist its state.
      * @throws MongoDbException
      *             On an error finding the documents.
      * @see AsyncMongoCollection#stream(StreamCallback, Find)
      */
     public MongoCursorControl stream(final StreamCallback<Document> results,
             final Find query) throws MongoDbException {
         final Query queryMessage = createQuery(query, query.getLimit(),
                 query.getBatchSize(), query.isTailable(), query.isAwaitData(),
                 query.isImmortalCursor());
 
         final CursorStreamingCallback callback = new CursorStreamingCallback(
                 myClient, queryMessage, false, results);
 
         myClient.send(queryMessage, callback);
 
         return callback;
     }
 
     /**
      * Constructs a {@code text} command and sends it to the server via the
      * {@link Client}.
      * 
      * @param results
      *            Callback for the {@code text} results returned.
      * @param command
      *            The details of the {@code text} request.
      * @throws MongoDbException
      *             On an error executing the {@code text} command.
      * @see <a
      *      href="http://docs.mongodb.org/manual/release-notes/2.4/#text-queries">
      *      MongoDB Text Queries</a>
      * @since MongoDB 2.4
      * @see AsyncMongoCollection#textSearchAsync(Callback,
      *      com.allanbank.mongodb.builder.Text)
      * @deprecated Support for the {@code text} command was deprecated in the
      *             2.6 version of MongoDB. Use the
      *             {@link ConditionBuilder#text(String) $text} query operator
      *             instead. This method will not be removed until two releases
      *             after the MongoDB 2.6 release (e.g. 2.10 if the releases are
      *             2.8 and 2.10).
      */
     @Deprecated
     public void textSearchAsync(
             final Callback<MongoIterator<com.allanbank.mongodb.builder.TextResult>> results,
             final com.allanbank.mongodb.builder.Text command)
             throws MongoDbException {
         final Version minVersion = com.allanbank.mongodb.builder.Text.REQUIRED_VERSION;
         final DocumentBuilder builder = BuilderFactory.start();
 
         builder.addString("text", getName());
         builder.addString("search", command.getSearchTerm());
         if (command.getQuery() != null) {
             builder.add("filter", command.getQuery());
         }
         if (command.getLimit() > 0) {
             builder.add("limit", command.getLimit());
         }
         if (command.getReturnFields() != null) {
             builder.add("project", command.getReturnFields());
         }
         if (command.getLanguage() != null) {
             builder.add("language", command.getLanguage());
         }
 
         // Should be last since might wrap command in a $query element.
         final ReadPreference readPreference = updateReadPreference(builder,
                 command.getReadPreference(), true);
 
         final Command commandMsg = new Command(getDatabaseName(), getName(),
                 builder.build(), readPreference,
                 VersionRange.minimum(minVersion));
         myClient.send(commandMsg,
                 new ReplyResultCallback(
                         new com.allanbank.mongodb.client.callback.TextCallback(
                                 results)));
     }
 
     /**
      * Constructs a {@link Update} message and sends it to the server via the
      * {@link Client}.
      * 
      * @param results
      *            The {@link Callback} that will be notified of the number of
      *            documents updated. If the durability of the operation is NONE
      *            then this will be -1.
      * @param query
      *            The query to select the documents to update.
      * @param update
      *            The updates to apply to the selected documents.
      * @param multiUpdate
      *            If true then the update is applied to all of the matching
      *            documents, otherwise only the first document found is updated.
      * @param upsert
      *            If true then if no document is found then a new document is
      *            created and updated, otherwise no operation is performed.
      * @param durability
      *            The durability for the update.
      * @throws MongoDbException
      *             On an error updating the documents.
      * @see AsyncMongoCollection#updateAsync(Callback, DocumentAssignable,
      *      DocumentAssignable, boolean, boolean, Durability)
      */
     public void updateAsync(final Callback<Long> results,
             final DocumentAssignable query, final DocumentAssignable update,
             final boolean multiUpdate, final boolean upsert,
             final Durability durability) throws MongoDbException {
 
         final ClusterStats stats = myClient.getClusterStats();
         if ((durability != Durability.NONE) && useWriteCommand()
                 && isWriteCommandsSupported(stats)) {
             final BatchedWrite write = BatchedWrite.update(query, update,
                     multiUpdate, upsert, durability);
 
             doWriteAsync(stats, results, write);
         }
         else {
             final Update updateMessage = new Update(getDatabaseName(), myName,
                     query.asDocument(), update.asDocument(), multiUpdate,
                     upsert);
 
             if (Durability.NONE == durability) {
                 myClient.send(updateMessage, null);
                 results.callback(Long.valueOf(-1));
             }
             else {
                 myClient.send(updateMessage, asGetLastError(durability),
                         new ReplyLongCallback(results));
             }
         }
     }
 
     /**
      * Constructs the appropriate set of write commands to send to the server.
      * 
      * @param results
      *            The {@link Callback} that will be notified of the number of
      *            documents inserted, updated, and deleted. If the durability of
      *            the operation is NONE then this will be -1.
      * @param write
      *            The batched writes
      * @throws MongoDbException
      *             On an error submitting the write operations.
      * 
      * @since MongoDB 2.6
      * @see AsyncMongoCollection#writeAsync(Callback,BatchedWrite)
      */
     public void writeAsync(final Callback<Long> results,
             final BatchedWrite write) throws MongoDbException {
         final ClusterStats stats = myClient.getClusterStats();
 
         doWriteAsync(stats, results, write);
     }
 
     /**
      * Converts the {@link Durability} into a {@link GetLastError} command.
      * 
      * @param durability
      *            The {@link Durability} to convert.
      * @return The {@link GetLastError} command.
      */
     protected GetLastError asGetLastError(final Durability durability) {
         return new GetLastError(getDatabaseName(), durability);
     }
 
     /**
      * Creates a properly configured {@link Query} message.
      * 
      * @param query
      *            The {@link Find} to construct the {@link Query} from.
      * @param limit
      *            The limit for the query.
      * @param batchSize
      *            The batch size for the query.
      * @param tailable
      *            If the query should create a tailable cursor.
      * @param awaitData
      *            If the query should await data.
      * @param immortal
      *            If the query should create a cursor that does not timeout,
      *            e.g., immortal.
      * @return The {@link Query} message.
      */
     protected Query createQuery(final Find query, final int limit,
             final int batchSize, final boolean tailable,
             final boolean awaitData, final boolean immortal) {
         ReadPreference readPreference = query.getReadPreference();
         if (readPreference == null) {
             readPreference = getReadPreference();
         }
 
         Document queryDoc;
         if (!readPreference.isLegacy()
                 && (myClient.getClusterType() == ClusterType.SHARDED)) {
             queryDoc = query.toQueryRequest(false, readPreference);
         }
         else {
             queryDoc = query.toQueryRequest(false);
         }
 
         return new Query(getDatabaseName(), myName, queryDoc,
                 query.getProjection(), batchSize, limit,
                 query.getNumberToSkip(), tailable, readPreference, immortal,
                 awaitData, false /* exhaust */, query.isPartialOk());
     }
 
     /**
      * Sends an {@link Insert} message to the server. This version is private to
      * this class since most inserts do not need the server version.
      * 
      * @param results
      *            {@link Callback} that will be notified with the results of the
      *            insert.
      * @param continueOnError
      *            If the insert should continue if one of the documents causes
      *            an error.
      * @param durability
      *            The durability for the insert.
      * @param requiredServerVersion
      *            The required version of the server to support processing the
      *            message.
      * @param documents
      *            The documents to add to the collection.
      * @throws MongoDbException
      *             On an error inserting the documents.
      */
     protected void doInsertAsync(final Callback<Integer> results,
             final boolean continueOnError, final Durability durability,
             final Version requiredServerVersion,
             final DocumentAssignable... documents) throws MongoDbException {
 
         final ClusterStats stats = myClient.getClusterStats();
         if ((durability != Durability.NONE) && useWriteCommand()
                 && isWriteCommandsSupported(stats)) {
             final BatchedWrite write = BatchedWrite.insert(continueOnError,
                     durability, documents);
 
             doWriteAsync(stats, new LongToIntCallback(results), write);
         }
         else {
             // Make sure the documents have an _id.
             final List<Document> docs = new ArrayList<Document>(
                     documents.length);
             for (final DocumentAssignable docAssignable : documents) {
                 final Document doc = docAssignable.asDocument();
                 if (!doc.contains(ID_FIELD_NAME)
                         && (doc instanceof RootDocument)) {
                     ((RootDocument) doc).injectId();
                 }
                 docs.add(doc);
             }
 
             final Insert insertMessage = new Insert(getDatabaseName(), myName,
                     docs, continueOnError,
                     VersionRange.minimum(requiredServerVersion));
             if (Durability.NONE == durability) {
                 myClient.send(insertMessage, null);
                 results.callback(Integer.valueOf(-1));
             }
             else {
                 myClient.send(insertMessage, asGetLastError(durability),
                         new ReplyIntegerCallback(results));
             }
         }
     }
 
     /**
      * Performs a async write operation.
      * 
      * @param stats
      *            The stats for verifying the server support write commands.
      * @param results
      *            The callback for the write.
      * @param write
      *            The write to send.
      */
     protected void doWriteAsync(final ClusterStats stats,
             final Callback<Long> results, final BatchedWrite write) {
         if (isWriteCommandsSupported(stats)) {
 
             final List<BatchedWrite.Bundle> bundles = write.toBundles(
                     getName(), stats.getSmallestMaxBsonObjectSize(),
                     stats.getSmallestMaxBatchedWriteOperations());
             if (bundles.isEmpty()) {
                 results.callback(Long.valueOf(0));
                 return;
             }
 
             final BatchedWriteCallback callback = new BatchedWriteCallback(
                     getDatabaseName(), getName(), results, write, myClient,
                     bundles);
 
             // Push the messages out.
             callback.send();
         }
         else {
             final List<WriteOperation> operations = write.getWrites();
             if (operations.isEmpty()) {
                 results.callback(Long.valueOf(0));
                 return;
             }
 
             final BatchedNativeWriteCallback callback = new BatchedNativeWriteCallback(
                     results, write, this, operations);
 
             // Push the messages out.
             callback.send();
         }
     }
 
     /**
      * Determines if all of the servers in the cluster support the write
      * commands.
      * 
      * @param stats
      *            The cluster stats if they have already been retrieved.
      * @return True if all servers in the cluster are at least the
      *         {@link BatchedWrite#REQUIRED_VERSION}.
      */
     protected boolean isWriteCommandsSupported(final ClusterStats stats) {
         final ClusterStats clusterStats = (stats == null) ? myClient
                 .getClusterStats() : stats;
         final VersionRange serverVersionRange = clusterStats
                 .getServerVersionRange();
         final Version minServerVersion = serverVersionRange.getLowerBounds();
 
         return (BatchedWrite.REQUIRED_VERSION.compareTo(minServerVersion) <= 0);
     }
 
     /**
      * Converts the {@link Aggregate} object to an {@link AggregateCommand}.
      * 
      * @param command
      *            The {@link Aggregate} to convert.
      * @param explain
      *            If rue then have the server explain the aggregation instead of
      *            performing the aggregation.
      * @return The command to send to the server for the {@link Aggregate}.
      */
     protected AggregateCommand toCommand(final Aggregate command,
             final boolean explain) {
         Version minVersion = command.getRequiredVersion();
 
         final DocumentBuilder builder = BuilderFactory.start();
 
         builder.addString("aggregate", getName());
 
         // Pipeline of operations.
         final ArrayBuilder pipeline = builder.pushArray("pipeline");
         for (final Element e : command.getPipeline()) {
             pipeline.add(e);
         }
 
         // Options
         if (command.isAllowDiskUsage()) {
             builder.add("allowDiskUsage", true);
         }
         if (command.isUseCursor()) {
             final DocumentBuilder cursor = builder.push("cursor");
             if (command.getBatchSize() > 0) {
                 cursor.add("batchSize", command.getBatchSize());
             }
         }
         if (explain) {
             minVersion = Version.later(minVersion, Aggregate.EXPLAIN_VERSION);
             builder.add("explain", true);
         }
         if (command.getMaximumTimeMilliseconds() > 0) {
             builder.add("maxTimeMS", command.getMaximumTimeMilliseconds());
         }
 
         // Should be last since might wrap command in a $query element.
         final ReadPreference readPreference = updateReadPreference(builder,
                 command.getReadPreference(), true);
 
         final AggregateCommand commandMsg = new AggregateCommand(command,
                 getDatabaseName(), getName(), builder.build(), readPreference,
                 VersionRange.minimum(minVersion));
         return commandMsg;
     }
 
     /**
      * Determines the {@link ReadPreference} to be used based on the command's
      * {@code ReadPreference} or the collection's if the command's
      * {@code ReadPreference} is <code>null</code>. Updates the command's
      * {@link DocumentBuilder} with the {@code ReadPreference} details if
      * connected to a sharded cluster and the resulting {@code ReadPreference}
      * is not supported by the legacy settings.
      * 
      * @param builder
      *            The builder for the command document to augment with the read
      *            preferences if connected to a sharded cluster.
      * @param commandReadPreference
      *            The read preferences from the command.
      * @param createQueryElement
      *            If true then the existing builder's contents will be pushed
      *            into a $query sub-document. This is required to ensure the
      *            command is not rejected by the {@code mongod} after processing
      *            by the {@code mongos}.
      * @return The {@link ReadPreference} to use.
      */
     protected ReadPreference updateReadPreference(
             final DocumentBuilder builder,
             final ReadPreference commandReadPreference,
             final boolean createQueryElement) {
 
         ReadPreference readPreference = commandReadPreference;
         if (readPreference == null) {
             readPreference = getReadPreference();
         }
 
         if (!readPreference.isLegacy()
                 && (myClient.getClusterType() == ClusterType.SHARDED)) {
             if (createQueryElement) {
                 final Document query = builder.asDocument();
                 builder.reset();
                 builder.add("$query", query);
             }
             builder.add(ReadPreference.FIELD_NAME, readPreference);
         }
 
         return readPreference;
     }
 
     /**
      * Extension point for derived classes to force the
      * {@link #insertAsync(Callback, boolean, Durability, DocumentAssignable...)}
      * ,
      * {@link #updateAsync(Callback, DocumentAssignable, DocumentAssignable, boolean, boolean, Durability)}
      * , and
      * {@link #deleteAsync(Callback, DocumentAssignable, boolean, Durability)}
      * methods to use the legacy {@link Insert}, {@link Update}, and
      * {@link Delete} messages regardless of the server version.
      * <p>
      * This version of the method always returns true.
      * </p>
      * 
      * @return Return true to allow the use of the write commands added in
      *         MongoDB 2.6. Use false to force the use of the {@link Insert},
      *         {@link Update}, and {@link Delete}
      */
     protected boolean useWriteCommand() {
         return true;
     }
 }