bundles/org.eclipse.team.core/src/org/eclipse/team/internal/core/subscribers/SubscriberEventHandler.java

/*******************************************************************************
 * Copyright (c) 2000, 2017 IBM Corporation and others.
 *
 * This program and the accompanying materials
 * are made available under the terms of the Eclipse Public License 2.0
 * which accompanies this distribution, and is available at
 * https://www.eclipse.org/legal/epl-2.0/
 *
 * SPDX-License-Identifier: EPL-2.0
 *
 * Contributors:
 *     IBM Corporation - initial API and implementation
 *******************************************************************************/
package org.eclipse.team.internal.core.subscribers;

import java.util.ArrayList;
import java.util.List;

import org.eclipse.core.resources.IResource;
import org.eclipse.core.resources.IWorkspaceRunnable;
import org.eclipse.core.resources.mapping.ResourceTraversal;
import org.eclipse.core.runtime.CoreException;
import org.eclipse.core.runtime.IProgressMonitor;
import org.eclipse.core.runtime.OperationCanceledException;
import org.eclipse.core.runtime.jobs.IJobChangeEvent;
import org.eclipse.core.runtime.jobs.Job;
import org.eclipse.osgi.util.NLS;
import org.eclipse.team.core.ITeamStatus;
import org.eclipse.team.core.TeamException;
import org.eclipse.team.core.mapping.ISynchronizationScope;
import org.eclipse.team.core.mapping.ISynchronizationScopeChangeListener;
import org.eclipse.team.core.subscribers.Subscriber;
import org.eclipse.team.internal.core.BackgroundEventHandler;
import org.eclipse.team.internal.core.Messages;
import org.eclipse.team.internal.core.Policy;

/**
 * This handler collects changes and removals to resources and calculates their
 * synchronization state in a background job. The result is fed input the SyncSetInput.
 *
 * Exceptions that occur when the job is processing the events are collected and
 * returned as part of the Job's status.
 */
public abstract class SubscriberEventHandler extends BackgroundEventHandler {

	// Changes accumulated by the event handler
	private List<Event> resultCache = new ArrayList<>();

	private boolean started = false;
	private boolean initializing = true;

	private IProgressMonitor progressGroup;

	private int ticks;

	private final Subscriber subscriber;
	private ISynchronizationScope scope;

	private ISynchronizationScopeChangeListener scopeChangeListener;

	/**
	 * Internal resource synchronization event. Can contain a result.
	 */
	class SubscriberEvent extends ResourceEvent{
		static final int REMOVAL = 1;
		static final int CHANGE = 2;
		static final int INITIALIZE = 3;

		SubscriberEvent(IResource resource, int type, int depth) {
			super(resource, type, depth);
		}
		@Override
		protected String getTypeString() {
			switch (getType()) {
				case REMOVAL :
					return "REMOVAL"; //$NON-NLS-1$
				case CHANGE :
					return "CHANGE"; //$NON-NLS-1$
				case INITIALIZE :
					return "INITIALIZE"; //$NON-NLS-1$
				default :
					return "INVALID"; //$NON-NLS-1$
			}
		}
		public ResourceTraversal asTraversal() {
			return new ResourceTraversal(new IResource[] { getResource() }, getDepth(), IResource.NONE);
		}
	}

	/**
	 * Create a handler. This will initialize all resources for the subscriber associated with
	 * the set.
	 * @param subscriber the subscriber
	 * @param scope the scope
	 */
	public SubscriberEventHandler(Subscriber subscriber, ISynchronizationScope scope) {
		super(
			NLS.bind(Messages.SubscriberEventHandler_jobName, new String[] { subscriber.getName() }),
			NLS.bind(Messages.SubscriberEventHandler_errors, new String[] { subscriber.getName() }));
		this.subscriber = subscriber;
		this.scope = scope;
		scopeChangeListener = (scope1, newMappings, newTraversals) -> reset(new ResourceTraversal[0], scope1.getTraversals());
		this.scope.addScopeChangeListener(scopeChangeListener);
	}

	/**
	 * The traversals of the scope have changed
	 * @param oldTraversals the old traversals
	 * @param newTraversals the new traversals
	 */
	protected synchronized void reset(ResourceTraversal[] oldTraversals, ResourceTraversal[] newTraversals) {
		reset(newTraversals, SubscriberEvent.CHANGE);
	}

	/**
	 * Start the event handler by queuing events to prime the sync set input with the out-of-sync
	 * resources of the subscriber.
	 */
	public synchronized void start() {
		// Set the started flag to enable event queuing.
		// We are guaranteed to be the first since this method is synchronized.
		started = true;
		ResourceTraversal[] traversals = scope.getTraversals();
		reset(traversals, SubscriberEvent.INITIALIZE);
		initializing = false;
	}

