no face threading

Author: tburch

src/main/java/com/lowtuna/jsonblob/core/BlobCleanupConsumer.java

@@ -1,19 +1,21 @@
 package com.lowtuna.jsonblob.core;
 
-import com.codahale.metrics.Gauge;
-import com.codahale.metrics.MetricRegistry;
+import com.fasterxml.jackson.core.JsonParseException;
+import com.fasterxml.jackson.databind.JsonMappingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Optional;
 import com.google.common.base.Stopwatch;
 import io.dropwizard.util.Duration;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.io.DirectoryWalker;
+import org.joda.time.DateTime;
 
 import java.io.File;
 import java.io.IOException;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.time.LocalDate;
 import java.util.Collection;
-import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
@@ -24,14 +26,15 @@
 public class BlobCleanupProducer extends DirectoryWalker<Void> implements Runnable {
   private final Path dataDirectoryPath;
   private final Duration blobAccessTtl;
-  private final BlockingQueue<File> filesToProcess;
+  private final FileSystemJsonBlobManager fileSystemJsonBlobManager;
+  private final ObjectMapper om;
 
-  public BlobCleanupProducer(Path dataDirectoryPath, Duration blobAccessTtl, BlockingQueue<File> filesToProcess, MetricRegistry metricRegistry) {
+  public BlobCleanupProducer(Path dataDirectoryPath, Duration blobAccessTtl, FileSystemJsonBlobManager fileSystemJsonBlobManager, ObjectMapper om) {
     super(null, 3);
     this.dataDirectoryPath = dataDirectoryPath;
     this.blobAccessTtl = blobAccessTtl;
-    this.filesToProcess = filesToProcess;
-    metricRegistry.register(MetricRegistry.name(getClass(), "filesToProcessCount"), (Gauge<Integer>) () -> filesToProcess.size());
+    this.fileSystemJsonBlobManager = fileSystemJsonBlobManager;
+    this.om = om;
   }
 
 
@@ -51,10 +54,41 @@ protected boolean handleDirectory(File directory, int depth, Collection<Void> re
                   if (file.getName().startsWith(FileSystemJsonBlobManager.BLOB_METADATA_FILE_NAME)) {
                     return;
                   }
+
                   try {
-                    filesToProcess.put(file);
-                  } catch (InterruptedException e) {
-                    log.warn("Interrupted while trying to add file to be processed at {}", file.getAbsolutePath(), e);
+                    log.debug("Processing {}", file.getAbsolutePath());
+                    String blobId = file.getName().split("\\.", 2)[0];
+                    File metadataFile = fileSystemJsonBlobManager.getMetaDataFile(file.getParentFile());
+
+                    if (file.equals(metadataFile)) {
+                      return;
+                    }
+
+                    BlobMetadataContainer metadataContainer = metadataFile.exists() ? om.readValue(fileSystemJsonBlobManager.readFile(metadataFile), BlobMetadataContainer.class) : new BlobMetadataContainer();
+
+                    Optional<DateTime> lastAccessed = fileSystemJsonBlobManager.resolveTimestamp(blobId);
+                    if (metadataContainer.getLastAccessedByBlobId().containsKey(blobId)) {
+                      lastAccessed = Optional.of(metadataContainer.getLastAccessedByBlobId().get(blobId));
+                    }
+
+                    if (!lastAccessed.isPresent()) {
+                      log.warn("Couldn't get last accessed timestamp for blob {}", blobId);
+                      return;
+                    }
+
+                    log.debug("Blob {} was last accessed {}", blobId, lastAccessed.get());
+
+                    if (lastAccessed.get().plusMillis((int) blobAccessTtl.toMilliseconds()).isBefore(DateTime.now())) {
+                      if (file.delete()) {
+                        log.info("Blob {} hasn't been accessed in {} (last accessed {}), so it's going to be deleted", blobId, blobAccessTtl, lastAccessed.get());
+                      }
+                    }
+                  } catch (JsonParseException e) {
+                    log.warn("Couldn't parse JSON from BlobMetadataContainer", e);
+                  } catch (JsonMappingException e) {
+                    log.warn("Couldn't map JSON from BlobMetadataContainer", e);
+                  } catch (IOException e) {
+                    log.warn("Couldn't read json for BlobMetadataContainer file", e);
                   }
                 });
 

src/main/java/com/lowtuna/jsonblob/core/BlobCleanupProducer.java

@@ -32,8 +32,6 @@
 import java.util.Map;
 import java.util.TimeZone;
 import java.util.UUID;
-import java.util.concurrent.ArrayBlockingQueue;
-import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
@@ -50,7 +48,6 @@
 public class FileSystemJsonBlobManager implements JsonBlobManager, Runnable, Managed {
   static final String BLOB_METADATA_FILE_NAME = "blobMetadata";
 
-  private static final int CONSUMER_COUNT = 7;
   private static final DateTimeFormatter DIRECTORY_FORMAT = DateTimeFormat.forPattern("yyyy/MM/dd");
 
   @GuardedBy("lastAccessedLock")
@@ -313,13 +310,7 @@ public void start() throws Exception {
     log.info("Scheduling the updating of blob last accessed timestamps");
     scheduledExecutorService.scheduleWithFixedDelay(this, 1, 1, TimeUnit.MINUTES);
 
-    log.info("Scheduling blob cleanup job");
-    BlockingQueue<File> filesToProcess = new ArrayBlockingQueue<>(1024);
-    for (int i = 0; i < CONSUMER_COUNT; i++) {
-      cleanupScheduledExecutorService.scheduleAtFixedRate(new BlobCleanupConsumer(filesToProcess, blobAccessTtl, this, objectMapper), 0, 1, TimeUnit.SECONDS);
-    }
-
-    BlobCleanupProducer dataDirectoryCleaner = new BlobCleanupProducer(blobDataDirectory.toPath(), blobAccessTtl, filesToProcess, metricRegistry);
+    BlobCleanupProducer dataDirectoryCleaner = new BlobCleanupProducer(blobDataDirectory.toPath(), blobAccessTtl, this, objectMapper);
     cleanupScheduledExecutorService.scheduleWithFixedDelay(dataDirectoryCleaner, 0, 1, TimeUnit.DAYS);
   }
 

src/main/java/com/lowtuna/jsonblob/core/FileSystemJsonBlobManager.java

@@ -36,7 +36,7 @@ public void run() {
       updatesByMetadataFile.put(metadataFile, Pair.of(entry.getKey(), entry.getValue()));
     }
 
-    updatesByMetadataFile.keySet().parallelStream().forEach(metadataFile -> {
+    updatesByMetadataFile.keySet().stream().forEach(metadataFile -> {
       try {
         log.info(metadataFile.exists() ? "Reading metadata file at {}" : "No metadata file exists yet, so creating a new one", metadataFile.getAbsolutePath());
         BlobMetadataContainer metadataContainer = metadataFile.exists() ? om.readValue(fileSystemJsonBlobManager.readFile(metadataFile), BlobMetadataContainer.class) : new BlobMetadataContainer();