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.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);

    DBCursor curs = mongoDbJsonBlobManager.getCollection().find(new BasicDBObject(), new BasicDBObject()).limit(limit);
    try {
      while (curs.hasNext()) {
          if (curs.hasNext()) {
            DBObject o = curs.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));
  }
}

@@ -394,16 +394,39 @@
 l.Date;%0A
+import java.util.List;%0A
 import j
@@ -1498,24 +1498,30 @@
 );%0A%0A    
-DBCursor
+List%3CDBObject%3E
  curs = 
@@ -1618,16 +1618,26 @@
 t(limit)
+.toArray()
 ;%0A    tr
@@ -1662,39 +1662,18 @@
 urs.
-hasNext()) %7B%0A          if (curs
+iterator()
 .has
@@ -1678,26 +1678,24 @@
 asNext()) %7B%0A
-  
           DB
@@ -1710,16 +1710,27 @@
  = curs.
+iterator().
 next();%0A
@@ -1739,18 +1739,16 @@
         
-  
 ObjectId
@@ -1810,34 +1810,32 @@
 AME);%0A          
-  
 try %7B%0A          
@@ -1836,18 +1836,16 @@
         
- 
- 
 final St
@@ -1915,18 +1915,16 @@
         
-  
 executor
@@ -1951,26 +1951,24 @@
 unnable() %7B%0A
-  
             
@@ -1993,18 +1993,16 @@
         
-  
 public v
@@ -2009,26 +2009,24 @@
 oid run() %7B%0A
-  
             
@@ -2090,34 +2090,30 @@
             
- 
- 
 try %7B%0A
-  
             
@@ -2172,18 +2172,16 @@
         
- 
- 
 fileSyst
@@ -2241,34 +2241,32 @@
                 
-  
 log.trace(%22Compl
@@ -2287,34 +2287,32 @@
  blob %7B%7D%22, id);%0A
-  
                 
@@ -2350,34 +2350,32 @@
                 
-  
 log.warn(%22Error 
@@ -2393,34 +2393,32 @@
 ting blob%22, e);%0A
-  
                 
@@ -2451,34 +2451,32 @@
                 
-  
 log.warn(%22Caught
@@ -2529,18 +2529,16 @@
         
-  
 %7D finall
@@ -2555,26 +2555,24 @@
             
- 
- 
 latch.countD
@@ -2574,26 +2574,24 @@
 ountDown();%0A
-  
             
@@ -2639,26 +2639,24 @@
             
- 
- 
 log.info(%22Mi
@@ -2764,38 +2764,34 @@
                 
-  %7D%0A  
+%7D%0A
                 
@@ -2810,34 +2810,32 @@
                 
-  
 mongoDbJsonBlobM
@@ -2878,34 +2878,32 @@
                 
-  
 %7D catch (BlobNot
@@ -2942,18 +2942,16 @@
         
- 
- 
 //this s
@@ -2979,38 +2979,34 @@
                 
-  %7D%0A  
+%7D%0A
                 
@@ -3013,38 +3013,34 @@
 %7D%0A              
-  %7D%0A  
+%7D%0A
             %7D);%0A
@@ -3041,34 +3041,32 @@
   %7D);%0A          
-  
 %7D catch (NullPoi
@@ -3076,34 +3076,32 @@
 rException e) %7B%0A
-  
             mong
@@ -3146,30 +3146,16 @@
 ing());%0A
-            %7D%0A
         
@@ -3291,24 +3291,26 @@
 blobs%22, e);%0A
+//
     %7D finall
@@ -3309,24 +3309,26 @@
 %7D finally %7B%0A
+//
       curs.c

use an array