D-Net

package eu.dnetlib.msro.worker.nodes;

import javax.annotation.Resource;

import org.apache.commons.lang.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.Engine;

import com.googlecode.sarasvati.NodeToken;

import com.googlecode.sarasvati.env.Env;

import eu.dnetlib.common.ifaces.BlackboardCallback;

import eu.dnetlib.common.services.locators.DnetServiceLocator;

import eu.dnetlib.enabling.blackboard.BlackboardDispatcher;

import eu.dnetlib.msro.worker.WorkflowConstants;

import eu.dnetlib.rmi.blackboard.AbstractBlackboardMessage;

public abstract class BlackboardJobNode<T> extends SarasvatiJobNode implements BlackboardCallback<T> {

	/**

	 * logger.

	 */

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

	@Resource

	private DnetServiceLocator serviceLocator;

	@Resource

	private BlackboardDispatcher blackboardDispatcher;

	private Engine engine;

	private NodeToken token;

	@Override

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

		super.execute(engine, token);

		this.engine = engine;

		this.token = token;

		log.info("executing blackboard node");

		try {

			token.getEnv().setAttribute(WorkflowConstants.BLACKBOARD_IS_BLACKBOARD, true);

			final String serviceId = obtainServiceId(token);

			if (StringUtils.isBlank(serviceId)) {

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

				final String msg = "cannot locate target service profile: " + serviceId;

				token.getEnv().setAttribute(WorkflowConstants.SYSTEM_ERROR, msg);

				log.error(msg);

				engine.complete(token, "failed");

				return;

			}

			blackboardDispatcher.createDispatcher(serviceId, prepareMessage(token), this);

		} catch (final Throwable e) {

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

			token.getEnv().setAttribute(WorkflowConstants.SYSTEM_ERROR, "cannot prepare blackboard job: " + e);

			engine.complete(token, "failed");

			log.error("cannot prepare blackboard job", e);

		}

	}

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

	abstract protected T prepareMessage(final NodeToken token) throws Exception;

	abstract protected void saveResponseInEnv(final T t, final Env env);

	@Override

	public void onSuccess(final T t) {

		log.debug("Blackboard workflow node DONE");

		saveResponseInEnv(t, token.getEnv());

		engine.complete(token, Arc.DEFAULT_ARC);

		engine.executeQueuedArcTokens(token.getProcess());

	}

	@Override

	public void onFail(final T t) {

		final String error = t instanceof AbstractBlackboardMessage ? ((AbstractBlackboardMessage) t).getError() : "Unknown Error";

		log.warn("Blackboard workflow node FAILED: " + error);

		saveResponseInEnv(t, token.getEnv());

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

		token.getEnv().setAttribute(WorkflowConstants.SYSTEM_ERROR, error);

		engine.complete(token, "failed");

	}

	@Override

	public void onOngoing(final T t) {

		getToken().getEnv().setAttribute(WorkflowConstants.BLACKBOARD_IS_GOING, true);

	}

	protected Engine getEngine() {

		return engine;

	}

	protected NodeToken getToken() {

		return token;

	}

	protected DnetServiceLocator getServiceLocator() {

		return serviceLocator;

	}

}