/*

  * #%L

  * BatchedWrite.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.builder;

 import java.io.Serializable;

 import java.util.ArrayList;

 import java.util.Collections;

 import java.util.LinkedHashMap;

 import java.util.LinkedList;

 import java.util.List;

 import java.util.Map;

 import java.util.SortedMap;

 import java.util.TreeMap;

 import com.allanbank.mongodb.BatchedAsyncMongoCollection;

 import com.allanbank.mongodb.Durability;

 import com.allanbank.mongodb.MongoCollection;

 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.builder.write.DeleteOperation;

 import com.allanbank.mongodb.builder.write.InsertOperation;

 import com.allanbank.mongodb.builder.write.UpdateOperation;

 import com.allanbank.mongodb.builder.write.WriteOperation;

 import com.allanbank.mongodb.builder.write.WriteOperationType;

 import com.allanbank.mongodb.error.DocumentToLargeException;

 /**

  * BatchedWrite provides a container for a group of write operations to be sent

  * to the server as one group.

  * <p>

  * The default mode ({@link BatchedWriteMode#SERIALIZE_AND_CONTINUE}) for this

  * class is to submit the operations to the server in the order that they are

  * added to the Builder and to apply as many of the writes as possible (commonly

  * referred to as continue-on-error). This has the effect of causing the fewest

  * surprises and optimizing the performance of the writes since the driver can

  * send multiple distinct writes to the server at once.

  * </p>

  * <p>

  * The {@link BatchedWriteMode#SERIALIZE_AND_STOP} mode also sends each write as

  * a separate request but instead of attempting all writes the driver will stop

  * sending requests once one of the writes fails. This also prevents the driver

  * from sending multiple write messages to the server which can degrade

  * performance.

  * </p>

  * <p>

  * The last mode, {@link BatchedWriteMode#REORDERED}, may re-order writes to

  * maximize performance. Similar to the

  * {@link BatchedWriteMode#SERIALIZE_AND_CONTINUE} this mode will also attempt

  * all writes. The reordering of writes is across all {@link WriteOperationType}

  * s.

  * </p>

  * <p>

  * If using a MongoDB server after {@link #REQUIRED_VERSION 2.5.5} a batched

  * write will result in use of the new write commands.

  * </p>

  * <p>

  * For a more generalized batched write and query capability see the

  * {@link BatchedAsyncMongoCollection} and {@link MongoCollection#startBatch()}.

  * </p>

  * 

  * @api.yes This class is part of the driver's API. Public and protected members

  *          will be deprecated for at least 1 non-bugfix release (version

  *          numbers are <major>.<minor>.<bugfix>) before being

  *          removed or modified.

  * @copyright 2014, Allanbank Consulting, Inc., All Rights Reserved

  */

 public class BatchedWrite implements Serializable {

     /** The first version of MongoDB to support the {@code aggregate} command. */

     public static final Version REQUIRED_VERSION = Version.parse("2.5.5");

     /** Serialization version for the class. */

     private static final long serialVersionUID = 6984498574755719178L;

     /**

      * Creates a new builder for a {@link BatchedWrite}.

      * 

      * @return The builder to construct a {@link BatchedWrite}.

      */

     public static Builder builder() {

         return new Builder();

     }

     /**

      * Create a batched write with a single delete operation. Users can just use

      * the {@link MongoCollection#delete} variants and the driver will convert

      * the deletes to batched writes as appropriate.

      * <p>

      * This method avoids the construction of a builder.

      * </p>

      * 

      * @param query

      *            The query to find the documents to delete.

      * @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 of the delete.

      * @return The BatchedWrite with the single delete.

      */

     public static BatchedWrite delete(final DocumentAssignable query,

             final boolean singleDelete, final Durability durability) {

         final DeleteOperation op = new DeleteOperation(query, singleDelete);

         return new BatchedWrite(op, BatchedWriteMode.SERIALIZE_AND_CONTINUE,

                 durability);

     }

     /**

      * Create a batched write with a single inserts operation. Users can just

      * use the {@link MongoCollection#insert} variants and the driver will

      * convert the inserts to batched writes as appropriate.

      * <p>

      * This method avoids the construction of a builder.

      * </p>

      * 

      * @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.

      * @return The BatchedWrite with the inserts.

      */

     public static BatchedWrite insert(final boolean continueOnError,

             final Durability durability, final DocumentAssignable... documents) {

         final List<WriteOperation> ops = new ArrayList<WriteOperation>(

                 documents.length);

         for (final DocumentAssignable doc : documents) {

             ops.add(new InsertOperation(doc));

         }

         return new BatchedWrite(ops,

                 continueOnError ? BatchedWriteMode.SERIALIZE_AND_CONTINUE

                         : BatchedWriteMode.SERIALIZE_AND_STOP, durability);

     }

     /**

      * Create a batched write with a single update operation. Users can just use

      * the {@link MongoCollection#update} variants and the driver will convert

      * the updates to batched writes as appropriate.

      * 

      * @param query

      *            The query for the update.

      * @param update

      *            The update for the update.

      * @param multiUpdate

      *            If true then the update will update multiple documents.

      * @param upsert

      *            If no document is found then upsert the document.

      * @param durability

      *            The durability of the update.

      * @return The BatchedWrite with the single update.

      */

     public static BatchedWrite update(final DocumentAssignable query,

             final DocumentAssignable update, final boolean multiUpdate,

             final boolean upsert, final Durability durability) {

         final UpdateOperation op = new UpdateOperation(query, update,

                 multiUpdate, upsert);

         return new BatchedWrite(op, BatchedWriteMode.SERIALIZE_AND_CONTINUE,

                 durability);

     }

     /** The durability for the writes. */

     private final Durability myDurability;

     /** The mode for submitting the writes to the server. */

     private final BatchedWriteMode myMode;

     /** The writes to submit to the server. */

     private final List<WriteOperation> myWrites;

     /**

      * Creates a new BatchedWrite.

      * 

      * @param builder

      *            The builder for the writes.

      */

     protected BatchedWrite(final Builder builder) {

         myWrites = Collections.unmodifiableList(new ArrayList<WriteOperation>(

                 builder.myWrites));

         myMode = builder.myMode;

         myDurability = builder.myDurability;

     }

     /**

      * Creates a new BatchedWrite.

      * 

      * @param ops

      *            The operations for the batch.

      * @param mode

      *            The mode for the batch.

      * @param durability

      *            The durability for the batch.

      */

     private BatchedWrite(final List<WriteOperation> ops,

             final BatchedWriteMode mode, final Durability durability) {

         myWrites = Collections.unmodifiableList(ops);

         myMode = mode;

         myDurability = durability;

     }

     /**

      * Creates a new BatchedWrite.

      * 

      * @param op

      *            The single operation for the batch.

      * @param mode

      *            The mode for the batch.

      * @param durability

      *            The durability for the batch.

      */

     private BatchedWrite(final WriteOperation op, final BatchedWriteMode mode,

             final Durability durability) {

         this(Collections.singletonList(op), mode, durability);

     }

     /**

      * Returns the durability for the writes.

      * 

      * @return The durability for the writes.

      */

     public Durability getDurability() {

         return myDurability;

     }

     /**

      * Returns the mode for submitting the writes to the server.

      * 

      * @return The mode for submitting the writes to the server.

      */

     public BatchedWriteMode getMode() {

         return myMode;

     }

     /**

      * Returns the writes to submit to the server.

      * 

      * @return The writes to submit to the server.

      */

     public List<WriteOperation> getWrites() {

         return myWrites;

     }

     /**

      * Creates write commands for all of the insert, updates and deletes. The

      * number and order of the writes is based on the {@link #getMode() mode}.

      * 

      * @param collectionName

      *            The name of the collection the documents will be inserted

      *            into.

      * @param maxCommandSize

      *            The maximum document size.

      * @param maxOperationsPerBundle

      *            The maximum number of writes to include in each bundle.

      * @return The list of command documents to be sent.

      */

     public List<Bundle> toBundles(final String collectionName,

             final long maxCommandSize, final int maxOperationsPerBundle) {

         switch (getMode()) {

         case REORDERED: {

             return createOptimized(collectionName, maxCommandSize,

                     maxOperationsPerBundle);

         }

         case SERIALIZE_AND_CONTINUE: {

             return createSerialized(collectionName, maxCommandSize,

                     maxOperationsPerBundle, false);

         }

         default: {

             return createSerialized(collectionName, maxCommandSize,

                     maxOperationsPerBundle, true);

         }

         }

     }

     /**

      * Adds the document to the array of documents.

      * 

      * @param array

      *            The array to add the operation to.

      * @param operation

      *            The operation to add.

      */

     private void add(final ArrayBuilder array, final WriteOperation operation) {

         switch (operation.getType()) {

         case INSERT: {

             final InsertOperation insertOperation = (InsertOperation) operation;

             array.add(insertOperation.getDocument());

             break;

         }

         case UPDATE: {

             final UpdateOperation updateOperation = (UpdateOperation) operation;

             final DocumentBuilder update = array.push();

             update.add("q", updateOperation.getQuery());

             update.add("u", updateOperation.getUpdate());

             if (updateOperation.isUpsert()) {

                 update.add("upsert", true);

             }

             if (updateOperation.isMultiUpdate()) {

                 update.add("multi", true);

             }

             break;

         }

         case DELETE: {

             final DeleteOperation deleteOperation = (DeleteOperation) operation;

             array.push().add("q", deleteOperation.getQuery())

                     .add("limit", deleteOperation.isSingleDelete() ? 1 : 0);

             break;

         }

         }

     }

     /**

      * Adds the durability ('writeConcern') to the command document.

      * 

      * @param command

      *            The command document to add the durability to.

      * @param durability

      *            The durability to add. May be <code>null</code>.

      */

     private void addDurability(final DocumentBuilder command,

             final Durability durability) {

         if (durability != null) {

             final DocumentBuilder durabilityDoc = command.push("writeConcern");

             if (durability.equals(Durability.NONE)) {

                 durabilityDoc.add("w", 0);

             }

             else if (durability.equals(Durability.ACK)) {

                 durabilityDoc.add("w", 1);

             }

             else {

                 boolean first = true;

                 for (final Element part : durability.asDocument()) {

                     if (first) {

                         // The first element is "getlasterror".

                         first = false;

                     }

                     else {

                         durabilityDoc.add(part);

                     }

                 }

             }

         }

     }

     /**

      * Creates a {@link DocumentToLargeException} for the operation.

      * 

      * @param operation

      *            The large operation.

      * @param size

      *            The size of the operation.

      * @param maxCommandSize

      *            The maximum size of the operation.

      * @return The created exception.

      */

     private DocumentToLargeException createDocumentToLargeException(

             final WriteOperation operation, final int size,

             final int maxCommandSize) {

         Document doc = EmptyDocument.INSTANCE;

         switch (operation.getType()) {

         case INSERT: {

             final InsertOperation insertOperation = (InsertOperation) operation;

             doc = insertOperation.getDocument();

             break;

         }

         case UPDATE: {

             final UpdateOperation updateOperation = (UpdateOperation) operation;

             doc = updateOperation.getQuery();

             final Document update = updateOperation.getUpdate();

             if (doc.size() < update.size()) {

                 doc = update;

             }

             break;

         }

         case DELETE: {

             final DeleteOperation deleteOperation = (DeleteOperation) operation;

             doc = deleteOperation.getQuery();

             break;

         }

         }

         return new DocumentToLargeException(size, maxCommandSize, doc);

     }

     /**

      * Reorders the writes into as few write commands as possible.

      * <p>

      * <b>Note</b>: MongoDB gives a slightly larger document for the command (<a

      * href=

      * "https://github.com/mongodb/mongo/blob/master/src/mongo/bson/util/builder.h#L56"

      * >16K</a>). This is for the command overhead. We don't explicitly use the

      * overhead but we may end up implicitly using it in the case of a operation

      * that is just at or below maxCommandSize. For those cases we start the

      * 'head' map below with the full map. That allows the big operations to be

      * added to command documents of there own once the command overhead has

      * been factored in.

      * </p>

      * 

      * @param collectionName

      *            The name of the collection the documents will be inserted

      *            into.

      * @param maxCommandSize

      *            The maximum document size.

      * @param maxOperationsPerBundle

      *            The maximum number of writes to include in each bundle.

      * @return The list of command documents to be sent.

      */

     private List<Bundle> createOptimized(final String collectionName,

             final long maxCommandSize, final int maxOperationsPerBundle) {

         // Bucket the operations and sort by size.

         Map<WriteOperationType, SortedMap<Long, List<WriteOperation>>> operationsBuckets;

         operationsBuckets = new LinkedHashMap<WriteOperationType, SortedMap<Long, List<WriteOperation>>>();

         for (final WriteOperation writeOp : getWrites()) {

             SortedMap<Long, List<WriteOperation>> operations = operationsBuckets

                     .get(writeOp.getType());

             if (operations == null) {

                 operations = new TreeMap<Long, List<WriteOperation>>();

                 operationsBuckets.put(writeOp.getType(), operations);

             }

             final Long size = Long.valueOf(sizeOf(-1, writeOp));

             List<WriteOperation> list = operations.get(size);

             if (list == null) {

                 list = new LinkedList<WriteOperation>();

                 operations.put(size, list);

             }

             list.add(writeOp);

         }

         // Check if any operation is too big.

         final Long maxMessageSize = Long.valueOf(maxCommandSize + 1);

         for (final SortedMap<Long, List<WriteOperation>> operations : operationsBuckets

                 .values()) {

             if (!operations.tailMap(maxMessageSize).isEmpty()) {

                 final Long biggest = operations.lastKey();

                 final List<WriteOperation> operation = operations.get(biggest);

                 throw createDocumentToLargeException(operation.get(0),

                         biggest.intValue(), (int) maxCommandSize);

             }

         }

         // Now build commands packing the operations into a few messages as

         // possible.

         final List<Bundle> commands = new ArrayList<Bundle>();

         final List<WriteOperation> bundled = new ArrayList<WriteOperation>(

                 Math.min(maxOperationsPerBundle, myWrites.size()));

         final DocumentBuilder command = BuilderFactory.start();

         for (final Map.Entry<WriteOperationType, SortedMap<Long, List<WriteOperation>>> entry : operationsBuckets

                 .entrySet()) {

             final SortedMap<Long, List<WriteOperation>> operations = entry

                     .getValue();

             while (!operations.isEmpty()) {

                 final ArrayBuilder docs = start(entry.getKey(), collectionName,

                         false, command);

                 long remaining = maxCommandSize - command.build().size();

                 SortedMap<Long, List<WriteOperation>> head = operations;

                 int index = 0;

                 while (!head.isEmpty()

                         && (bundled.size() < maxOperationsPerBundle)) {

                     final Long biggest = head.lastKey();

                     final List<WriteOperation> bigOps = head.get(biggest);

                     final WriteOperation operation = bigOps.remove(0);

                     if (bigOps.isEmpty()) {

                         head.remove(biggest);

                     }

                     add(docs, operation);

                     bundled.add(operation);

                     remaining -= sizeOf(index, operation);

                     index += 1;

                     head = operations.headMap(Long.valueOf(remaining

                             - sizeOfIndex(index)));

                 }

                 commands.add(new Bundle(command.build(), bundled));

                 bundled.clear();

             }

         }

         return commands;

     }

     /**

      * Creates write commands for each sequence of insert, updates and deletes.

      * <p>

      * <b>Note</b>: MongoDB gives a slightly larger document for the command (<a

      * href=

      * "https://github.com/mongodb/mongo/blob/master/src/mongo/bson/util/builder.h#L56"

      * >16K</a>). This is for the command overhead. We don't explicitly use the

      * overhead but we may end up using it in the case of a operation that is

      * just at or below maxCommandSize. That is why we start the 'head' map

      * below with the full map. That allows those big operations to be added to

      * commands of there own once the command overhead has been factored in.

      * </p>

      * 

      * @param collectionName

      *            The name of the collection the documents will be inserted

      *            into.

      * @param maxCommandSize

      *            The maximum document size.

      * @param stopOnError

      *            If true then the ordered flag is set to true.

      * @param maxOperationsPerBundle

      *            The maximum number of writes to include in each bundle.

      * @return The list of command documents to be sent.

      */

     private List<Bundle> createSerialized(final String collectionName,

             final long maxCommandSize, final int maxOperationsPerBundle,

             final boolean stopOnError) {

         final List<Bundle> commands = new ArrayList<Bundle>();

         final DocumentBuilder command = BuilderFactory.start();

         final List<WriteOperation> toSend = getWrites();

         final List<WriteOperation> bundled = new ArrayList<WriteOperation>(

                 Math.min(maxOperationsPerBundle, myWrites.size()));

         ArrayBuilder opsArray = null;

         WriteOperationType lastType = null;

         long remaining = maxCommandSize;

         for (final WriteOperation writeOp : toSend) {

             long size = sizeOf(-1, writeOp);

             final long indexSize = sizeOfIndex(bundled.size());

             if (maxCommandSize < size) {

                 throw createDocumentToLargeException(writeOp, (int) size,

                         (int) maxCommandSize);

             }

             size += indexSize; // Add in the index overhead.

             // Close a command if change type or too big.

             if (!bundled.isEmpty()

                     && ((lastType != writeOp.getType())

                             || ((remaining - size) < 0) || (maxOperationsPerBundle <= bundled

                             .size()))) {

                 commands.add(new Bundle(command.build(), bundled));

                 bundled.clear();

             }

             // Start a command? - Maybe after closing?

             if (bundled.isEmpty()) {

                 opsArray = start(writeOp.getType(), collectionName,

                         stopOnError, command);

                 lastType = writeOp.getType();

                 remaining = (maxCommandSize - command.build().size());

             }

             // Add the operation.

             add(opsArray, writeOp);

             bundled.add(writeOp);

             // Remove the size of the operation from the remaining.

             remaining -= size;

         }

         if (!bundled.isEmpty()) {

             commands.add(new Bundle(command.build(), bundled));

         }

         return commands;

     }

     /**

      * Returns the size of the encoded operation.

      * <p>

      * For an {@code InsertOperation} this is the size of the document to

      * insert.

      * <p>

      * For an {@code UpdateOperation} this includes the space for:

      * <dl>

      * <dt>Document Overhead</dt>

      * <dd>type (1 byte), length (4 bytes), trailing null (1 byte).</dd>

      * <dt>'q' field</dt>

      * <dd>name (2 bytes), type (1 byte), value (document size)</dd>

      * <dt>'u' field</dt>

      * <dd>name (2 bytes), type (1 byte), value (document size)</dd>

      * <dt>'upsert' field</dt>

      * <dd>name (7 bytes), type (1 byte), value (1 byte)</dd>

      * <dt>'multi' field</dt>

      * <dd>name (6 bytes), type (1 byte), value (1 byte)</dd>

      * </dl>

      * </p>

      * <p>

      * For a {@code DeleteOperation} this includes the space for:

      * <dl>

      * <dt>Document Overhead</dt>

      * <dd>type (1 byte), length (4 bytes), trailing null (1 byte).</dd>

      * <dt>'q' field</dt>

      * <dd>name (2 bytes), type (1 byte), value (document size)</dd>

      * <dt>'limit' field</dt>

      * <dd>name (6 bytes), type (1 byte), value (4 bytes)</dd>

      * </dl>

      * 

      * @param index

      *            The index of the operation in the operations array.

      * @param operation

      *            The operation to determine the size of.

      * @return The size of the operation.

      */

     private long sizeOf(final int index, final WriteOperation operation) {

         long result = 0;

         switch (operation.getType()) {

         case INSERT: {

             final InsertOperation insertOperation = (InsertOperation) operation;

             result = sizeOfIndex(index) + insertOperation.getDocument().size();

             break;

         }

         case UPDATE: {

             final UpdateOperation updateOperation = (UpdateOperation) operation;

             result = sizeOfIndex(index) + updateOperation.getQuery().size()

                     + updateOperation.getUpdate().size() + 29;

             break;

         }

         case DELETE: {

             final DeleteOperation deleteOperation = (DeleteOperation) operation;

             result = sizeOfIndex(index) + deleteOperation.getQuery().size()

                     + 20;

             break;

         }

         }

         return result;

     }

     /**

      * Returns the number of bytes required to encode the index within the array

      * element.

      * 

      * @param index

      *            The index to return the size of.

      * @return The length of the encoded index.

      */

     private long sizeOfIndex(final int index) {

         // For 2.6 the number of items in the array is capped at 1000. This

         // allows up to 99,999 without resorting to turning the value into

         // a string which seems like safe enough padding.

         if (index < 0) {

             return 0; // For estimating operation sizes.

         }

         else if (index < 10) {

             return 3; // single character plus a null plus a type.

         }

         else if (index < 100) {

             return 4; // two characters plus a null plus a type.

         }

         else if (index < 1000) {

             return 5; // three characters plus a null plus a type.

         }

         else if (index < 10000) {

             return 6; // four characters plus a null plus a type.

         }

         return Integer.toString(index).length() + 2;

     }

     /**

      * Starts a new command document.

      * 

      * @param operation

      *            The operation to start.

      * @param collectionName

      *            The collection to operate on.

      * @param stopOnError

      *            If true then the operations should stop once an error is

      *            encountered. Is mapped to the {@code ordered} field in the

      *            command document.

      * @param command

      *            The command builder.

      * @return The {@link ArrayBuilder} for the operations array.

      */

     private ArrayBuilder start(final WriteOperationType operation,

             final String collectionName, final boolean stopOnError,

             final DocumentBuilder command) {

         String commandName = "";

         String arrayName = "";

         switch (operation) {

         case INSERT: {

             commandName = "insert";

             arrayName = "documents";

             break;

         }

         case UPDATE: {

             commandName = "update";

             arrayName = "updates";

             break;

         }

         case DELETE: {

             commandName = "delete";

             arrayName = "deletes";

             break;

         }

         }

         command.reset();

         command.add(commandName, collectionName);

         if (!stopOnError) {

             command.add("ordered", stopOnError);

         }

         addDurability(command, getDurability());

         return command.pushArray(arrayName);

     }

     /**

      * Builder for creating {@link BatchedWrite}s.

      * 

      * @api.yes This class is part of the driver's API. Public and protected

      *          members will be deprecated for at least 1 non-bugfix release

      *          (version numbers are <major>.<minor>.<bugfix>)

      *          before being removed or modified.

      * @copyright 2012-2013, Allanbank Consulting, Inc., All Rights Reserved

      */

     public static class Builder {

         /** The durability for the writes. */

         protected Durability myDurability;

         /** The mode for submitting the writes to the server. */

         protected BatchedWriteMode myMode;

         /** The writes to submit to the server. */

         protected final List<WriteOperation> myWrites;

         /**

          * Creates a new Builder.

          */

         public Builder() {

             myWrites = new ArrayList<WriteOperation>();

             reset();

         }

         /**

          * Constructs a new {@link BatchedWrite} object from the state of the

          * builder.

          * 

          * @return The new {@link BatchedWrite} object.

          */

         public BatchedWrite build() {

             return new BatchedWrite(this);

         }

         /**

          * Update a document based on a query.

          * <p>

          * Defaults to deleting as many documents as match the query.

          * </p>

          * <p>

          * This method is delegates to

          * {@link #delete(DocumentAssignable, boolean) delete(query, false)}

          * </p>

          * 

          * @param query

          *            The query to find the document to delete.

          * @return This builder for chaining method calls.

          */

         public Builder delete(final DocumentAssignable query) {

             return delete(query, false);

         }

         /**

          * Update a document based on a query.

          * <p>

          * Defaults to deleting as many documents as match the query.

          * </p>

          * 

          * @param query

          *            The query to find the document to delete.

          * @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.

          * @return This builder for chaining method calls.

          */

         public Builder delete(final DocumentAssignable query,

                 final boolean singleDelete) {

             return write(new DeleteOperation(query, singleDelete));

         }

         /**

          * Sets the durability for the writes.

          * <p>

          * This method delegates to {@link #setDurability(Durability)}.

          * </p>

          * 

          * @param durability

          *            The new value for the durability for the writes.

          * @return This builder for chaining method calls.

          */

         public Builder durability(final Durability durability) {

             return setDurability(durability);

         }

         /**

          * Returns the durability for the write.

          * 

          * @return This durability for the write.

          */

         public Durability getDurability() {

             return myDurability;

         }

         /**

          * Adds an insert operation to the batched write.

          * 

          * @param document

          *            The document to insert.

          * @return This builder for chaining method calls.

          */

         public Builder insert(final DocumentAssignable document) {

             return write(new InsertOperation(document));

         }

         /**

          * Sets the mode for submitting the writes to the server.

          * <p>

          * This method delegates to {@link #setMode(BatchedWriteMode)}.

          * </p>

          * 

          * @param mode

          *            The new value for the mode for submitting the writes to

          *            the server.

          * @return This builder for chaining method calls.

          */

         public Builder mode(final BatchedWriteMode mode) {

             return setMode(mode);

         }

         /**

          * Resets the builder back to its initial state for reuse.

          * 

          * @return This builder for chaining method calls.

          */

         public Builder reset() {

             myWrites.clear();

             myMode = BatchedWriteMode.SERIALIZE_AND_CONTINUE;

             myDurability = null;

             return this;

         }

         /**

          * Saves the {@code document} to MongoDB.

          * <p>

          * If the {@code document} does not contain an {@code _id} field then

          * this method is equivalent to: {@link #insert(DocumentAssignable)

          * insert(document)}.

          * </p>

          * <p>

          * If the {@code document} does contain an {@code _id} field then this

          * method is equivalent to:

          * {@link #update(DocumentAssignable, DocumentAssignable)

          * updateAsync(BuilderFactory.start().add(document.get("_id")),

          * document, false, true)}.

          * </p>

          * 

          * @param document

          *            The document to save.

          * @return This builder for chaining method calls.

          */

         public Builder save(final DocumentAssignable document) {

             final Document doc = document.asDocument();

             final Element id = doc.get("_id");

             if (id == null) {

                 return insert(doc);

             }

             return update(BuilderFactory.start().add(id), doc, false, true);

         }

         /**

          * Sets the durability for the writes.

          * 

          * @param durability

          *            The new value for the durability for the writes.

          * @return This builder for chaining method calls.

          */

         public Builder setDurability(final Durability durability) {

             myDurability = durability;

             return this;

         }

         /**

          * Sets the mode for submitting the writes to the server.

          * 

          * @param mode

          *            The new value for the mode for submitting the writes to

          *            the server.

          * @return This builder for chaining method calls.

          */

         public Builder setMode(final BatchedWriteMode mode) {

             myMode = mode;

             return this;

         }

         /**

          * Sets the writes to submit to the server.

          * 

          * @param writes

          *            The new value for the writes to submit to the server.

          * @return This builder for chaining method calls.

          */

         public Builder setWrites(final List<WriteOperation> writes) {

             myWrites.clear();

             if (writes != null) {

                 myWrites.addAll(writes);

             }

             return this;

         }

         /**

          * Update a document based on a query.

          * <p>

          * Defaults to updating a single document and not performing an upsert

          * if no document is found.

          * </p>

          * <p>

          * This method is delegates to

          * {@link #update(DocumentAssignable, DocumentAssignable, boolean, boolean)

          * update(query, update, false, false)}

          * </p>

          * 

          * @param query

          *            The query to find the document to update.

          * @param update

          *            The update operations to apply to the document.

          * @return This builder for chaining method calls.

          */

         public Builder update(final DocumentAssignable query,

                 final DocumentAssignable update) {

             return update(query, update, false, false);

         }

         /**

          * Update a document based on a query.

          * <p>

          * Defaults to updating a single document and not performing an upsert

          * if no document is found.

          * </p>

          * 

          * @param query

          *            The query to find the document to update.

          * @param update

          *            The update operations to apply to the document.

          * @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.

          * @return This builder for chaining method calls.

          */

         public Builder update(final DocumentAssignable query,

                 final DocumentAssignable update, final boolean multiUpdate,

                 final boolean upsert) {

             return write(new UpdateOperation(query, update, multiUpdate, upsert));

         }

         /**

          * Adds a single write to the list of writes to send to the server.

          * 

          * @param write

          *            The write to add to the list of writes to send to the

          *            server.

          * @return This builder for chaining method calls.

          */

         public Builder write(final WriteOperation write) {

             myWrites.add(write);

             return this;

         }

         /**

          * Sets the writes to submit to the server.

          * <p>

          * This method delegates to {@link #setWrites(List)}.

          * </p>

          * 

          * @param writes

          *            The new value for the writes to submit to the server.

          * @return This builder for chaining method calls.

          */

         public Builder writes(final List<WriteOperation> writes) {

             return setWrites(writes);

         }

     }

     /**

      * Bundle is a container for the write command and the

      * {@link WriteOperation} it contains.

      * 

      * @api.yes This class is part of the driver's API. Public and protected

      *          members will be deprecated for at least 1 non-bugfix release

      *          (version numbers are <major>.<minor>.<bugfix>)

      *          before being removed or modified.

      */

     public static final class Bundle {

         /** The command containing the bundled write operations. */

         private final Document myCommand;

         /** The writes that are bundled in the command. */

         private final List<WriteOperation> myWrites;

         /**

          * Creates a new Bundle.

          * 

          * @param command

          *            The command containing the bundled write operations.

          * @param writes

          *            The writes that are bundled in the command.

          */

         protected Bundle(final Document command,

                 final List<WriteOperation> writes) {

             super();

             myCommand = command;

             myWrites = Collections

                     .unmodifiableList(new ArrayList<WriteOperation>(writes));

         }

         /**

          * Returns the command containing the bundled write operations.

          * 

          * @return The command containing the bundled write operations.

          */

         public Document getCommand() {

             return myCommand;

         }

         /**

          * Returns the writes that are bundled in the command.

          * 

          * @return The writes that are bundled in the command.

          */

         public List<WriteOperation> getWrites() {

             return myWrites;

         }

     }

 }