D-Net

package eu.dnetlib.msro.openaireplus.workflows.nodes.actions;

import java.util.List;

import java.util.Map;

import com.google.common.base.Predicate;

import com.google.common.collect.Maps;

import com.google.gson.Gson;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.actionmanager.set.RawSet;

import eu.dnetlib.enabling.is.lookup.rmi.ISLookUpException;

import eu.dnetlib.enabling.is.lookup.rmi.ISLookUpService;

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

import eu.dnetlib.miscutils.datetime.DateUtils;

import eu.dnetlib.msro.rmi.MSROException;

import eu.dnetlib.msro.workflows.nodes.SimpleJobNode;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.apache.hadoop.fs.Path;

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

public class PrepareActionSetsJobNode extends SimpleJobNode {

	/**

	 * logger.

	 */

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

	private String ACTIONSET_PATH_XQUERY_TEMPLATE =

			"for $x in collection('/db/DRIVER/ActionManagerSetDSResources/ActionManagerSetDSResourceType') "

					+ "return $x/RESOURCE_PROFILE/BODY/SET[@id = '%s']/@directory/string()";

	@Autowired

	private UniqueServiceLocator serviceLocator;

	private String sets;

	@Override

	protected String execute(final NodeToken token) throws Exception {

		final List<Map<String, String>> setList = getSetList();

		final String now = DateUtils.now_ISO8601();

		final ISLookUpService lookUpService = serviceLocator.getService(ISLookUpService.class);

		final String basePath = lookUpService.getResourceProfileByQuery(

				"/RESOURCE_PROFILE[./HEADER/RESOURCE_TYPE/@value='ActionManagerServiceResourceType']//SERVICE_PROPERTIES/PROPERTY[@key='basePath']/@value/string()");

		for (Map<String, String> set : setList) {

			set.put("rawset", RawSet.newInstance().getId());

			set.put("creationDate", now);

			set.put("path", getFullPath(basePath, set, lookUpService));

			if (set.get("enabled").equals("true")) {

				log.info("preparing set: " + simplifySetInfo(set));

			}

			// setting the job properties needed to name the rawsets

			token.getEnv().setAttribute(set.get("jobProperty"), set.get("rawset"));

		}

		token.getEnv().setAttribute("sets", new Gson().toJson(setList));

		token.getEnv().setAttribute("actionManagerBasePath", basePath);

		return Arc.DEFAULT_ARC;

	}

	private String getFullPath(final String basePath, final Map<String, String> set, final ISLookUpService lookUpService) throws MSROException {

		return new Path(basePath + "/" + getPath(set.get("set"), lookUpService) + "/" + set.get("rawset")).toString();

	}

	private String getPath(final String setId, final ISLookUpService lookUpService) throws MSROException {

		try {

			return lookUpService.getResourceProfileByQuery(String.format(ACTIONSET_PATH_XQUERY_TEMPLATE, setId));

		} catch (ISLookUpException e) {

			throw new MSROException(String.format("Error obtaining directory from ActionSet profile %s", setId), e);

		}

	}

	private Map<String, String> simplifySetInfo(final Map<String, String> set) {

		return Maps.filterKeys(set, new Predicate<String>() {

			@Override

			public boolean apply(final String k) {

				return k.equals("set") || k.equals("rawset");

			}

		});

	}

	@SuppressWarnings("unchecked")

	protected List<Map<String, String>> getSetList() {

		return new Gson().fromJson(getSets(), List.class);

	}

	public String getSets() {

		return sets;

	}

	public void setSets(final String sets) {

		this.sets = sets;

	}

}