D-Net

package eu.dnetlib.msro.workflows.nodes;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.Executors;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import com.googlecode.sarasvati.Engine;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.msro.workflows.util.WorkflowsConstants;

public abstract class AsyncJobNode extends SarasvatiJobNode {

	/**

	 * logger.

	 */

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

	private final ExecutorService executor = Executors.newCachedThreadPool();

	@Override

	public void execute(final Engine engine, final NodeToken token) {

		super.execute(engine, token);

		log.info("executing async node");

		executor.execute(new Runnable() {

			@Override

			public void run() {

				try {

					log.debug("START NODE: " + getBeanName());

					beforeStart(token);

					String arc = execute(token);

					beforeCompleted(token);

					log.debug("END NODE (SUCCESS): " + getBeanName());

					engine.complete(token, arc);

				} catch (Throwable e) {

					log.error("got exception while executing workflow node", e);

					log.debug("END NODE (FAILED): " + getBeanName());

					beforeFailed(token);

					token.getEnv().setAttribute(WorkflowsConstants.SYSTEM_HAS_FAILED, true);

					token.getEnv().setAttribute(WorkflowsConstants.SYSTEM_ERROR, e.getMessage());

					engine.complete(token, "failed");

				}

			}

		});

	}

	abstract protected String execute(final NodeToken token) throws Exception;

	protected void beforeStart(final NodeToken token) {}

	protected void beforeCompleted(final NodeToken token) {}

	protected void beforeFailed(final NodeToken token) {}

}