package com.atlassian.bonnie.index;
   
   import java.io.IOException;
   import java.util.concurrent.BlockingQueue;
   import java.util.concurrent.LinkedBlockingQueue;
   
   import org.apache.lucene.analysis.Analyzer;
   import org.apache.lucene.analysis.standard.StandardAnalyzer;
   import org.apache.lucene.index.IndexReader;
  import org.apache.lucene.index.IndexWriter;
  import org.apache.lucene.index.Term;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import com.atlassian.bonnie.ILuceneConnection;
  import com.atlassian.bonnie.LuceneException;
  import com.atlassian.core.util.ProgressWrapper;
  
  /**
   * Performs batch indexing in multiple threads.
   * <p>This class is designed for:
   * <ul>
   * <li>an online environment where changes are being made to the index as batch reindexing occurs</li>
   * <li>indexing large files</li>
   * </ul>
   * Each thread writes to its own temporary index and when all threads finish, these are merged into the final index.
   * This has the desirable effect where a few large files do not bottleneck the entire operation.
   * </p>
   */
  public class OnlineMultiThreadedIndexer extends BaseMultiThreadedIndexer implements SingleObjectIndexer
  {
      private static final Logger log = LoggerFactory.getLogger(OnlineMultiThreadedIndexer.class);
      protected BlockingQueue<? super Object> reindexAddedQueue = new LinkedBlockingQueue<Object>();
      protected BlockingQueue<? super Object> reindexDeletedQueue = new LinkedBlockingQueue<Object>();
      private ObjectToDocumentConverter objectToDocumentConverter;
  
      protected void allThreadsComplete(final DocumentWritingScheme scheme, final boolean truncate, final ProgressWrapper progress)
      {
          luceneConnection.withBatchUpdate(new ILuceneConnection.BatchUpdateAction()
          {
              public void perform() throws Exception
              {
                  setReindexingStatus(true);
  
                  if (truncate)
                  {
                      progress.setStatus("Truncating index");
                      truncateIndex();
                  }
                  progress.setStatus("Closing index");
                  scheme.close(luceneConnection);
                  progress.setStatus("Flushing queues");
                  if (flushReindexingQueues())
                  {
                      progress.setStatus("Optimizing index");
  
                      optimize(luceneConnection);
                  }
  
                  setReindexingStatus(false);
              }
          });
      }
  
      public void index(final Object o)
      {
          luceneConnection.withBatchUpdate(new ILuceneConnection.BatchUpdateAction()
          {
              public void perform()
              {
                  if (isReindexing())
                  {
                      try
                      {
                          if (log.isDebugEnabled())
                              log.debug("Adding object:" + o + " to reindexaddedqueue");
                          reindexAddedQueue.put(o);
                      }
                      catch (InterruptedException e)
                      {
                          log.error("Error encountered adding object to reindexaddedqueue", e);
                      }
                  }
                  unindex(o);
                  DocumentWritingScheme scheme = getDocumentWritingScheme(false);
                  Runnable r = new QueueProcessingRunnableImpl(new SingletonObjectQueue(o, objectToDocumentConverter), scheme);
                  r.run();
              }
          });
      }
  
      /**
       * Performs the actual adding of the object to the index.
       * @param o
       * @param writer
       * @throws IOException
       */
      protected void doAdd(final Object o, IndexWriter writer) throws IOException
      {
         writer.addDocument(objectToDocumentConverter.convert(o, null));
     }
 
     public void unindex(final Object o)
     {
         luceneConnection.withBatchUpdate(new ILuceneConnection.BatchUpdateAction()
         {
             public void perform()
             {
                 if (isReindexing())
                 {
                     try
                     {
                         if (log.isDebugEnabled())
                             log.debug("Adding object:" + o + " to reindexdeletedqueue");
                         reindexDeletedQueue.put(o);
                     }
                     catch (InterruptedException e)
                     {
                         log.error("Error encountered adding object to reindexdeletedqueue", e);
                     }
                 }
                 luceneConnection.withReaderAndDeletes(new ILuceneConnection.ReaderAction()
                 {
                     public Object perform(IndexReader reader) throws IOException
                     {
                         doDelete(o, reader);
                         return null;
                     }
                 });
             }
         });
     }
 
     /**
      * Perform the actual deleting.
      * @param o object to delete
      * @param reader reader
      * @throws IOException
      */
     protected void doDelete(Object o, IndexReader reader) throws IOException
     {
         final String[] identity = objectToDocumentConverter.getObjectIdentity(o);
         reader.deleteDocuments(new Term(identity[0], identity[1]));
     }
 
     /**
      * Optimize the index.
      * @param conn
      */
     protected final void optimize(ILuceneConnection conn)
     {
         conn.withWriter(new ILuceneConnection.WriterAction()
         {
             public void perform(IndexWriter writer) throws IOException
             {
                 writer.optimize();
             }
         });
     }
 
     /**
      * Synchronize the index with whatever happened since it started reindexing.
      * @return true if the index was updated, false otherwise
      */
     private boolean flushReindexingQueues()
     {
         boolean somethingflushed = false;
         // flush reindexing queues (delete before adding!)
         if (!reindexDeletedQueue.isEmpty())
         {
             luceneConnection.withReaderAndDeletes(new ILuceneConnection.ReaderAction()
             {
                 public Object perform(IndexReader reader) throws IOException
                 {
                     try
                     {
                         for (Object o = reindexDeletedQueue.take(); !reindexDeletedQueue.isEmpty();)
                         {
                             doDelete(o, reader);
                             o = reindexDeletedQueue.take();
                         }
                     }
                     catch (InterruptedException e)
                     {
                         throw new LuceneException(e);
                     }
                     return null;
                 }
             });
             somethingflushed = true;
         }
         if (!reindexAddedQueue.isEmpty())
         {
             luceneConnection.withWriter(new ILuceneConnection.WriterAction()
             {
                 public void perform(IndexWriter writer) throws IOException
                 {
                     try
                     {
                         for (Object o = reindexAddedQueue.take(); !reindexAddedQueue.isEmpty();)
                         {
                             doAdd(o, writer);
                             o = reindexAddedQueue.take();
                         }
                     }
                     catch (InterruptedException e)
                     {
                         throw new LuceneException(e);
                     }
                 }
             });
             somethingflushed = true;
         }
         return somethingflushed;
     }
 
 
     /**
      * Creates a new instance of a DocumentWritingScheme.
      *
      * @param reindex  is this a reindex
      */
     protected DocumentWritingScheme getDocumentWritingScheme(boolean reindex)
     {
         if (reindex)
             return new TempDirectoryDocumentWritingScheme(this);
         else return new SingleDocumentWritingScheme(luceneConnection);
     }
 
 
     public void setLuceneConnection(ILuceneConnection luceneConnection)
     {
         this.luceneConnection = luceneConnection;
     }
 
 
     protected ObjectToDocumentConverter getObjectToDocumentConverter()
     {
         return objectToDocumentConverter;
     }
 
     public void setObjectToDocumentConverter(ObjectToDocumentConverter objectToDocumentConverter)
     {
         this.objectToDocumentConverter = objectToDocumentConverter;
     }
 
     public Analyzer getAnalyzer()
     {
         return new StandardAnalyzer();
     }
 
     protected void setReindexingStatus(boolean status)
             throws IOException
     {
         setReindexing(status);
     }
 }