	@Override
	protected synchronized void queueEvent(Event event, boolean front) {
		// Only post events if the handler is started
		if (started) {
			super.queueEvent(event, front);
		}
	}
	/**
	 * Schedule the job or process the events now.
	 */
	@Override
	public void schedule() {
		Job job = getEventHandlerJob();
		if (job.getState() == Job.NONE) {
			if(progressGroup != null) {
				job.setSystem(false);
				job.setProgressGroup(progressGroup, ticks);
			} else {
				job.setSystem(isSystemJob());
			}
		}
		getEventHandlerJob().schedule();
	}

	protected boolean isSystemJob() {
		return !initializing;
	}


	@Override
	protected void jobDone(IJobChangeEvent event) {
		super.jobDone(event);
		progressGroup = null;
	}

	/**
	 * Called by a client to indicate that a resource has changed and its synchronization state
	 * should be recalculated.
	 * @param resource the changed resource
	 * @param depth the depth of the change calculation
	 */
	public void change(IResource resource, int depth) {
		queueEvent(new SubscriberEvent(resource, SubscriberEvent.CHANGE, depth), false);
	}

	/**
	 * Called by a client to indicate that a resource has been removed and should be removed. The
	 * removal will propagate to the set.
	 * @param resource the resource that was removed
	 */
	public void remove(IResource resource) {
		queueEvent(
			new SubscriberEvent(resource, SubscriberEvent.REMOVAL, IResource.DEPTH_INFINITE), false);
	}

	/**
	 * Collect the calculated synchronization information for the given resource at the given depth. The
	 * results are added to the provided list.
	 */
	private void collect(
		IResource resource,
		int depth,
		IProgressMonitor monitor) {

		Policy.checkCanceled(monitor);

		// handle any preemptive events before continuing
		handlePreemptiveEvents(monitor);

		if (resource.getType() != IResource.FILE
			&& depth != IResource.DEPTH_ZERO) {
			try {
				IResource[] members =
					getSubscriber().members(resource);
				for (int i = 0; i < members.length; i++) {
					collect(
						members[i],
						depth == IResource.DEPTH_INFINITE
							? IResource.DEPTH_INFINITE
							: IResource.DEPTH_ZERO,
						monitor);
				}
			} catch (TeamException e) {
				// We only handle the exception if the resource's project is accessible.
				// The project close delta will clean up.
				if (resource.getProject().isAccessible())
					handleException(e, resource, ITeamStatus.SYNC_INFO_SET_ERROR, NLS.bind(Messages.SubscriberEventHandler_8, new String[] { resource.getFullPath().toString(), e.getMessage() }));
			}
		}

		monitor.subTask(NLS.bind(Messages.SubscriberEventHandler_2, new String[] { resource.getFullPath().toString() }));
		try {
			handleChange(resource);
			handlePendingDispatch(monitor);
		} catch (CoreException e) {
			handleException(e, resource, ITeamStatus.RESOURCE_SYNC_INFO_ERROR, NLS.bind(Messages.SubscriberEventHandler_9, new String[] { resource.getFullPath().toString(), e.getMessage() }));
		}
		monitor.worked(1);
	}

	/**
	 * Return the subscriber associated with this event handler
	 * @return the subscriber associated with this event handler
	 */
	protected Subscriber getSubscriber() {
		return subscriber;
	}

	/**
	 * The given resource has changed. Subclasses should handle
	 * this in an appropriate fashion
	 * @param resource the resource whose state has changed
	 */
	protected abstract void handleChange(IResource resource) throws CoreException;

	protected void handlePendingDispatch(IProgressMonitor monitor) {
		if (isReadyForDispatch(false /*don't wait if queue is empty*/)) {
			try {
				dispatchEvents(Policy.subMonitorFor(monitor, 5));
			} catch (TeamException e) {
				handleException(e, null, ITeamStatus.SYNC_INFO_SET_ERROR, e.getMessage());
			}
		}
	}

	/**
	 * Handle the exception by returning it as a status from the job but also by
	 * dispatching it to the sync set input so any down stream views can react
	 * accordingly.
	 * The resource passed may be null.
	 */
	protected void handleException(CoreException e, IResource resource, int code, String message) {
		handleException(e);
	}

	/**
	 * Called to initialize to calculate the synchronization information using the optimized subscriber method. For
	 * subscribers that don't support the optimization, all resources in the subscriber are manually re-calculated.
	 * @param resource the resources to check
	 * @param depth the depth
	 * @param monitor
	 */
	protected abstract void collectAll(
		IResource resource,
		int depth,
		IProgressMonitor monitor);

	/**
	 * Feed the given events to the set. The appropriate method on the set is called
	 * for each event type.
	 * @param events
	 */
	protected abstract void dispatchEvents(SubscriberEvent[] events, IProgressMonitor monitor);

