/*
    * #%L
    * BatchedNativeWriteCallback.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.callback;
  
  import java.util.ArrayList;
  import java.util.Collections;
  import java.util.IdentityHashMap;
  import java.util.LinkedList;
  import java.util.List;
  import java.util.Map;
  
  import com.allanbank.mongodb.Callback;
  import com.allanbank.mongodb.Durability;
  import com.allanbank.mongodb.builder.BatchedWrite;
  import com.allanbank.mongodb.builder.BatchedWriteMode;
  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.client.AbstractMongoOperations;
  import com.allanbank.mongodb.error.BatchedWriteException;
  
  /**
   * BatchedWriteCallback provides the global callback for the batched writes when
   * the server does not support the write commands. This callback will issue the
   * writes using the original wire protocol.
   * 
   * @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 class BatchedNativeWriteCallback extends ReplyLongCallback {
  
      /** The collection to send individual operations with. */
      private final AbstractMongoOperations myCollection;
  
      /** The list of write operations which failed. */
      private final Map<WriteOperation, Throwable> myFailedOperations;
  
      /** The count of finished bundles or operations. */
      private int myFinished;
  
      /** The result. */
      private long myN = 0;
  
      /** The list of write operations to send. */
      private final List<WriteOperation> myOperations;
  
      /** The list of write operations waiting to be sent to the server. */
      private final List<WriteOperation> myPendingOperations;
  
      /** The original write operation. */
      private final BatchedWrite myWrite;
  
      /**
       * Creates a new BatchedWriteCallback.
       * 
       * @param results
       *            The callback for the final results.
       * @param write
       *            The original write.
       * @param collection
       *            The collection for sending the operations.
       * @param operations
       *            The operations to send.
       */
      public BatchedNativeWriteCallback(final Callback<Long> results,
              final BatchedWrite write, final AbstractMongoOperations collection,
              final List<WriteOperation> operations) {
          super(results);
  
          myWrite = write;
          myCollection = collection;
          myOperations = Collections
                  .unmodifiableList(new ArrayList<WriteOperation>(operations));
  
          myPendingOperations = new LinkedList<WriteOperation>(myOperations);
  
          myFinished = 0;
          myN = 0;
  
         myFailedOperations = new IdentityHashMap<WriteOperation, Throwable>();
     }
 
     /**
      * Sends the next set of operations to the server.
      */
     public void send() {
 
         List<WriteOperation> toSendOperations;
         Durability durability = null;
 
         synchronized (this) {
             List<WriteOperation> toSend = myPendingOperations;
             if (BatchedWriteMode.SERIALIZE_AND_STOP.equals(myWrite.getMode())) {
                 toSend = myPendingOperations.subList(0, 1);
             }
 
             durability = myWrite.getDurability();
             if ((durability == null) || (durability == Durability.NONE)) {
                 durability = Durability.ACK;
             }
 
             // Clear toSend before sending so the callbacks see the right
             // state for the operations.
             toSendOperations = new ArrayList<WriteOperation>(toSend);
             toSend.clear();
         } // Release lock.
 
         // Release the lock before sending to avoid deadlock in processing
         // replies.
 
         for (final WriteOperation operation : toSendOperations) {
             switch (operation.getType()) {
             case INSERT: {
                 final InsertOperation insert = (InsertOperation) operation;
                 myCollection.insertAsync(new NativeCallback<Integer>(insert),
                         true, durability, insert.getDocument());
                 break;
             }
             case UPDATE: {
                 final UpdateOperation update = (UpdateOperation) operation;
                 myCollection.updateAsync(new NativeCallback<Long>(operation),
                         update.getQuery(), update.getUpdate(),
                         update.isMultiUpdate(), update.isUpsert(), durability);
                 break;
             }
             case DELETE: {
                 final DeleteOperation delete = (DeleteOperation) operation;
                 myCollection.deleteAsync(new NativeCallback<Long>(operation),
                         delete.getQuery(), delete.isSingleDelete(), durability);
                 break;
             }
             }
         }
     }
 
     /**
      * Callback for a single write operation sent via the native messages.
      * 
      * @param operation
      *            The write operation.
      * @param result
      *            The result of the write operation.
      */
     protected synchronized void callback(final WriteOperation operation,
             final long result) {
         myN += result;
         myFinished += 1;
 
         if (!myPendingOperations.isEmpty()) {
             send();
         }
         else if (myFinished == myOperations.size()) {
             publishResults();
         }
     }
 
     /**
      * Callback for a single write operation sent via the native messages has
      * failed.
      * 
      * @param operation
      *            The write operation.
      * @param thrown
      *            The error for the operation.
      */
     protected synchronized void exception(final WriteOperation operation,
             final Throwable thrown) {
         myFinished += 1;
         myFailedOperations.put(operation, thrown);
 
         if (myWrite.getMode() == BatchedWriteMode.SERIALIZE_AND_STOP) {
             publishResults();
         }
         // No need to check if we have to send. Would have already sent all of
         // the operations if not SERIALIZE_AND_STOP.
         else if (myFinished == myOperations.size()) {
             publishResults();
         }
 
     }
 
     /**
      * Publishes the final results.
      */
     private void publishResults() {
         if (myFailedOperations.isEmpty()) {
             myForwardCallback.callback(Long.valueOf(myN));
         }
         else {
             myForwardCallback.exception(new BatchedWriteException(myWrite, myN,
                     myPendingOperations, myFailedOperations));
         }
     }
 
     /**
      * NativeCallback provides the callback for a single operation within a
      * batched write. This callback is used when the batched write commands are
      * not supported and the driver falls back to using the native insert,
      * update, and delete messages.
      * 
      * @param <T>
      *            The type for the callback. Expected to be either Integer or
      *            Long.
      * 
      * @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
      */
     /* package */class NativeCallback<T extends Number> implements Callback<T> {
 
         /** The operation this callback is waiting for the reply from. */
         private final WriteOperation myOperation;
 
         /**
          * Creates a new BatchedWriteNativeCallback.
          * 
          * @param operation
          *            The operation this callback is waiting for the reply from.
          */
         public NativeCallback(final WriteOperation operation) {
             myOperation = operation;
         }
 
         /**
          * {@inheritDoc}
          * <p>
          * Overridden to forward the results to the parent callback.
          * </p>
          */
         @Override
         public void callback(final T result) {
             BatchedNativeWriteCallback.this.callback(myOperation,
                     result.longValue());
         }
 
         /**
          * {@inheritDoc}
          * <p>
          * Overridden to forward the error to the parent callback.
          * </p>
          */
         @Override
         public void exception(final Throwable thrown) {
             BatchedNativeWriteCallback.this.exception(myOperation, thrown);
         }
     }
 }