D-Net

package eu.dnetlib.actionmanager;

import java.util.List;

import javax.annotation.Resource;

import javax.jws.WebParam;

import eu.dnetlib.data.hadoop.HadoopServiceCore;

import eu.dnetlib.data.hadoop.config.ClusterName;

import eu.dnetlib.data.hadoop.rmi.HadoopServiceException;

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

import org.apache.hadoop.fs.Path;

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

import eu.dnetlib.actionmanager.actions.XsltInfoPackageAction;

import eu.dnetlib.actionmanager.common.Agent;

import eu.dnetlib.actionmanager.common.Operation;

import eu.dnetlib.actionmanager.common.Provenance;

import eu.dnetlib.actionmanager.hbase.HBaseActionManagerCore;

import eu.dnetlib.actionmanager.rmi.ActionManagerException;

import eu.dnetlib.actionmanager.rmi.ActionManagerService;

import eu.dnetlib.actionmanager.set.ActionManagerSet;

import eu.dnetlib.enabling.tools.AbstractBaseService;

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

public class ActionManagerServiceImpl extends AbstractBaseService implements ActionManagerService {

	private HBaseActionManagerCore core;

	private NotificationHandler notificationHandler;

	@Resource

	private HadoopServiceCore hadoopServiceCore;

	@Override

	public String createSet(@WebParam(name = "set") final ActionManagerSet set) throws ActionManagerException {

		try {

			final String basePath = core.getBasePathHDFS();

			final Path actionSetPath = new Path(basePath + "/" + set.getId());

			hadoopServiceCore.createHdfsDir(ClusterName.DM, actionSetPath.toString(), true);

			return core.getInformationServiceClient().registerSetProfile(set);

		} catch (HadoopServiceException e) {

			throw new ActionManagerException(e);

		} catch (ISLookUpException e) {

			throw new ActionManagerException(e);

		}

	}

	@Override

	public boolean deleteSet(@WebParam(name = "set") final String setId) throws ActionManagerException {

		try {

			final String basePath = core.getBasePathHDFS();

			final Path actionSetPath = new Path(basePath + "/" + setId);

			hadoopServiceCore.deleteFromHdfs(ClusterName.DM, actionSetPath.toString());

			return core.getInformationServiceClient().deleteSetProfile(setId);

		} catch (HadoopServiceException e) {

			throw new ActionManagerException(e);

		} catch (ISLookUpException e) {

			throw new ActionManagerException(e);

		}

	}

	@Override

	public List<ActionManagerSet> ListSets() throws ActionManagerException {

		return core.listAvailableSets();

	}

	@Override

	public String createAction(final String splitRuleId,

			final String set,

			final Agent agent,

			final Operation operation,

			final String infoPackage,

			final Provenance provenance,

			final String trust,

			final String nsprefix) throws ActionManagerException {

		final XsltInfoPackageAction action = core.applyInfoPackageAction(splitRuleId, set, agent, operation, infoPackage, provenance, trust, nsprefix);

		return action.getRowKey();

	}

	@Override

	public String updateAction(final String actionId,

			final String splitRuleId,

			final String set,

			final Agent agent,

			final Operation operation,

			final String infoPackage,

			final Provenance provenance,

			final String trust,

			final String nsprefix) throws ActionManagerException {

		core.deleteInfoPackageAction(actionId);

		final XsltInfoPackageAction action = core.applyInfoPackageAction(splitRuleId, set, agent, operation, infoPackage, provenance, trust, nsprefix);

		return action.getRowKey();

	}

	@Override

	public boolean deleteAction(final String actionId) throws ActionManagerException {

		core.deleteInfoPackageAction(actionId);

		return true;

	}

	@Override

	public List<String> getActionsByAgent(final String agentId, final int limit) throws ActionManagerException {

		return core.getHbaseClient().retrieveActionsByAgent(agentId, limit);

	}

	@Override

	public void notify(final String subscriptionId, final String topic, final String isId, final String message) {

		notificationHandler.notified(subscriptionId, topic, isId, message);

	}

	public HBaseActionManagerCore getCore() {

		return core;

	}

	public void setCore(final HBaseActionManagerCore core) {

		this.core = core;

	}

	public NotificationHandler getNotificationHandler() {

		return notificationHandler;

	}

	@Required

	public void setNotificationHandler(final NotificationHandler notificationHandler) {

		this.notificationHandler = notificationHandler;

	}

}