	/**
	 * Initialize all resources for the subscriber associated with the set. This will basically recalculate
	 * all synchronization information for the subscriber.
	 * @param type can be Event.CHANGE to recalculate all states or Event.INITIALIZE to perform the
	 *   optimized recalculation if supported by the subscriber.
	 */
	protected void reset(ResourceTraversal[] traversals, int type) {
		for (int i = 0; i < traversals.length; i++) {
			ResourceTraversal traversal = traversals[i];
			IResource[] resources = traversal.getResources();
			for (int j = 0; j < resources.length; j++) {
				queueEvent(new SubscriberEvent(resources[j], type, traversal.getDepth()), false);
			}
		}
	}

	@Override
	protected void processEvent(Event event, IProgressMonitor monitor) {
		try {
			// Cancellation is dangerous because this will leave the sync info in a bad state.
			// Purposely not checking -
			int type = event.getType();
			switch (type) {
				case BackgroundEventHandler.RUNNABLE_EVENT :
					executeRunnable(event, monitor);
					break;
				case SubscriberEvent.REMOVAL :
					queueDispatchEvent(event);
					break;
				case SubscriberEvent.CHANGE :
					collect(
					    event.getResource(),
					    ((ResourceEvent)event).getDepth(),
						monitor);
					break;
				case SubscriberEvent.INITIALIZE :
					monitor.subTask(NLS.bind(Messages.SubscriberEventHandler_2, new String[] { event.getResource().getFullPath().toString() }));
					collectAll(
					        event.getResource(),
					        ((ResourceEvent)event).getDepth(),
							Policy.subMonitorFor(monitor, 64));
					break;
			}
		} catch (OperationCanceledException e) {
			// the job has been canceled.
			// Clear the queue and propagate the cancellation through the sets.
			handleCancel(e);
		} catch (RuntimeException e) {
			// handle the exception and keep processing
			if (event.getType() == BackgroundEventHandler.RUNNABLE_EVENT ) {
				handleException(new TeamException(Messages.SubscriberEventHandler_10, e));
			} else {
				handleException(new TeamException(Messages.SubscriberEventHandler_10, e), event.getResource(), ITeamStatus.SYNC_INFO_SET_ERROR, NLS.bind(Messages.SubscriberEventHandler_11, new String[] { event.getResource().getFullPath().toString(), e.getMessage() }));
			}
		}
	}

	/**
	 * Queue the event to be handle during the dispatch phase.
	 * @param event the event
	 */
	protected void queueDispatchEvent(Event event) {
		resultCache.add(event);
	}

	/**
	 * Handle the cancel exception
	 * @param e the cancel exception
	 */
	protected void handleCancel(OperationCanceledException e) {
		resultCache.clear();
	}

	/*
	 * Execute the RunnableEvent
	 */
	private void executeRunnable(Event event, IProgressMonitor monitor) {
		try {
			// Dispatch any queued results to clear pending output events
			dispatchEvents(Policy.subMonitorFor(monitor, 1));
		} catch (TeamException e) {
			handleException(e, null, ITeamStatus.SYNC_INFO_SET_ERROR, e.getMessage());
		}
		try {
			((RunnableEvent)event).run(Policy.subMonitorFor(monitor, 1));
		} catch (CoreException e) {
			handleException(e, null, ITeamStatus.SYNC_INFO_SET_ERROR, e.getMessage());
		}
	}

	@Override
	protected boolean  doDispatchEvents(IProgressMonitor monitor) {
		if (!resultCache.isEmpty()) {
			dispatchEvents(resultCache.toArray(new SubscriberEvent[resultCache.size()]), monitor);
			resultCache.clear();
			return true;
		}
		return false;
	}

	/**
	 * Queue up the given runnable in an event to be processed by this job
	 *
	 * @param runnable
	 *            the runnable to be run by the handler
	 * @param frontOnQueue
	 *            the frontOnQueue flag is used to indicate that the runnable
	 *            should be placed on the front of the queue and be processed as
	 *            soon as possible
	 */
	public void run(IWorkspaceRunnable runnable, boolean frontOnQueue) {
		queueEvent(new RunnableEvent(runnable, frontOnQueue), frontOnQueue);
	}

	public void setProgressGroupHint(IProgressMonitor progressGroup, int ticks) {
		this.progressGroup = progressGroup;
		this.ticks = ticks;
	}

	protected void handlePreemptiveEvents(IProgressMonitor monitor) {
		Event event = peek();
		if (event instanceof RunnableEvent && ((RunnableEvent)event).isPreemtive()) {
			executeRunnable(nextElement(), monitor);
		}
	}

	/**
	 * Return the scope of this event handler. The scope is
	 * used to determine the resources that are processed by the handler
	 * @return the scope of this event handler
	 */
	protected ISynchronizationScope getScope() {
		return scope;
	}

	@Override
	public void shutdown() {
		super.shutdown();
		scope.removeScopeChangeListener(scopeChangeListener);
	}
}