VIVO-869 Don't submit updates while indexer is paused.

When the pause flag is set, the IndexingChangeListener will continue to accumulate changes even if there
are gaps in the stream. The listener will not submit a task until the indexer is unpaused.

webapp/config/example.developer.properties

@@ -68,7 +68,7 @@
 # developer.searchIndex.showDocuments = false
 # developer.searchIndex.uriOrNameRestriction = .*
 # developer.searchIndex.documentRestriction = .*
-# developer.searchIndex.logIndexingBreakdownTimings = .*
+# developer.searchIndex.logIndexingBreakdownTimings = false
 # developer.searchIndex.suppressModelChangeListener = false
 # developer.searchDeletions.enable = false
 

webapp/src/edu/cornell/mannlib/vitro/webapp/modules/searchIndexer/SearchIndexer.java

@@ -73,7 +73,7 @@ public interface SearchIndexer extends Application.Module {
 
 	/**
 	 * Stop processing new tasks. Requests will be queued until a call to
-	 * unpause().
+	 * unpause(). Fires a PAUSED event to listeners.
 	 * 
 	 * The SearchIndexer is paused when created. When fully initialized, it
 	 * should be unpaused.
@@ -87,7 +87,8 @@ public interface SearchIndexer extends Application.Module {
 
 	/**
 	 * Resume processing new tasks. Any requests that were received since the
-	 * call to pause() will now be scheduled for processing.
+	 * call to pause() will now be scheduled for processing. Fires an UNPAUSED
+	 * event to listeners.
 	 * 
 	 * The SearchIndexer is paused when created. When fully initialized, it
 	 * should be unpaused.
@@ -149,6 +150,8 @@ public interface SearchIndexer extends Application.Module {
 		public enum Type {
 			STARTUP, PROGRESS,
 
+			PAUSE, UNPAUSE,
+
 			START_PROCESSING_URIS, STOP_PROCESSING_URIS,
 
 			START_PROCESSING_STATEMENTS, STOP_PROCESSING_STATEMENTS,

webapp/src/edu/cornell/mannlib/vitro/webapp/searchindex/IndexingChangeListener.java

@@ -2,6 +2,9 @@
 
 package edu.cornell.mannlib.vitro.webapp.searchindex;
 
+import static edu.cornell.mannlib.vitro.webapp.modules.searchIndexer.SearchIndexer.Event.Type.PAUSE;
+import static edu.cornell.mannlib.vitro.webapp.modules.searchIndexer.SearchIndexer.Event.Type.UNPAUSE;
+
 import java.io.ByteArrayInputStream;
 import java.io.UnsupportedEncodingException;
 import java.util.ArrayList;
@@ -20,6 +23,7 @@ import com.hp.hpl.jena.rdf.model.StmtIterator;
 
 import edu.cornell.mannlib.vitro.webapp.dao.jena.event.EditEvent;
 import edu.cornell.mannlib.vitro.webapp.modules.searchIndexer.SearchIndexer;
+import edu.cornell.mannlib.vitro.webapp.modules.searchIndexer.SearchIndexer.Event;
 import edu.cornell.mannlib.vitro.webapp.rdfservice.ChangeListener;
 import edu.cornell.mannlib.vitro.webapp.utils.threads.VitroBackgroundThread;
 
@@ -27,13 +31,18 @@ import edu.cornell.mannlib.vitro.webapp.utils.threads.VitroBackgroundThread;
  * When a change is heard, wait for an interval to see if more changes come in.
  * When changes stop coming in for a specified interval, send what has
  * accumulated.
+ * 
+ * When the SearchIndexer pauses, stop sending changes until the SearchIndexer
+ * unpauses.
  */
-public class IndexingChangeListener implements ChangeListener {
+public class IndexingChangeListener implements ChangeListener,
+		SearchIndexer.Listener {
 	private static final Log log = LogFactory
 			.getLog(IndexingChangeListener.class);
 
 	private final SearchIndexer searchIndexer;
 	private final Ticker ticker;
+	private volatile boolean paused = true;
 
 	/** All access to the list must be synchronized. */
 	private final List<Statement> changes;
@@ -42,17 +51,33 @@ public class IndexingChangeListener implements ChangeListener {
 		this.searchIndexer = searchIndexer;
 		this.ticker = new Ticker();
 		this.changes = new ArrayList<>();
+		
+		searchIndexer.addListener(this);
 	}
 
 	private synchronized void noteChange(Statement stmt) {
 		changes.add(stmt);
+		if (!paused) {
 			ticker.start();
 		}
+	}
+
+	@Override
+	public void receiveSearchIndexerEvent(Event event) {
+		if (event.getType() == PAUSE) {
+			paused = true;
+		} else if (event.getType() == UNPAUSE) {
+			paused = false;
+			respondToTicker();
+		}
+	}
 
 	private synchronized void respondToTicker() {
+		if (!paused && !changes.isEmpty()) {
 			searchIndexer.scheduleUpdatesForStatements(changes);
 			changes.clear();
 		}
+	}
 
 	public void shutdown() {
 		ticker.shutdown();

webapp/src/edu/cornell/mannlib/vitro/webapp/searchindex/SearchIndexerImpl.java

@@ -3,9 +3,11 @@
 package edu.cornell.mannlib.vitro.webapp.searchindex;
 
 import static edu.cornell.mannlib.vitro.webapp.modelaccess.ModelNames.DISPLAY;
+import static edu.cornell.mannlib.vitro.webapp.modules.searchIndexer.SearchIndexer.Event.Type.PAUSE;
 import static edu.cornell.mannlib.vitro.webapp.modules.searchIndexer.SearchIndexer.Event.Type.SHUTDOWN_COMPLETE;
 import static edu.cornell.mannlib.vitro.webapp.modules.searchIndexer.SearchIndexer.Event.Type.SHUTDOWN_REQUESTED;
 import static edu.cornell.mannlib.vitro.webapp.modules.searchIndexer.SearchIndexer.Event.Type.STARTUP;
+import static edu.cornell.mannlib.vitro.webapp.modules.searchIndexer.SearchIndexer.Event.Type.UNPAUSE;
 import static edu.cornell.mannlib.vitro.webapp.utils.developer.Key.SEARCH_INDEX_LOG_INDEXING_BREAKDOWN_TIMINGS;
 import static edu.cornell.mannlib.vitro.webapp.utils.threads.VitroBackgroundThread.WorkLevel.IDLE;
 import static edu.cornell.mannlib.vitro.webapp.utils.threads.VitroBackgroundThread.WorkLevel.WORKING;
@@ -39,6 +41,7 @@ import edu.cornell.mannlib.vitro.webapp.modelaccess.ModelAccess;
 import edu.cornell.mannlib.vitro.webapp.modules.Application;
 import edu.cornell.mannlib.vitro.webapp.modules.ComponentStartupStatus;
 import edu.cornell.mannlib.vitro.webapp.modules.searchIndexer.SearchIndexer;
+import edu.cornell.mannlib.vitro.webapp.modules.searchIndexer.SearchIndexer.Event.Type;
 import edu.cornell.mannlib.vitro.webapp.modules.searchIndexer.SearchIndexerStatus;
 import edu.cornell.mannlib.vitro.webapp.modules.searchIndexer.SearchIndexerStatus.State;
 import edu.cornell.mannlib.vitro.webapp.searchindex.documentBuilding.DocumentModifier;
@@ -235,11 +238,13 @@ public class SearchIndexerImpl implements SearchIndexer {
 	@Override
 	public void pause() {
 		scheduler.pause();
+		listeners.fireEvent(new Event(PAUSE, getStatus()));
 	}
 
 	@Override
 	public void unpause() {
 		scheduler.unpause();
+		listeners.fireEvent(new Event(UNPAUSE, getStatus()));
 	}
 
 	@Override

webapp/src/edu/cornell/mannlib/vitro/webapp/searchindex/SearchIndexerSetup.java

@@ -62,8 +62,10 @@ public class SearchIndexerSetup implements ServletContextListener {
 
 			listener = new IndexingChangeListener(searchIndexer);
 			
+			// Wrap it so it can be disabled by a developer flag.
 			listenerWrapper = new DeveloperDisabledChangeListener(listener,
 					Key.SEARCH_INDEX_SUPPRESS_MODEL_CHANGE_LISTENER);
+
 			RDFServiceUtils.getRDFServiceFactory(ctx).registerListener(
 					listenerWrapper);