From 099435be059e7986ff458a53e12ded419e03e37e Mon Sep 17 00:00:00 2001
From: chenson42 <chenson42@gmail.com>
Date: Sun, 16 Feb 2014 17:16:49 +0000
Subject: [PATCH] 0001586: Stage manager doesn't check to see if a file is
 actually deleted when it qualifies for cleaning.

---
 .../symmetric/io/stage/IStagedResource.java   |  4 ++-
 .../symmetric/io/stage/StagedResource.java    | 17 +++++++++---
 .../symmetric/io/stage/StagingManager.java    | 27 ++++++++++++-------
 3 files changed, 35 insertions(+), 13 deletions(-)

diff --git a/symmetric-io/src/main/java/org/jumpmind/symmetric/io/stage/IStagedResource.java b/symmetric-io/src/main/java/org/jumpmind/symmetric/io/stage/IStagedResource.java
index c8af10afa2..06a92d8e02 100644
--- a/symmetric-io/src/main/java/org/jumpmind/symmetric/io/stage/IStagedResource.java
+++ b/symmetric-io/src/main/java/org/jumpmind/symmetric/io/stage/IStagedResource.java
@@ -63,8 +63,10 @@ public String getExtensionName() {
     
     public boolean isFileResource();
     
-    public void delete();
+    public boolean delete();
     
     public boolean exists();
+    
+    public boolean isInUse();
 
 }
diff --git a/symmetric-io/src/main/java/org/jumpmind/symmetric/io/stage/StagedResource.java b/symmetric-io/src/main/java/org/jumpmind/symmetric/io/stage/StagedResource.java
index 095e5ebabf..9fddecddc4 100644
--- a/symmetric-io/src/main/java/org/jumpmind/symmetric/io/stage/StagedResource.java
+++ b/symmetric-io/src/main/java/org/jumpmind/symmetric/io/stage/StagedResource.java
@@ -103,6 +103,10 @@ public StagedResource(long threshold, File directory, String path, StagingManage
         this.state = State.CREATE;
     }
     
+    public boolean isInUse() {
+        return readers.size() > 0 || writers.size() > 0;
+    }
+    
     public boolean isFileResource() {     
         return file != null && file.exists();
     }
@@ -250,7 +254,7 @@ public BufferedWriter getWriter() {
                 log.warn("We had to delete {} because it already existed", file.getAbsolutePath());
                 file.delete();
             } else if (this.memoryBuffer != null) {
-                log.warn("We had to delete the memory buffer because it already existed");
+                log.warn("We had to delete the memory buffer for {} because it already existed", getPath());
                 this.memoryBuffer = null;
             }
             this.memoryBuffer = new StringBuilder();
@@ -283,12 +287,15 @@ public void refreshLastUpdateTime() {
         this.lastUpdateTime = System.currentTimeMillis();
     }
 
-    public void delete() {
+    public boolean delete() {
+        
+        boolean deleted = true;
         
         close();
         
         if (file.exists()) {
             FileUtils.deleteQuietly(file);
+            deleted = !file.exists();            
         }
 
         if (memoryBuffer != null) {
@@ -296,7 +303,11 @@ public void delete() {
             memoryBuffer = null;
         }
         
-        stagingManager.resourceList.remove(getPath());
+        if (deleted) {
+            stagingManager.resourceList.remove(getPath());
+        }
+        
+        return deleted;
         
     }
 
diff --git a/symmetric-io/src/main/java/org/jumpmind/symmetric/io/stage/StagingManager.java b/symmetric-io/src/main/java/org/jumpmind/symmetric/io/stage/StagingManager.java
index 5ee530a354..6692832175 100644
--- a/symmetric-io/src/main/java/org/jumpmind/symmetric/io/stage/StagingManager.java
+++ b/symmetric-io/src/main/java/org/jumpmind/symmetric/io/stage/StagingManager.java
@@ -91,17 +91,26 @@ public long clean(long ttlInMs) {
             for (String key : keys) {
                 IStagedResource resource = resourceList.get(key);
                 boolean resourceIsOld = (System.currentTimeMillis() - resource.getLastUpdateTime()) > ttlInMs;
-                if ((resource.getState() == State.READY || resource.getState() == State.DONE)
-                        && (resourceIsOld || !resource.exists())) {
-                    if (resource.isFileResource()) {
-                        purgedFileCount++;
-                        purgedFileSize += resource.getSize();
+                if (resource.isInUse() && (resource.getState() == State.READY || resource.getState() == State.DONE)
+                        && (resourceIsOld || !resource.exists())) {                    
+                    if (!resource.isInUse()) {
+                        boolean file = resource.isFileResource();
+                        long size = resource.getSize(); 
+                        if (resource.delete()) {                            
+                            if (file) {
+                                purgedFileCount++;
+                                purgedFileSize += size;
+                            } else {
+                                purgedMemCount++;
+                                purgedMemSize += size;
+                            }
+                           resourceList.remove(key);
+                        } else {
+                            log.warn("Failed to delete the '{}' staging resource", resource.getPath());
+                        }
                     } else {
-                        purgedMemCount++;
-                        purgedMemSize += resource.getSize();
+                        log.warn("The '{}' staging resource qualified for being cleaned, but was in use.  It will not be cleaned right now", resource.getPath());
                     }
-                    resource.delete();
-                    resourceList.remove(key);
                 }
             }
             if (purgedFileCount > 0) {