removed workingDir, no extends runnable

debug-init-env.sh

@@ -22,7 +22,6 @@ echo "OK"
 
 echo -n "Creating directories for test data and processing..."
 mkdir $TEMPDIR/transfer-inbox
-mkdir $TEMPDIR/easy-mirror-deposit-working-directory
 mkdir $TEMPDIR/easy-ingest-flow-inbox
 mkdir $TEMPDIR/easy-mirror-deposit-failed
 mkdir $TEMPDIR/easy-mirror-store

src/main/assembly/dist/cfg/config.yml

@@ -26,7 +26,6 @@ mirroringService:
     - path: '/var/opt/dans.knaw.nl/tmp/transfer-inboxes/phys-techsciences'
 
   pollingInterval: 1000
-  workDir: '/var/opt/dans.knaw.nl/tmp/easy-mirror-deposit-working-directory'
   failedBox: '/var/opt/dans.knaw.nl/tmp/easy-mirror-deposit-failed'
   easyMirrorStore: '/data/easy-mirror-store'
 

src/main/java/nl/knaw/dans/easy/mirror/EasyMirrorDepositApplication.java

@@ -38,8 +38,7 @@ public String getName() {
     @Override
     public void run(final EasyMirrorDepositConfiguration configuration, final Environment environment) {
         final ExecutorService taskExecutor = configuration.getTaskQueue().build(environment);
-        final MirroringService mirroringService = configuration.getMirroringService()
-            .build(taskExecutor);
+        final MirroringService mirroringService = configuration.getMirroringService().build();
         environment.lifecycle().manage(mirroringService);
         for (Inbox inbox : configuration.getMirroringService().getInboxes()) {
             environment.healthChecks().register(String.format("Inbox-%s", inbox.getPath().toString()), new InboxHealth(inbox.getPath()));

src/main/java/nl/knaw/dans/easy/mirror/MirroringServiceFactory.java

@@ -32,10 +32,6 @@ public class MirroringServiceFactory {
 
     private int pollingInterval;
 
-    @NotNull
-    @Valid
-    private Path workDir;
-
     @NotNull
     @Valid
     private Path failedBox;
@@ -44,8 +40,8 @@ public class MirroringServiceFactory {
     @Valid
     private Path easyMirrorStore;
 
-    public MirroringService build(ExecutorService executorService) {
-        return new MirroringService(executorService, pollingInterval, inboxes, workDir,
+    public MirroringService build() {
+        return new MirroringService(pollingInterval, inboxes,
             failedBox, easyMirrorStore);
     }
 
@@ -65,14 +61,6 @@ public void setPollingInterval(int pollingInterval) {
         this.pollingInterval = pollingInterval;
     }
 
-    public Path getWorkDir() {
-        return workDir;
-    }
-
-    public void setWorkDir(Path workDir) {
-        this.workDir = workDir;
-    }
-
     public Path getFailedBox() {
         return failedBox;
     }

src/main/java/nl/knaw/dans/easy/mirror/core/MirrorTask.java

@@ -22,21 +22,18 @@
 import java.nio.file.Files;
 import java.nio.file.Path;
 
-public class MirrorTask implements Runnable {
+public class MirrorTask  {
     private static final Logger log = LoggerFactory.getLogger(MirrorTask.class);
 
-    private final Path datasetVersionExportZip;
     private final Path failedBox;
     private final MirrorStore mirrorStore;
 
-    public MirrorTask(Path datasetVersionExportZip, Path failedBox, MirrorStore mirrorStore) {
-        this.datasetVersionExportZip = datasetVersionExportZip;
+    public MirrorTask(Path failedBox, MirrorStore mirrorStore) {
         this.failedBox = failedBox;
         this.mirrorStore = mirrorStore;
     }
 
-    @Override
-    public void run() {
+    public void move(Path datasetVersionExportZip) {
         log.info("Processing {}", datasetVersionExportZip.getFileName());
 
         try {

src/main/java/nl/knaw/dans/easy/mirror/core/MirroringService.java

@@ -30,20 +30,13 @@
 import java.nio.file.Paths;
 import java.util.List;
 import java.util.Optional;
-import java.util.concurrent.ExecutorService;
 import java.util.stream.Stream;
 
 public class MirroringService implements Managed {
     private static final Logger log = LoggerFactory.getLogger(MirroringService.class);
-    private final ExecutorService executorService;
     private final int pollingInterval;
     private final List<Inbox> inboxes;
-    private final Path failedBox;
-    private final Path workDirectory;
-    private final MirrorStore mirrorStore;
-
-    private boolean initialized = false;
-    private boolean tasksCreatedInitialization = false;
+    private final MirrorTask mirrorTask;
 
     private class EventHandler extends FileAlterationListenerAdaptor {
         private final Inbox inbox;
@@ -55,32 +48,21 @@ public EventHandler(Inbox inbox) {
         @Override
         public void onStart(FileAlterationObserver observer) {
             log.trace("onStart called");
-            if (!initialized) {
-                initialized = true;
-                processAllFromInbox(inbox);
-            }
+            processAllFromInbox(inbox);
         }
 
         @Override
         public void onFileCreate(File file) {
             log.trace("onFileCreate: {}", file);
-            if (tasksCreatedInitialization) {
-                tasksCreatedInitialization = false;
-                return; // file already added to queue by onStart
-            }
-            scheduleDatasetVersionExport(file.toPath());
+            executeMirrorTask(file.toPath());
         }
     }
 
-    public MirroringService(ExecutorService executorService, int pollingInterval, List<Inbox> inboxes,
-        Path workDirectory,
+    public MirroringService(int pollingInterval, List<Inbox> inboxes,
         Path failedBox, Path mirrorStore) {
-        this.executorService = executorService;
         this.pollingInterval = pollingInterval;
         this.inboxes = inboxes;
-        this.workDirectory = workDirectory;
-        this.failedBox = failedBox;
-        this.mirrorStore = new MirrorStore(mirrorStore);
+        this.mirrorTask =new MirrorTask(failedBox, new MirrorStore(mirrorStore));
     }
 
     @Override
@@ -113,21 +95,17 @@ private void processAllFromInbox(Inbox inbox) {
             DveFileFilter fileFilter = new DveFileFilter(inbox.getPath());
             try (Stream<Path> files = Files.list(inbox.getPath())) {
                 files.filter(f -> fileFilter.accept(f.toFile()))
-                    .forEach(dve -> {
-                        scheduleDatasetVersionExport(dve);
-                        tasksCreatedInitialization = true;
-                    });
+                    .forEach(this::executeMirrorTask);
             }
         }
         catch (IOException e) {
             throw new IllegalStateException("Could not read DVEs from inbox", e);
         }
     }
 
-    private void scheduleDatasetVersionExport(Path dve) {
-        log.info("Scheduling " + dve.getFileName());
+    private void executeMirrorTask(Path dve) {
+        log.info("Executing " + dve.getFileName());
         try {
-            Path workingDve = Files.move(dve, workDirectory.resolve(dve.getFileName()));
             Optional<Path> optXmlFile = getAssociatedXmlFile(dve);
             if (optXmlFile.isPresent()) {
                 log.debug("Removing associated XML file {}", optXmlFile.get());
@@ -136,10 +114,10 @@ private void scheduleDatasetVersionExport(Path dve) {
             else {
                 log.warn("Associated XML file was not found");
             }
-            executorService.execute(new MirrorTask(workingDve, failedBox, mirrorStore));
+            mirrorTask.move(dve);
         }
         catch (IOException e) {
-            log.error("Could not move DVE to work directory", e);
+            log.error("Could not move DVE to mirror store", e);
         }
     }
 
@@ -152,8 +130,4 @@ private Optional<Path> getAssociatedXmlFile(Path path) {
             return Optional.empty();
         }
     }
-
-    public void stop() {
-        executorService.shutdown();
-    }
 }

src/test/java/nl/knaw/dans/easy/mirror/core/MirrorTaskTest.java

@@ -43,31 +43,31 @@ public void setUp() throws Exception {
         Files.createDirectories(mirrorStoreDir);
     }
 
-    private MirrorTask createTask(Path dve) throws Exception {
+    private void executeTask(Path dve) throws Exception {
         Path dveInInbox = inbox.resolve(dve.getFileName());
         Files.copy(dveRootDir.resolve(dve), dveInInbox);
-        return new MirrorTask(dveInInbox, failedBox, mirrorStore);
+        new MirrorTask(failedBox, mirrorStore).move(dveInInbox);
     }
 
     @Test
     public void dve_with_invalid_name_goes_to_failedBox() throws Exception {
         Path dve = Paths.get("invalid-names/not-a-dve.zip");
-        createTask(dve).run();
+        executeTask(dve);
         assertTrue(Files.exists(failedBox.resolve(dve.getFileName())));
         assertFalse(mirrorStore.contains(dve));
     }
 
     @Test
     public void dve_V1_1_goes_only_to_mirror_store() throws Exception {
         Path dve = Paths.get("valid/doi-10-5072-fk2-xcfq1bv1.1.zip");
-        createTask(dve).run();
+        executeTask(dve);
         assertTrue(mirrorStore.contains(dve));
     }
 
     @Test
     public void dve_V1_goes_to_mirror_store_and_produces_deposit() throws Exception {
         Path dve = Paths.get("valid/doi-10-5072-fk2-xcfq1bv1.0.zip");
-        createTask(dve).run();
+        executeTask(dve);
         assertTrue(mirrorStore.contains(dve));
     }
 }

src/test/resources/debug-etc/config.yml

@@ -21,7 +21,6 @@ taskQueue:
 mirroringService:
   inbox: 'data/transfer-inbox'
   pollingInterval: 500
-  workDir: 'data/easy-mirror-deposit-working-directory'
   failedBox: 'data/easy-mirror-deposit-failed'
   easyMirrorStore: 'data/easy-mirror-store'