/**
 * Copyright (c) 2000-2013 Liferay, Inc. All rights reserved.
 *
 * The contents of this file are subject to the terms of the Liferay Enterprise
 * Subscription License ("License"). You may not use this file except in
 * compliance with the License. You can obtain a copy of the License by
 * contacting Liferay, Inc. See the License for the specific language governing
 * permissions and limitations under the License, including but not limited to
 * distribution rights of the Software.
 *
 *
 *
 */

package com.liferay.portal.scheduler;

import com.liferay.portal.cluster.ClusterableContextThreadLocal;
import com.liferay.portal.kernel.bean.BeanReference;
import com.liferay.portal.kernel.bean.IdentifiableBean;
import com.liferay.portal.kernel.bean.PortalBeanLocatorUtil;
import com.liferay.portal.kernel.cluster.BaseClusterMasterTokenTransitionListener;
import com.liferay.portal.kernel.cluster.ClusterExecutorUtil;
import com.liferay.portal.kernel.cluster.ClusterInvokeAcceptor;
import com.liferay.portal.kernel.cluster.ClusterInvokeThreadLocal;
import com.liferay.portal.kernel.cluster.ClusterMasterExecutorUtil;
import com.liferay.portal.kernel.cluster.ClusterMasterTokenTransitionListener;
import com.liferay.portal.kernel.cluster.ClusterRequest;
import com.liferay.portal.kernel.cluster.Clusterable;
import com.liferay.portal.kernel.log.Log;
import com.liferay.portal.kernel.log.LogFactoryUtil;
import com.liferay.portal.kernel.messaging.Message;
import com.liferay.portal.kernel.messaging.proxy.ProxyModeThreadLocal;
import com.liferay.portal.kernel.scheduler.SchedulerEngine;
import com.liferay.portal.kernel.scheduler.SchedulerEngineHelperUtil;
import com.liferay.portal.kernel.scheduler.SchedulerException;
import com.liferay.portal.kernel.scheduler.StorageType;
import com.liferay.portal.kernel.scheduler.Trigger;
import com.liferay.portal.kernel.scheduler.TriggerFactoryUtil;
import com.liferay.portal.kernel.scheduler.TriggerState;
import com.liferay.portal.kernel.scheduler.messaging.SchedulerResponse;
import com.liferay.portal.kernel.servlet.PluginContextLifecycleThreadLocal;
import com.liferay.portal.kernel.util.CharPool;
import com.liferay.portal.kernel.util.MethodHandler;
import com.liferay.portal.kernel.util.MethodKey;
import com.liferay.portal.kernel.util.ObjectValuePair;
import com.liferay.portal.kernel.util.StringBundler;
import com.liferay.portal.kernel.util.StringPool;
import com.liferay.portal.scheduler.quartz.QuartzTriggerHelperUtil;
import com.liferay.portal.spring.aop.ServiceBeanAopProxy;
import com.liferay.portal.util.PropsValues;

import java.io.Serializable;

import java.util.Date;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import org.springframework.aop.TargetSource;
import org.springframework.aop.framework.AdvisedSupport;

/**
 * @author Tina Tian
 */
