package com.lowtuna.jsonblob.core;

import com.google.common.base.Stopwatch;
import com.mongodb.BasicDBObject;
import com.mongodb.BasicDBObjectBuilder;
import com.mongodb.DBCursor;
import com.mongodb.DBObject;
import com.mongodb.MongoException;
import lombok.extern.slf4j.Slf4j;
import org.bson.types.ObjectId;
import org.joda.time.DateTime;
import org.joda.time.DateTimeZone;

import java.util.Date;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Created by tburch on 11/15/16.
 */
@Slf4j
public class BlobMigrationJob implements Runnable {

  private final MongoDbJsonBlobManager mongoDbJsonBlobManager;
  private final FileSystemJsonBlobManager fileSystemJsonBlobManager;
  private final ExecutorService executorService = Executors.newFixedThreadPool(50);

  public BlobMigrationJob(MongoDbJsonBlobManager mongoDbJsonBlobManager, FileSystemJsonBlobManager fileSystemJsonBlobManager) {
    this.mongoDbJsonBlobManager = mongoDbJsonBlobManager;
    this.fileSystemJsonBlobManager = fileSystemJsonBlobManager;
  }

  @Override
  public void run() {
    Stopwatch stopwatch = new Stopwatch();
    stopwatch.start();
    final AtomicInteger migratedBlobs = new AtomicInteger(0);

    log.info("Starting blob migration");
    int limit = 398;
    final CountDownLatch latch = new CountDownLatch(limit);

    List<DBObject> curs = mongoDbJsonBlobManager.getCollection().find(new BasicDBObject(), new BasicDBObject()).limit(limit).toArray();
    try {
      while (curs.iterator().hasNext()) {
          DBObject o = curs.iterator().next();
          ObjectId id = (ObjectId) o.get(MongoDbJsonBlobManager.ID_ATTR_NAME);
          try {
            final String json = mongoDbJsonBlobManager.getBlob(id.toString());
            executorService.submit(new Runnable() {
              @Override
              public void run() {
                int completed = migratedBlobs.incrementAndGet();
                try {
                  log.trace("Migrating blob {}", id);
                  fileSystemJsonBlobManager.createBlob(json, id.toString());
                  log.trace("Completed migrating blob {}", id);
                } catch (MongoException e) {
                  log.warn("Error while migrating blob", e);
                } catch (Exception e) {
                  log.warn("Caught exception while migrating blob", e);
                } finally {
                  latch.countDown();
                  if (completed % 25 == 0) {
                    log.info("Migrated {} blobs... (~{} per second)", completed, completed / stopwatch.elapsed(TimeUnit.SECONDS));
                  }
                  try {
                    mongoDbJsonBlobManager.deleteBlob(id.toString());
                  } catch (BlobNotFoundException e) {
                    //this shouldn't happen...
                  }
                }
              }
            });
          } catch (NullPointerException e) {
            mongoDbJsonBlobManager.deleteBlob(id.toString());
          }
      }
      latch.await(1, TimeUnit.MINUTES);
    } catch (Exception e) {
      log.warn("Caught exception while migrating blobs", e);
//    } finally {
//      curs.close();
    }
    log.info("Completed migrating {} blobs in {}ms", migratedBlobs, stopwatch.elapsed(TimeUnit.MILLISECONDS));
  }
}

@@ -398,31 +398,8 @@
 te;%0A
-import java.util.List;%0A
 impo
@@ -1479,22 +1479,16 @@
     
-List%3CDBObject%3E
+DBCursor
  cur
@@ -1593,18 +1593,8 @@
 mit)
-.toArray()
 ;%0A  
@@ -1619,26 +1619,47 @@
 e (curs.
-iterator()
+hasNext()) %7B%0A          if (curs
 .hasNext
@@ -1656,32 +1656,34 @@
 rs.hasNext()) %7B%0A
+  
           DBObje
@@ -1698,19 +1698,8 @@
 urs.
-iterator().
 next
@@ -1712,16 +1712,18 @@
         
+  
 ObjectId
@@ -1785,32 +1785,34 @@
 AME);%0A          
+  
 try %7B%0A          
@@ -1809,24 +1809,26 @@
             
+ 
+ 
 final String
@@ -1894,16 +1894,18 @@
         
+  
 executor
@@ -1932,24 +1932,26 @@
 unnable() %7B%0A
+  
             
@@ -1976,16 +1976,18 @@
         
+  
 public v
@@ -1994,24 +1994,26 @@
 oid run() %7B%0A
+  
             
@@ -2073,38 +2073,42 @@
                 
+ 
+ 
 try %7B%0A
+  
                 
@@ -2163,16 +2163,18 @@
         
+ 
+ 
 fileSyst
@@ -2234,32 +2234,34 @@
                 
+  
 log.trace(%22Compl
@@ -2282,32 +2282,34 @@
  blob %7B%7D%22, id);%0A
+  
                 
@@ -2347,32 +2347,34 @@
                 
+  
 log.warn(%22Error 
@@ -2392,32 +2392,34 @@
 ting blob%22, e);%0A
+  
                 
@@ -2452,32 +2452,34 @@
                 
+  
 log.warn(%22Caught
@@ -2528,24 +2528,26 @@
             
+  
 %7D finally %7B%0A
@@ -2560,24 +2560,26 @@
             
+ 
+ 
 latch.countD
@@ -2581,24 +2581,26 @@
 ountDown();%0A
+  
             
@@ -2648,24 +2648,26 @@
             
+ 
+ 
 log.info(%22Mi
@@ -2775,34 +2775,38 @@
                 
-%7D%0A
+  %7D%0A  
                 
@@ -2825,32 +2825,34 @@
                 
+  
 mongoDbJsonBlobM
@@ -2895,32 +2895,34 @@
                 
+  
 %7D catch (BlobNot
@@ -2961,16 +2961,18 @@
         
+ 
+ 
 //this s
@@ -3000,34 +3000,38 @@
                 
-%7D%0A
+  %7D%0A  
                 
@@ -3038,34 +3038,38 @@
 %7D%0A              
-%7D%0A
+  %7D%0A  
             %7D);%0A
@@ -3070,32 +3070,34 @@
   %7D);%0A          
+  
 %7D catch (NullPoi
@@ -3107,32 +3107,34 @@
 rException e) %7B%0A
+  
             mong
@@ -3171,32 +3171,46 @@
 id.toString());%0A
+            %7D%0A
           %7D%0A    
@@ -3342,18 +3342,16 @@
 s%22, e);%0A
-//
     %7D fi
@@ -3358,18 +3358,16 @@
 nally %7B%0A
-//
       cu

Revert "use an array"