D-Net

package eu.dnetlib.index.action;

import java.util.UUID;

import eu.dnetlib.enabling.resultset.client.ResultSetClient;

import eu.dnetlib.enabling.tools.blackboard.BlackboardJob;

import eu.dnetlib.enabling.tools.blackboard.BlackboardServerAction;

import eu.dnetlib.enabling.tools.blackboard.BlackboardServerHandler;

import eu.dnetlib.index.actors.*;

import eu.dnetlib.index.feed.FeedMode;

import eu.dnetlib.rmi.common.ResultSet;

import eu.dnetlib.rmi.provision.IndexServiceException;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.quartz.*;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.beans.factory.annotation.Required;

import static org.quartz.SimpleScheduleBuilder.simpleSchedule;

import static org.quartz.TriggerBuilder.newTrigger;

/**

 * The Class FeedIndexAction.

 */

public class FeedIndexAction extends AbstractIndexAction implements BlackboardServerAction<IndexAction> {

	/**

	 * The Constant log.

	 */

	private static final Log log = LogFactory.getLog(FeedIndexAction.class);

	/**

	 * The actor map.

	 */

	@Autowired

	private ActorMap actorMap;

	/**

	 * The feed actor factory.

	 */

	@Autowired

	private IndexFeedActorFactory feedActorFactory;

	/**

	 * The result set client factory.

	 */

	@Autowired

	private transient ResultSetClient resultSetClient;

	/**

	 * Start deferred jobs, used to keep resultsets alive.

	 */

	private Scheduler jobScheduler;

	/**

	 * ResultSet keep alive JobDetail.

	 */

	private transient JobDetail rsKeepaliveJob;

	/**

	 * ResultSet keepAlive trigger's repeat delay.

	 */

	private long repeatDelay;

	/**

	 * {@inheritDoc}

	 *

	 * @see BlackboardServerAction#execute(BlackboardServerHandler,

	 * BlackboardJob)

	 */

	@Override

	public void execute(final BlackboardServerHandler handler, final BlackboardJob job) throws Exception {

		handler.ongoing(job);

		log.info("FEED job set to ONGOING");

		final JobDetail tmp = getJobScheduler().getJobDetail(new JobKey(ResultsetKeepAliveJob.JOB_NAME, ResultsetKeepAliveJob.JOB_GROUP));

		final String epr = getEpr(job);

		final String triggerId = UUID.randomUUID().toString();

		final String dsId = getIndexDSId(job);

		final FeedMode feedMode = getFeedMode(job);

		final String backendId = getBackend(job);

		final boolean emptyResult = getEmptyResult(job);

		if (backendId == null) throw new IndexServiceException("No backend identifier information in CREATE message");

		if (tmp == null) {

			log.fatal("re-registering job detail");

			getJobScheduler().addJob(getRsKeepaliveJob(), true);

		}

		log.debug("\n\n scheduling resultSet keepalive trigger: " + triggerId + "\n\n");

		getJobScheduler().scheduleJob(getResultsetTrigger(epr, triggerId));

		if (!actorMap.hasActor(backendId)) {

			actorMap.addActor(backendId, feedActorFactory.newInstance());

		}

		final ResultSet<?> resultSet = ResultSet.fromJson(epr);

		final Iterable<String> records = resultSetClient.iter(resultSet, String.class);

		actorMap.getActor(backendId)

				.feedIndex(dsId, feedMode, records, newRSKeepAliveCallback(triggerId), newBBActorCallback(handler, job), backendId, emptyResult);

	}

	private boolean getEmptyResult(BlackboardJob job) {

		if (job.getParameters().containsKey("emptyResult")) {

			final String emptyResult = job.getParameters().get("emptyResult").toLowerCase().trim();

			return "true".equals(emptyResult);

		} else return false;

	}

	/**

	 * Constructor for triggers used by the resultSet keepAlive job.

	 *

	 * @param rsEpr     resultSet epr to keep alive.

	 * @param triggerId trigger identifier.

	 * @return a new org.quartz.SimpleTrigger instance.

	 */

	private SimpleTrigger getResultsetTrigger(final String rsEpr, final String triggerId) {

		final SimpleTrigger trigger = newTrigger()

				.withIdentity(triggerId, ResultsetKeepAliveJob.JOB_GROUP)

				.withSchedule(simpleSchedule().repeatForever().withIntervalInMilliseconds(getRepeatDelay())).forJob(getRsKeepaliveJob()).build();

		trigger.getJobDataMap().put(BBParam.RS_EPR, rsEpr);

		return trigger;

	}

	/**

	 * Constructor for a blackboard callback to handle the job termination.

	 *

	 * @param handler the handler

	 * @param job     the BB job.

	 * @return a new eu.dnetlib.functionality.index.solr.actors.BlackboardActorCallback

	 */

	private BlackboardActorCallback newBBActorCallback(final BlackboardServerHandler handler, final BlackboardJob job) {

		return new BlackboardActorCallback() {

			@Override

			public void setJobDone() {

				log.info(job.getAction() + " job set to DONE");

				handler.done(job);

			}

			@Override

			public void setJobFailed(final Throwable exception) {

				log.error(job.getAction() + " job set to FAILED ", exception);

				handler.failed(job, exception);

			}

		};

	}

	/**

	 * Constructor for a resultSet keepAlive callbacks.

	 *

	 * @param triggerId trigger identifier.

	 * @return a new eu.dnetlib.functionality.index.solr.actors.ResultsetKeepAliveCallback

	 */

	private ResultsetKeepAliveCallback newRSKeepAliveCallback(final String triggerId) {

		return () -> {

			try {

				log.info("\n\n unscheduling resultSet keepalive trigger: " + triggerId + "\n\n");

				jobScheduler.unscheduleJob(new TriggerKey(triggerId, ResultsetKeepAliveJob.JOB_GROUP));

			} catch (SchedulerException e) {

				log.warn("cannot unschedule RSKeepAlive triggerId: " + triggerId);

				throw new RuntimeException(e); // NOPMD

			}

		};

	}

	/**

	 * Gets the rs keepalive job.

	 *

	 * @return the rsKeepaliveJob

	 */

	public JobDetail getRsKeepaliveJob() {

		return rsKeepaliveJob;

	}

	/**

	 * Sets the rs keepalive job.

	 *

	 * @param rsKeepaliveJob the rsKeepaliveJob to set

	 */

	@Required

	public void setRsKeepaliveJob(final JobDetail rsKeepaliveJob) {

		this.rsKeepaliveJob = rsKeepaliveJob;

	}

	/**

	 * Gets the job scheduler.

	 *

	 * @return the jobScheduler

	 */

	public Scheduler getJobScheduler() {

		return jobScheduler;

	}

	/**

	 * Sets the job scheduler.

	 *

	 * @param jobScheduler the jobScheduler to set

	 */

	@Required

	public void setJobScheduler(final Scheduler jobScheduler) {

		this.jobScheduler = jobScheduler;

	}

	/**

	 * Gets the repeat delay.

	 *

	 * @return the repeatDelay

	 */

	public long getRepeatDelay() {

		return repeatDelay;

	}

	/**

	 * Sets the repeat delay.

	 *

	 * @param repeatDelay the repeatDelay to set

	 */

	@Required

	public void setRepeatDelay(final long repeatDelay) {

		this.repeatDelay = repeatDelay;

	}

}