public class ClusterSchedulerEngine
	implements IdentifiableBean, SchedulerEngine {

	public static SchedulerEngine createClusterSchedulerEngine(
		SchedulerEngine schedulerEngine) {

		if (PropsValues.CLUSTER_LINK_ENABLED && PropsValues.SCHEDULER_ENABLED) {
			schedulerEngine = new ClusterSchedulerEngine(schedulerEngine);
		}

		return schedulerEngine;
	}

	public ClusterSchedulerEngine(SchedulerEngine schedulerEngine) {
		_schedulerEngine = schedulerEngine;

		ReadWriteLock readWriteLock = new ReentrantReadWriteLock();

		_readLock = readWriteLock.readLock();
		_writeLock = readWriteLock.writeLock();
	}

	@Clusterable(acceptor = SchedulerClusterInvokeAcceptor.class)
	@Override
	public void delete(String groupName) throws SchedulerException {
		boolean memoryClusteredSlaveJob = isMemoryClusteredSlaveJob(groupName);

		_readLock.lock();

		try {
			if (memoryClusteredSlaveJob) {
				removeMemoryClusteredJobs(groupName);
			}
			else {
				_schedulerEngine.delete(groupName);
			}
		}
		finally {
			_readLock.unlock();
		}

		setClusterableThreadLocal(groupName);
	}

	@Clusterable(acceptor = SchedulerClusterInvokeAcceptor.class)
	@Override
	public void delete(String jobName, String groupName)
		throws SchedulerException {

		boolean memoryClusteredSlaveJob = isMemoryClusteredSlaveJob(groupName);

		_readLock.lock();

		try {
			if (memoryClusteredSlaveJob) {
				_memoryClusteredJobs.remove(getFullName(jobName, groupName));
			}
			else {
				_schedulerEngine.delete(jobName, groupName);
			}
		}
		finally {
			_readLock.unlock();
		}

		setClusterableThreadLocal(groupName);
	}

	@Override
	public String getBeanIdentifier() {
		return _beanIdentifier;
	}

	@Clusterable(onMaster = true)
	@Override
	public SchedulerResponse getScheduledJob(String jobName, String groupName)
		throws SchedulerException {

		_readLock.lock();

		try {
			return _schedulerEngine.getScheduledJob(jobName, groupName);
		}
		finally {
			_readLock.unlock();
		}
	}

	@Clusterable(onMaster = true)
	@Override
	public List<SchedulerResponse> getScheduledJobs()
		throws SchedulerException {

		_readLock.lock();

		try {
			return _schedulerEngine.getScheduledJobs();
		}
		finally {
			_readLock.unlock();
		}
	}

	@Clusterable(onMaster = true)
	@Override
	public List<SchedulerResponse> getScheduledJobs(String groupName)
		throws SchedulerException {

		_readLock.lock();

		try {
			return _schedulerEngine.getScheduledJobs(groupName);
		}
		finally {
			_readLock.unlock();
		}
	}

	@Clusterable(acceptor = SchedulerClusterInvokeAcceptor.class)
	@Override
	public void pause(String groupName) throws SchedulerException {
		boolean memoryClusteredSlaveJob = isMemoryClusteredSlaveJob(groupName);

		_readLock.lock();

		try {
			if (memoryClusteredSlaveJob) {
				updateMemoryClusteredJobs(groupName, TriggerState.PAUSED);
			}
			else {
				_schedulerEngine.pause(groupName);
			}
		}
		finally {
			_readLock.unlock();
		}

		setClusterableThreadLocal(groupName);
	}

	@Clusterable(acceptor = SchedulerClusterInvokeAcceptor.class)
	@Override
	public void pause(String jobName, String groupName)
		throws SchedulerException {

		boolean memoryClusteredSlaveJob = isMemoryClusteredSlaveJob(groupName);

		_readLock.lock();

		try {
			if (memoryClusteredSlaveJob) {
				updateMemoryClusteredJob(
					jobName, groupName, TriggerState.PAUSED);
			}
			else {
				_schedulerEngine.pause(jobName, groupName);
			}
		}
		finally {
			_readLock.unlock();
		}

		setClusterableThreadLocal(groupName);
	}

	@Clusterable(acceptor = SchedulerClusterInvokeAcceptor.class)
	@Override
	public void resume(String groupName) throws SchedulerException {
		boolean memoryClusteredSlaveJob = isMemoryClusteredSlaveJob(groupName);

		_readLock.lock();

		try {
			if (memoryClusteredSlaveJob) {
				updateMemoryClusteredJobs(groupName, TriggerState.NORMAL);
			}
			else {
				_schedulerEngine.resume(groupName);
			}
		}
		finally {
			_readLock.unlock();
		}

		setClusterableThreadLocal(groupName);
	}

	@Clusterable(acceptor = SchedulerClusterInvokeAcceptor.class)
	@Override
	public void resume(String jobName, String groupName)
		throws SchedulerException {

		boolean memoryClusteredSlaveJob = isMemoryClusteredSlaveJob(groupName);

		_readLock.lock();

		try {
			if (memoryClusteredSlaveJob) {
				updateMemoryClusteredJob(
					jobName, groupName, TriggerState.NORMAL);
			}
			else {
				_schedulerEngine.resume(jobName, groupName);
			}
		}
		finally {
			_readLock.unlock();
		}

		setClusterableThreadLocal(groupName);
	}

	@Clusterable(acceptor = SchedulerClusterInvokeAcceptor.class)
	@Override
	public void schedule(
			Trigger trigger, String description, String destinationName,
			Message message)
		throws SchedulerException {

		String groupName = trigger.getGroupName();

		ObjectValuePair<String, StorageType> objectValuePair = resolveGroupName(
			groupName);

		StorageType storageType = objectValuePair.getValue();

		_readLock.lock();

		try {
			if (storageType == StorageType.MEMORY_CLUSTERED) {
				String jobName = trigger.getJobName();

				if (ClusterMasterExecutorUtil.isMaster()) {
					_schedulerEngine.schedule(
						trigger, description, destinationName, message);

					if (_portalReady) {
						SchedulerResponse schedulerResponse =
							new SchedulerResponse();

						schedulerResponse.setDescription(description);
						schedulerResponse.setDestinationName(destinationName);
						schedulerResponse.setGroupName(groupName);
						schedulerResponse.setJobName(jobName);
						schedulerResponse.setMessage(message);
						schedulerResponse.setStorageType(storageType);
						schedulerResponse.setTrigger(trigger);

						_notifySlave(
							_addMemoryClusteredJobMethodKey, schedulerResponse,
							_beanIdentifier);
					}
				}
				else {
					ObjectValuePair<SchedulerResponse, TriggerState> value =
						_memoryClusteredJobs.get(
							getFullName(jobName, groupName));

					if (value == null) {
						MethodHandler methodHandler = new MethodHandler(
							_getScheduledJobMethodKey, jobName,
							objectValuePair.getKey(),
							objectValuePair.getValue());

						try {
							Future<SchedulerResponse> future =
								ClusterMasterExecutorUtil.executeOnMaster(
									methodHandler);

							SchedulerResponse schedulerResponse = future.get(
								PropsValues.
								CLUSTERABLE_ADVICE_CALL_MASTER_TIMEOUT,
								TimeUnit.SECONDS);

							if ((schedulerResponse == null) ||
								(schedulerResponse.getTrigger() == null)) {

								if (_log.isInfoEnabled()) {
									_log.info(
										"Memory clustered job is not yet " +
											"deployed on master");
								}
							}
							else {
								addMemoryClusteredJob(schedulerResponse);
							}
						}
						catch (Exception e) {
							_log.error("Unable to get response from master", e);
						}
					}
				}
			}
			else {
				_schedulerEngine.schedule(
					trigger, description, destinationName, message);
			}
		}
		finally {
			_readLock.unlock();
		}

		setClusterableThreadLocal(groupName);
	}

	@Override
	public void setBeanIdentifier(String beanIdentifier) {
		_beanIdentifier = beanIdentifier;
	}

	@Override
	public void shutdown() throws SchedulerException {
		_portalReady = false;

		ClusterMasterExecutorUtil.
			unregisterClusterMasterTokenTransitionListener(
				_schedulerClusterMasterTokenTransitionListener);

		_schedulerEngine.shutdown();
	}

	@Override
	public void start() throws SchedulerException {
		if (!ClusterMasterExecutorUtil.isMaster()) {
			initMemoryClusteredJobs();
		}

		_schedulerClusterMasterTokenTransitionListener =
			new SchedulerClusterMasterTokenTransitionListener();

		ClusterMasterExecutorUtil.registerClusterMasterTokenTransitionListener(
			_schedulerClusterMasterTokenTransitionListener);

		_schedulerEngine.start();

		_portalReady = true;
	}

	@Clusterable(acceptor = SchedulerClusterInvokeAcceptor.class)
	@Override
	public void suppressError(String jobName, String groupName)
		throws SchedulerException {

		boolean memoryClusteredSlaveJob = isMemoryClusteredSlaveJob(groupName);

		if (!memoryClusteredSlaveJob) {
			_readLock.lock();

			try {
				_schedulerEngine.suppressError(jobName, groupName);
			}
			finally {
				_readLock.unlock();
			}
		}

		setClusterableThreadLocal(groupName);
	}

	@Clusterable(acceptor = SchedulerClusterInvokeAcceptor.class)
	@Override
	public void unschedule(String groupName) throws SchedulerException {
		boolean memoryClusteredSlaveJob = isMemoryClusteredSlaveJob(groupName);

		_readLock.lock();

		try {
			if (memoryClusteredSlaveJob) {
				removeMemoryClusteredJobs(groupName);
			}
			else {
				_schedulerEngine.unschedule(groupName);
			}
		}
		finally {
			_readLock.unlock();
		}

		setClusterableThreadLocal(groupName);
	}

	@Clusterable(acceptor = SchedulerClusterInvokeAcceptor.class)
	@Override
	public void unschedule(String jobName, String groupName)
		throws SchedulerException {

		boolean memoryClusteredSlaveJob = isMemoryClusteredSlaveJob(groupName);

		_readLock.lock();

		try {
			if (memoryClusteredSlaveJob) {
				_memoryClusteredJobs.remove(getFullName(jobName, groupName));
			}
			else {
				_schedulerEngine.unschedule(jobName, groupName);
			}
		}
		finally {
			_readLock.unlock();
		}

		setClusterableThreadLocal(groupName);
	}

	@Clusterable(acceptor = SchedulerClusterInvokeAcceptor.class)
	@Override
	public void update(Trigger trigger) throws SchedulerException {
		String jobName = trigger.getJobName();
		String groupName = trigger.getGroupName();

		boolean memoryClusteredSlaveJob = isMemoryClusteredSlaveJob(groupName);

		_readLock.lock();

		try {
			if (memoryClusteredSlaveJob) {
				boolean updated = false;

				for (ObjectValuePair<SchedulerResponse, TriggerState>
						memoryClusteredJob : _memoryClusteredJobs.values()) {

					SchedulerResponse schedulerResponse =
						memoryClusteredJob.getKey();

					if (jobName.equals(schedulerResponse.getJobName()) &&
						groupName.equals(schedulerResponse.getGroupName())) {

						schedulerResponse.setTrigger(trigger);

						updated = true;

						break;
					}
				}

				if (!updated) {
					throw new SchedulerException(
						"Unable to update trigger for memory clustered job");
				}
			}
			else {
				_schedulerEngine.update(trigger);
			}
		}
		finally {
			_readLock.unlock();
		}

		setClusterableThreadLocal(groupName);
	}

	protected void addMemoryClusteredJob(SchedulerResponse schedulerResponse)
		throws SchedulerException {

		Trigger oldTrigger = schedulerResponse.getTrigger();

		String jobName = schedulerResponse.getJobName();
		String groupName = SchedulerEngineHelperUtil.namespaceGroupName(
			schedulerResponse.getGroupName(), StorageType.MEMORY_CLUSTERED);

		Trigger newTrigger = TriggerFactoryUtil.buildTrigger(
			oldTrigger.getTriggerType(), jobName, groupName,
			oldTrigger.getStartDate(), oldTrigger.getEndDate(),
			oldTrigger.getTriggerContent());

		schedulerResponse.setTrigger(newTrigger);

		TriggerState triggerState = SchedulerEngineHelperUtil.getJobState(
			schedulerResponse);

		Message message = schedulerResponse.getMessage();

		message.remove(JOB_STATE);

		_memoryClusteredJobs.put(
			getFullName(jobName, groupName),
			new ObjectValuePair<SchedulerResponse, TriggerState>(
				schedulerResponse, triggerState));
	}

	protected String getFullName(String jobName, String groupName) {
		return groupName.concat(StringPool.PERIOD).concat(jobName);
	}

	protected void initMemoryClusteredJobs() {
		MethodHandler methodHandler = new MethodHandler(
			_getScheduledJobsMethodKey, StorageType.MEMORY_CLUSTERED);

		while (!ClusterMasterExecutorUtil.isMaster()) {
			try {
				Future<List<SchedulerResponse>> future =
					ClusterMasterExecutorUtil.executeOnMaster(methodHandler);

				List<SchedulerResponse> schedulerResponses = future.get(
					PropsValues.CLUSTERABLE_ADVICE_CALL_MASTER_TIMEOUT,
					TimeUnit.SECONDS);

				_memoryClusteredJobs.clear();

				for (SchedulerResponse schedulerResponse : schedulerResponses) {
					addMemoryClusteredJob(schedulerResponse);
				}

				if (_log.isInfoEnabled()) {
					_log.info(
						"Load " + schedulerResponses.size() +
							" memory clustered jobs from master");
				}

				return;
			}
			catch (InterruptedException ie) {
				if (_log.isWarnEnabled()) {
					_log.warn(
						"Give up the master response waiting due to " +
							"interruption",
						ie);
				}

				return;
			}
			catch (Exception e) {
				StringBundler sb = new StringBundler(5);

				sb.append(
					"Unable to load memory clustered jobs from master in ");
				sb.append( PropsValues.CLUSTERABLE_ADVICE_CALL_MASTER_TIMEOUT);
				sb.append(" seconds, you might need to increase value set to ");
				sb.append("\"clusterable.advice.call.master.timeout\", ");
				sb.append("will retry again");

				_log.error(sb.toString(), e);
			}
		}
	}

	protected boolean isMemoryClusteredSlaveJob(String groupName)
		throws SchedulerException {

		ObjectValuePair<String, StorageType> objectValuePair = resolveGroupName(
			groupName);

		StorageType storageType = objectValuePair.getValue();

		if ((storageType != StorageType.MEMORY_CLUSTERED) ||
			ClusterMasterExecutorUtil.isMaster()) {

			return false;
		}

		return true;
	}

	protected void removeMemoryClusteredJobs(String groupName) {
		Set<Map.Entry<String, ObjectValuePair<SchedulerResponse, TriggerState>>>
			memoryClusteredJobs = _memoryClusteredJobs.entrySet();

		Iterator
			<Map.Entry<String,
				ObjectValuePair<SchedulerResponse, TriggerState>>> itr =
					memoryClusteredJobs.iterator();

		while (itr.hasNext()) {
			Map.Entry <String, ObjectValuePair<SchedulerResponse, TriggerState>>
				entry = itr.next();

			ObjectValuePair<SchedulerResponse, TriggerState>
				memoryClusteredJob = entry.getValue();

			SchedulerResponse schedulerResponse = memoryClusteredJob.getKey();

			if (groupName.equals(schedulerResponse.getGroupName())) {
				itr.remove();
			}
		}
	}

	protected ObjectValuePair<String, StorageType> resolveGroupName(
		String groupName) {

		int index = groupName.indexOf(CharPool.POUND);

		String storageTypeString = groupName.substring(0, index);

		StorageType storageType = StorageType.valueOf(storageTypeString);

		String orginalGroupName = groupName.substring(index + 1);

		return new ObjectValuePair<String, StorageType>(
			orginalGroupName, storageType);
	}

	protected void setClusterableThreadLocal(String groupName) {
		ObjectValuePair<String, StorageType> objectValuePair = resolveGroupName(
			groupName);

		ClusterableContextThreadLocal.putThreadLocalContext(
			STORAGE_TYPE, objectValuePair.getValue());
		ClusterableContextThreadLocal.putThreadLocalContext(
			_PORTAL_READY, _portalReady);

		boolean pluginReady = true;

		if (PluginContextLifecycleThreadLocal.isInitializing() ||
			PluginContextLifecycleThreadLocal.isDestroying()) {

			pluginReady = false;
		}

		ClusterableContextThreadLocal.putThreadLocalContext(
			_PLUGIN_READY, pluginReady);
	}

	protected void updateMemoryClusteredJob(
		String jobName, String groupName, TriggerState triggerState) {

		ObjectValuePair<SchedulerResponse, TriggerState>
			memoryClusteredJob = _memoryClusteredJobs.get(
				getFullName(jobName, groupName));

		if (memoryClusteredJob != null) {
			memoryClusteredJob.setValue(triggerState);
		}
	}

	protected void updateMemoryClusteredJobs(
		String groupName, TriggerState triggerState) {

		for (ObjectValuePair<SchedulerResponse, TriggerState>
				memoryClusteredJob : _memoryClusteredJobs.values()) {

			SchedulerResponse schedulerResponse = memoryClusteredJob.getKey();

			if (groupName.equals(schedulerResponse.getGroupName())) {
				memoryClusteredJob.setValue(triggerState);
			}
		}
	}

	private static void _addMemoryClusteredJob(
			SchedulerResponse schedulerResponse, String beanIdentifier)
		throws Exception {

		Object serviceBean = PortalBeanLocatorUtil.locate(beanIdentifier);

		AdvisedSupport advisedSupport = ServiceBeanAopProxy.getAdvisedSupport(
			serviceBean);

		TargetSource targetSource = advisedSupport.getTargetSource();

		ClusterSchedulerEngine clusterSchedulerEngine =
			(ClusterSchedulerEngine)targetSource.getTarget();

		if (!clusterSchedulerEngine._portalReady) {
			return;
		}

		String jobName = schedulerResponse.getJobName();
		String groupName = schedulerResponse.getGroupName();

		Lock writeLock = clusterSchedulerEngine._writeLock;

		writeLock.lock();

		try {
			Map<String, ObjectValuePair<SchedulerResponse, TriggerState>>
				memoryClusteredJobs =
					clusterSchedulerEngine._memoryClusteredJobs;

			memoryClusteredJobs.put(
				clusterSchedulerEngine.getFullName(jobName, groupName),
				new ObjectValuePair<SchedulerResponse, TriggerState>(
					schedulerResponse, TriggerState.NORMAL));

			if (_log.isInfoEnabled()) {
				_log.info(
					"Receive notification from master, add memory clustered " +
						"job " + schedulerResponse);
			}
		}
		finally {
			writeLock.unlock();
		}
	}

	private static void _reloadMemoryClusteredJobs(String beanIdentifier)
		throws Exception {

		Object serviceBean = PortalBeanLocatorUtil.locate(beanIdentifier);

		AdvisedSupport advisedSupport = ServiceBeanAopProxy.getAdvisedSupport(
			serviceBean);

		TargetSource targetSource = advisedSupport.getTargetSource();

		ClusterSchedulerEngine clusterSchedulerEngine =
			(ClusterSchedulerEngine)targetSource.getTarget();

		if (!clusterSchedulerEngine._portalReady) {
			return;
		}

		if (ClusterMasterExecutorUtil.isMaster()) {
			return;
		}

		Lock writeLock = clusterSchedulerEngine._writeLock;

		writeLock.lock();

		try {
			clusterSchedulerEngine.initMemoryClusteredJobs();

			if (_log.isInfoEnabled()) {
				_log.info(
					"Receive notification from master, reload memory " +
						"clustered jobs");
			}
		}
		finally {
			writeLock.unlock();
		}
	}

	private void _notifySlave(MethodKey methodKey, Object... arguments) {
		try {
			MethodHandler methodHandler = new MethodHandler(
				methodKey, arguments);

			ClusterRequest clusterRequest =
				ClusterRequest.createMulticastRequest(methodHandler, true);

			clusterRequest.setFireAndForget(true);

			ClusterExecutorUtil.execute(clusterRequest);
		}
		catch (Throwable t) {
			_log.error("Unable to notify slave", t);
		}
	}

	@BeanReference(
		name = "com.liferay.portal.scheduler.ClusterSchedulerEngineService")
	protected SchedulerEngine schedulerEngine;

	private static final String _PLUGIN_READY = "plugin.ready";

	private static final String _PORTAL_READY = "portal.ready";

	private static Log _log = LogFactoryUtil.getLog(
		ClusterSchedulerEngine.class);

	private static final MethodKey _addMemoryClusteredJobMethodKey =
		new MethodKey(
			ClusterSchedulerEngine.class, "_addMemoryClusteredJob",
			SchedulerResponse.class, String.class);
	private static final MethodKey _getScheduledJobMethodKey = new MethodKey(
		SchedulerEngineHelperUtil.class, "getScheduledJob", String.class,
		String.class, StorageType.class);
	private static final MethodKey _getScheduledJobsMethodKey = new MethodKey(
		SchedulerEngineHelperUtil.class, "getScheduledJobs", StorageType.class);
	private static final MethodKey _reloadMemoryClusteredJobsMethodKey =
		new MethodKey(
		ClusterSchedulerEngine.class, "_reloadMemoryClusteredJobs",
		String.class);
	private String _beanIdentifier;
	private final Map<String, ObjectValuePair<SchedulerResponse, TriggerState>>
		_memoryClusteredJobs = new ConcurrentHashMap
			<String, ObjectValuePair<SchedulerResponse, TriggerState>>();
	private boolean _portalReady;
	private final Lock _readLock;
	private ClusterMasterTokenTransitionListener
		_schedulerClusterMasterTokenTransitionListener;
	private final SchedulerEngine _schedulerEngine;
	private final Lock _writeLock;

	private static class SchedulerClusterInvokeAcceptor
		implements ClusterInvokeAcceptor {

		@Override
		public boolean accept(Map<String, Serializable> context) {
			if (ClusterInvokeThreadLocal.isEnabled()) {
				return false;
			}

			StorageType storageType = (StorageType)context.get(STORAGE_TYPE);
			boolean portalReady = (Boolean)context.get(_PORTAL_READY);
			boolean pluginReady = (Boolean)context.get(_PLUGIN_READY);

			if (storageType.equals(StorageType.PERSISTED) || !portalReady ||
				!pluginReady) {

				return false;
			}

			return true;
		}

	}

	private class SchedulerClusterMasterTokenTransitionListener
		extends BaseClusterMasterTokenTransitionListener {

		@Override
		protected void doMasterTokenAcquired() throws Exception {
			boolean forceSync = ProxyModeThreadLocal.isForceSync();

			ProxyModeThreadLocal.setForceSync(true);

			_writeLock.lock();

			try {
				for (ObjectValuePair<SchedulerResponse, TriggerState>
						memoryClusteredJob : _memoryClusteredJobs.values()) {

					SchedulerResponse schedulerResponse =
						memoryClusteredJob.getKey();

					Trigger oldTrigger = schedulerResponse.getTrigger();

					Date startDate = QuartzTriggerHelperUtil.getFireTimeAfter(
						oldTrigger, new Date());

					Trigger newTrigger = TriggerFactoryUtil.buildTrigger(
						oldTrigger.getTriggerType(), oldTrigger.getJobName(),
						oldTrigger.getGroupName(), startDate,
						oldTrigger.getEndDate(),
						oldTrigger.getTriggerContent());

					_schedulerEngine.schedule(
						newTrigger, schedulerResponse.getDescription(),
						schedulerResponse.getDestinationName(),
						schedulerResponse.getMessage());

					TriggerState triggerState = memoryClusteredJob.getValue();

					if (triggerState.equals(TriggerState.PAUSED)) {
						_schedulerEngine.pause(
							schedulerResponse.getJobName(),
							schedulerResponse.getGroupName());
					}
				}

				if (_log.isInfoEnabled()) {
					_log.info(
						_memoryClusteredJobs.size() +
							" MEMORY_CLUSTERED jobs started running on this " +
								"node");
				}

				_memoryClusteredJobs.clear();

				_notifySlave(
					_reloadMemoryClusteredJobsMethodKey, _beanIdentifier);
			}
			finally {
				ProxyModeThreadLocal.setForceSync(forceSync);

				_writeLock.unlock();
			}
		}

		@Override
		protected void doMasterTokenReleased() throws Exception {
			_writeLock.lock();

			int count = 0;

			try {
				initMemoryClusteredJobs();

				if (ClusterMasterExecutorUtil.isMaster()) {
					if (_log.isInfoEnabled()) {
						_log.info(
							"Current node is elected as master again, stop " +
								"transferring it to slave, all jobs running " +
									"on current node will keep running");
					}

					return;
				}

				for (SchedulerResponse schedulerResponse :
						_schedulerEngine.getScheduledJobs()) {

					if (StorageType.MEMORY_CLUSTERED ==
							schedulerResponse.getStorageType()) {

						String groupName =
							SchedulerEngineHelperUtil.namespaceGroupName(
								schedulerResponse.getGroupName(),
								StorageType.MEMORY_CLUSTERED);

						_schedulerEngine.delete(
							schedulerResponse.getJobName(), groupName);

						count++;
					}
				}

				if (_log.isInfoEnabled()) {
					_log.info(
						count +
							" MEMORY_CLUSTERED jobs stopped running on this " +
								"node");
				}

				_notifySlave(
					_reloadMemoryClusteredJobsMethodKey, _beanIdentifier);
			}
			finally {
				_writeLock.unlock();
			}
		}

	}

}