[HER-1776] H3: holding empty crawl for more URIs (other than 'paused')

* CrawlController
    change pauseAtFinish to runWhileEmpty
    add EMPTY state
    replace isStateRunning() with isActive() (RUNNING or EMPTY)
    trigger finish-test off frontier EMPTY rather than PAUSE
    in logging say 'running' rather than 'resumed' 
* Frontier
    add EMPTY state
* AbstractFrontier
    handle EMPTY like RUNNING, transition EMPTY<->RUNNING when appropriate
* StatisticsTracker
    handle EMPTY analogous to RUNNING
    in logging say 'running' rather than 'resumed' 
* CrawlJob, DiskSpaceMonitor
    replace isStateRunning() with isActive()

engine/src/main/java/org/archive/crawler/framework/CrawlController.java

@@ -190,13 +190,13 @@ implements Serializable,
         }
     }
     
-    /** whether to pause, rather than finish, when crawl appears done */
-    boolean pauseAtFinish = false; 
-    public boolean getPauseAtFinish() {
-        return pauseAtFinish;
+    /** whether to keep running (without pause or finish) when frontier is empty */
+    boolean runWhileEmpty = false; 
+    public boolean getRunWhileEmpty() {
+        return runWhileEmpty;
     }
-    public void setPauseAtFinish(boolean pauseAtFinish) {
-        this.pauseAtFinish = pauseAtFinish;
+    public void setRunWhileEmpty(boolean runWhileEmpty) {
+        this.runWhileEmpty = runWhileEmpty;
     }
 
     /** whether to pause at crawl start */
@@ -262,7 +262,7 @@ implements Serializable,
     private transient CrawlStatus sExit = CrawlStatus.CREATED;
 
     public static enum State {
-        NASCENT, RUNNING, PAUSED, PAUSING, 
+        NASCENT, RUNNING, EMPTY, PAUSED, PAUSING, 
         STOPPING, FINISHED, PREPARING 
     }
 
@@ -389,11 +389,12 @@ implements Serializable,
 
     private boolean shouldContinueCrawling() {
         Frontier frontier = getFrontier();
-        if (frontier.isEmpty()) {
+        if (frontier.isEmpty() && !getRunWhileEmpty()) {
             this.sExit = CrawlStatus.FINISHED;
             return false;
         }
-        return state == State.RUNNING;
+        // unsure this is correct; perhaps should be constant true
+        return isActive();
     }
 
     /**
@@ -468,8 +469,13 @@ implements Serializable,
         return state == State.PAUSING;
     }
     
-    public boolean isStateRunning() {
-        return state == State.RUNNING;
+    /**
+     * Is this crawl actively able/trying to crawl? Includes both 
+     * states RUNNING and EMPTY.
+     * @return
+     */
+    public boolean isActive() {
+        return state == State.RUNNING || state == State.EMPTY;
     }
 
     public boolean isFinished() {
@@ -539,15 +545,6 @@ implements Serializable,
         toePool.killThread(threadNumber, replace);
     }
 
-    /**
-     * Evaluate if the crawl should stop because it is finished.
-     */
-    public void checkFinish() {
-        if(atFinish()) {
-            beginCrawlStop();
-        }
-    }
-
     /**
      * Evaluate if the crawl should stop because it is finished,
      * without actually stopping the crawl.
@@ -555,7 +552,7 @@ implements Serializable,
      * @return true if crawl is at a finish-possible state
      */
     public boolean atFinish() {
-        return state == State.RUNNING && !shouldContinueCrawling();
+        return isActive() && !shouldContinueCrawling();
     }
 
     private void readObject(ObjectInputStream stream)
@@ -621,24 +618,20 @@ implements Serializable,
     public void noteFrontierState(Frontier.State reachedState) {
         switch (reachedState) {
         case RUN: 
-            LOGGER.info("Crawl resumed.");
+            LOGGER.info("Crawl running.");
             sendCrawlStateChangeEvent(State.RUNNING, CrawlStatus.RUNNING);
+            break;
+        case EMPTY: 
+            LOGGER.info("Crawl empty.");
+            if(!getRunWhileEmpty()) {
+                this.sExit = CrawlStatus.FINISHED;
+                beginCrawlStop();
+            }
+            sendCrawlStateChangeEvent(State.EMPTY, CrawlStatus.RUNNING);
+            break; 
         case PAUSE:
             if (state == State.PAUSING) {
                 completePause();
-                break;
-            }
-            if(atFinish()) { // really, "just reached finish"
-                if (getPauseAtFinish()) {
-                    completePause();
-                    break;
-                } else {
-                    beginCrawlStop();
-                }
-                break;
-            }
-            if(state == State.STOPPING || state == State.FINISHED) {
-                frontier.requestState(Frontier.State.FINISH);
             }
             break;
         case FINISH:

engine/src/main/java/org/archive/crawler/framework/CrawlJob.java

@@ -498,7 +498,7 @@ public class CrawlJob implements Comparable<CrawlJob>, ApplicationListener {
         if(cc==null) {
             return false;
         }
-        return cc.isStateRunning(); 
+        return cc.isActive(); 
     }
     
     public boolean isUnpausable() {

engine/src/main/java/org/archive/crawler/framework/Frontier.java

@@ -515,7 +515,8 @@ public interface Frontier extends Lifecycle, MultiReporter {
      */
     public enum State { 
         RUN,  // juggle/prioritize/emit; usual state
-        HOLD, // enter a consistent, stable, checkpointable state ASAP
+        EMPTY, // running/ready but no URIs queued/scheduled
+        HOLD, // NOT YET USED enter a consistent, stable, checkpointable state ASAP
         PAUSE, // enter a stable state where no URIs are in-progress; unlike
                // HOLD requires all in-process URIs to complete
         FINISH  // end and cleanup; may not return to any other state after

engine/src/main/java/org/archive/crawler/frontier/AbstractFrontier.java

@@ -345,20 +345,26 @@ public abstract class AbstractFrontier
         try {
             loop: while (true) {
                 try {
+                    State reachedState = null; 
                     switch (targetState) {
+                    case EMPTY:
+                        reachedState = State.EMPTY; 
                     case RUN:
                         // enable outbound takes if previously locked
                         while(outboundLock.isWriteLockedByCurrentThread()) {
                             outboundLock.writeLock().unlock();
                         }
-                        reachedState(State.RUN);
+                        if(reachedState==null) {
+                            reachedState = State.RUN; 
+                        }
+                        reachedState(reachedState);
                         
                         Thread.sleep(1000);
                         
-                        if(isEmpty()) {
-                            // pause when frontier exhausted; controller will
-                            // determine if this means to finish or not
-                            targetState = State.PAUSE;
+                        if(isEmpty()&&targetState==State.RUN) {
+                            requestState(State.EMPTY); 
+                        } else if (!isEmpty()&&targetState==State.EMPTY) {
+                            requestState(State.RUN); 
                         }
                         break;
                     case HOLD:

engine/src/main/java/org/archive/crawler/monitor/DiskSpaceMonitor.java

@@ -136,7 +136,7 @@ public class DiskSpaceMonitor implements ApplicationListener {
         long availBytes = path.getUsableSpace();
         long thresholdBytes = getPauseThresholdMiB() * 1024 * 1024;
 
-        if (availBytes < thresholdBytes && controller.isStateRunning()) {
+        if (availBytes < thresholdBytes && controller.isActive()) {
             // Enact pause
             controller.requestCrawlPause();
             

engine/src/main/java/org/archive/crawler/reporting/StatisticsTracker.java

@@ -541,7 +541,11 @@ public class StatisticsTracker
         if (this.crawlStartTime == 0) {
             noteStart();
         }
-        logNote("CRAWL RESUMED - " + statusMessage);
+        logNote("CRAWL RUNNING - " + statusMessage);
+    }
+    
+    public void crawlEmpty(String statusMessage) {
+        logNote("CRAWL EMPTY - " + statusMessage);
     }
 
     /**
@@ -979,6 +983,9 @@ public class StatisticsTracker
                 case RUNNING:
                     this.crawlResuming(event1.getMessage());
                     break;
+                case EMPTY:
+                    this.crawlEmpty(event1.getMessage());
+                    break;
                 case PAUSING:
                     this.crawlPausing(event1.getMessage());
                     break;