Commit 2c3fc849 authored by Brian Long's avatar Brian Long
Browse files

add index support to reconcile

parent bffceee2

shared/src/main/java/com/inteligr8/alfresco/asie/rest/ReconcileAcsNodesWebScript.java

+2 −1
Original line number Diff line number Diff line
@@ -32,6 +32,7 @@ public class ReconcileAcsNodesWebScript extends AbstractAsieWebScript { 
	public void executeAuthorized(WebScriptRequest request, WebScriptResponse response) throws IOException {

		final int fromDbId = this.getRequestTemplateIntegerVariable(request, "fromDbId");

		final int toDbId = this.getRequestTemplateIntegerVariable(request, "toDbId");

		final boolean index = Boolean.TRUE.equals(this.getOptionalQueryParameter(request, "index", Boolean.class));

		final boolean reindex = Boolean.TRUE.equals(this.getOptionalQueryParameter(request, "reindex", Boolean.class));

		final boolean includeReconciled = Boolean.TRUE.equals(this.getOptionalQueryParameter(request, "includeReconciled", Boolean.class));
@@ -101,7 +102,7 @@ public class ReconcileAcsNodesWebScript extends AbstractAsieWebScript { 
		};

		

		try {

			this.reconcileService.reconcile(fromDbId, toDbId, null, reindex, callback, 1L, TimeUnit.HOURS, 2L, TimeUnit.MINUTES);

			this.reconcileService.reconcile(fromDbId, toDbId, null, index, reindex, callback, 1L, TimeUnit.HOURS, 2L, TimeUnit.MINUTES);



			if (responseMap.containsKey("error")) {

				response.setStatus(HttpStatus.INTERNAL_SERVER_ERROR.value());

shared/src/main/java/com/inteligr8/alfresco/asie/service/AcsReconcileService.java

+84 −25
Original line number Diff line number Diff line
@@ -32,6 +32,7 @@ import org.springframework.beans.factory.annotation.Value; 
import org.springframework.stereotype.Component;



import com.inteligr8.alfresco.asie.model.ShardInstance;

import com.inteligr8.alfresco.asie.spi.IndexCallback;

import com.inteligr8.alfresco.asie.spi.ReconcileCallback;

import com.inteligr8.alfresco.asie.spi.ReindexCallback;

import com.inteligr8.alfresco.asie.util.CompositeFuture;
@@ -55,6 +56,9 @@ public class AcsReconcileService implements InitializingBean, DisposableBean { 
	@Autowired

	private SearchService searchService;

	

	@Autowired

	private IndexService indexService;

	

	@Autowired

	private ReindexService reindexService;
@@ -109,20 +113,21 @@ public class AcsReconcileService implements InitializingBean, DisposableBean { 
	 */

	public void reconcile(

			long fromDbId, long toDbId, Integer nodesChunkSize,

			boolean reindexUnreconciled,

			boolean indexUnreconciled,

			boolean reindexReconciled,

			ReconcileCallback callback,

			long queueTimeout, TimeUnit queueUnit,

			long execTimeout, TimeUnit execUnit) throws InterruptedException, TimeoutException {

		if (nodesChunkSize == null)

			nodesChunkSize = this.nodesChunkSize;

		if (this.logger.isTraceEnabled())

			this.logger.trace("reconcile({}, {}, {}, {}, {}, {})", fromDbId, toDbId, nodesChunkSize, reindexUnreconciled, queueUnit.toMillis(queueTimeout), execUnit.toMillis(execTimeout));

			this.logger.trace("reconcile({}, {}, {}, {}, {}, {}, {})", fromDbId, toDbId, nodesChunkSize, indexUnreconciled, reindexReconciled, queueUnit.toMillis(queueTimeout), execUnit.toMillis(execTimeout));



		CompositeFuture<Void> future = new CompositeFuture<>();

		

		for (long startDbId = fromDbId; startDbId < toDbId; startDbId += nodesChunkSize) {

			long endDbId = Math.min(toDbId, startDbId + nodesChunkSize);

			future.combine(this.reconcileChunk(startDbId, endDbId, reindexUnreconciled, callback, queueTimeout, queueUnit, execTimeout, execUnit));

			future.combine(this.reconcileChunk(startDbId, endDbId, indexUnreconciled, reindexReconciled, callback, queueTimeout, queueUnit, execTimeout, execUnit));

			future.purge(true);

		}
@@ -135,18 +140,19 @@ public class AcsReconcileService implements InitializingBean, DisposableBean { 
	

	public Future<Void> reconcile(

			long fromDbId, long toDbId, Integer nodesChunkSize,

			boolean reindexUnreconciled,

			boolean indexUnreconciled,

			boolean reindexReconciled,

			ReconcileCallback callback) throws InterruptedException {

		if (nodesChunkSize == null)

			nodesChunkSize = this.nodesChunkSize;

		this.logger.trace("reconcile({}, {}, {}, {})", fromDbId, toDbId, nodesChunkSize, reindexUnreconciled);

		this.logger.trace("reconcile({}, {}, {}, {}, {})", fromDbId, toDbId, nodesChunkSize, indexUnreconciled, reindexReconciled);



		CompositeFuture<Void> future = new CompositeFuture<>();

		

		try {

			for (long startDbId = fromDbId; startDbId < toDbId; startDbId += nodesChunkSize) {

				long endDbId = Math.min(toDbId, startDbId + nodesChunkSize);

				future.combine(this.reconcileChunk(startDbId, endDbId, reindexUnreconciled, callback, -1L, null, -1L, null));

				future.combine(this.reconcileChunk(startDbId, endDbId, indexUnreconciled, reindexReconciled, callback, -1L, null, -1L, null));

				future.purge(true);

			}

		} catch (TimeoutException te) {
@@ -158,12 +164,13 @@ public class AcsReconcileService implements InitializingBean, DisposableBean { 
	

	protected Future<Void> reconcileChunk(

			long fromDbId, long toDbId,

			boolean reindexUnreconciled,

			boolean indexUnreconciled,

			boolean reindexReconciled,

			ReconcileCallback callback,

			long queueTimeout, TimeUnit queueUnit,

			long execTimeout, TimeUnit execUnit) throws InterruptedException, TimeoutException {

		if (this.logger.isTraceEnabled())

			this.logger.trace("reconcileChunk({}, {}, {}, {}, {})", fromDbId, toDbId, reindexUnreconciled, queueUnit.toMillis(queueTimeout), execUnit.toMillis(execTimeout));

			this.logger.trace("reconcileChunk({}, {}, {}, {}, {}, {})", fromDbId, toDbId, indexUnreconciled, reindexReconciled, queueUnit.toMillis(queueTimeout), execUnit.toMillis(execTimeout));

		

		int dbIdCount = (int) (toDbId - fromDbId);
@@ -202,21 +209,31 @@ public class AcsReconcileService implements InitializingBean, DisposableBean { 
			final long nodeDbId = _nodeDbId;

			this.logger.trace("Attempting to reconcile ACS node: {}", nodeDbId);



			Callable<Void> callable;



			final int dbIdIndex = (int) (nodeDbId - fromDbId);

			if (nodeRefs[dbIdIndex] != null) {

				this.logger.trace("A node in the DB is already indexed in Solr: {}: {}", nodeDbId, nodeRefs[dbIdIndex]);

				this.reconcileLogger.info("RECONCILED: {} <=> {}", nodeDbId, nodeRefs[dbIdIndex]);

				callable = new Callable<Void>() {

					@Override

					public Void call() throws InterruptedException, TimeoutException {

						logger.trace("A node in the DB is already indexed in Solr: {}: {}", nodeDbId, nodeRefs[dbIdIndex]);

						reconcileLogger.info("RECONCILED: {} <=> {}", nodeDbId, nodeRefs[dbIdIndex]);

						callback.reconciled(nodeDbId);

				continue;

			}



			Callable<Void> callable = new Callable<Void>() {

						if (reindexReconciled)

							reindex(nodeDbId, nodeRefs[dbIdIndex], callback, execTimeout, execUnit);

						return null;

					}

				};

			} else {

				callable = new Callable<Void>() {

					@Override

					public Void call() throws InterruptedException, TimeoutException {

					reconcile(nodeDbId, reindexUnreconciled, callback, execTimeout, execUnit);

						reconcile(nodeDbId, indexUnreconciled, callback, execTimeout, execUnit);

						return null;

					}

				};

			}

			

			if (queueTimeout < 0L) {

				future.combine(this.executor.submit(callable, -1L, null));
@@ -229,7 +246,7 @@ public class AcsReconcileService implements InitializingBean, DisposableBean { 
	}

	

	public void reconcile(long nodeDbId,

			boolean reindexUnreconciled,

			boolean index,

			ReconcileCallback callback,

			long execTimeout, TimeUnit execUnit) throws InterruptedException, TimeoutException {

		NodeRef nodeRef = this.nodeService.getNodeRef(nodeDbId);
@@ -250,16 +267,58 @@ public class AcsReconcileService implements InitializingBean, DisposableBean { 
			return;

		}

		

		if (!reindexUnreconciled) {

		if (!index) {

			this.logger.debug("A node in the DB is not indexed in Solr: {}: {}", nodeDbId, nodeRef);

			this.reconcileLogger.info("UNRECONCILED: {} <=> {}", nodeDbId, nodeRef);

			callback.unreconciled(nodeDbId);

		} else {

			logger.debug("A node in the DB is not indexed in Solr; attempt to reindex: {}: {}", nodeDbId, nodeRef);

			this.reindex(nodeDbId, nodeRef, callback, execTimeout, execUnit);

			logger.debug("A node in the DB is not indexed in Solr; attempt to index: {}: {}", nodeDbId, nodeRef);

			this.index(nodeDbId, nodeRef, callback, execTimeout, execUnit);

		}

	}

	

	public void index(long nodeDbId, NodeRef nodeRef,

			ReconcileCallback callback,

			long execTimeout, TimeUnit execUnit) throws InterruptedException, TimeoutException {

		Set<ShardInstance> syncHosts = new HashSet<>();

		Set<ShardInstance> asyncHosts = new HashSet<>();

		Map<ShardInstance, String> errorHosts = new HashMap<>();

		

		IndexCallback indexCallback = new IndexCallback() {

			

			@Override

			public void success(ShardInstance instance) {

				reconcileLogger.info("INDEXED: {} <=> {}", nodeDbId, nodeRef);

				syncHosts.add(instance);

			}

			

			@Override

			public void scheduled(ShardInstance instance) {

				reconcileLogger.info("INDEXING: {} <=> {}", nodeDbId, nodeRef);

				asyncHosts.add(instance);

			}

			

			@Override

			public void error(ShardInstance instance, String message) {

				reconcileLogger.info("FAILED INDEX: {} <=> {}", nodeDbId, nodeRef);

				errorHosts.put(instance, message);

			}

		};

		

		try {

			if (execTimeout < 0L) {

				this.indexService.index(nodeDbId, indexCallback).get();

			} else {

				this.indexService.index(nodeDbId, indexCallback).get(execTimeout, execUnit);

			}

		} catch (ExecutionException ee) {

			throw new RuntimeException("An unexpected exception occurred: " + ee.getMessage(), ee);

		}

		

		if (callback != null)

			callback.processed(nodeDbId, syncHosts, asyncHosts, errorHosts);

	}

	

	public void reindex(long nodeDbId, NodeRef nodeRef,

			ReconcileCallback callback,

			long execTimeout, TimeUnit execUnit) throws InterruptedException, TimeoutException {
@@ -283,7 +342,7 @@ public class AcsReconcileService implements InitializingBean, DisposableBean { 
			

			@Override

			public void error(ShardInstance instance, String message) {

				reconcileLogger.info("UNINDEXED: {} <=> {}", nodeDbId, nodeRef);

				reconcileLogger.info("FAILED REINDEX: {} <=> {}", nodeDbId, nodeRef);

				errorHosts.put(instance, message);

			}

		};

shared/src/main/resources/alfresco/extension/templates/webscripts/com/inteligr8/alfresco/asie/reconcileAcsNodes.post.desc.xml

+8 −1
Original line number Diff line number Diff line
@@ -16,6 +16,13 @@ 
			<dt>toDbId</dt>

			<dd>A DB ID integer for the ending point of a range, exclusive.</dd>

		</dl>

		<p>The following query parameters are also supported:</p>

		<dl>

			<dt>index</dt>

			<dd>Schedule an `INDEX` action against all unreconciled nodes.</dd>

			<dt>reindex</dt>

			<dd>Schedule a `REINDEX` action against all reconciled nodes.</dd>

		</dl>

		<p>The following response body should be expected in most cases (200, 202, and 500 status codes):</p>

		<pre>

			{
@@ -58,7 +65,7 @@ 
	]]></description>

	

	<!-- Endpoint Configuration -->

	<url>/inteligr8/asie/acs/nodes/{fromDbId}/{toDbId}/reconcile?reindex={reindex?}</url>

	<url>/inteligr8/asie/acs/nodes/{fromDbId}/{toDbId}/reconcile?index={index?}&amp;reindex={reindex?}</url>

	<format default="json">any</format>

	

	<!-- Security -->