package eu.dnetlib.data.actionmanager;

import java.util.List;
import javax.jws.WebParam;

import eu.dnetlib.data.hadoop.HadoopIsClient;
import eu.dnetlib.data.hadoop.HadoopServiceCore;

import eu.dnetlib.rmi.data.hadoop.ClusterName;
import eu.dnetlib.rmi.data.hadoop.HadoopServiceException;
import eu.dnetlib.enabling.tools.AbstractBaseService;
import eu.dnetlib.enabling.tools.blackboard.NotificationHandler;
import eu.dnetlib.rmi.data.hadoop.actionmanager.ActionManagerException;
import eu.dnetlib.rmi.data.hadoop.actionmanager.ActionManagerService;
import eu.dnetlib.rmi.data.hadoop.actionmanager.ActionManagerSet;
import org.apache.hadoop.fs.Path;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Required;

public class ActionManagerServiceImpl extends AbstractBaseService implements ActionManagerService {

	@Autowired
	private HadoopIsClient isClient;

	private NotificationHandler notificationHandler;

	@Autowired
	private HadoopServiceCore hadoopServiceCore;

	@Override
	public String createSet(@WebParam(name = "set") final ActionManagerSet set) throws ActionManagerException {
		try {
			final String basePath = isClient.getBasePathHDFS();
			final Path actionSetPath = new Path(basePath + "/" + set.getId());
			hadoopServiceCore.createHdfsDir(ClusterName.DM, actionSetPath.toString(), true);
			return isClient.registerSetProfile(set);
		} catch (HadoopServiceException e) {
			throw new ActionManagerException(e);
		}
	}

	@Override
	public boolean deleteSet(@WebParam(name = "set") final String setId) throws ActionManagerException {
		try {
			final String basePath = isClient.getBasePathHDFS();
			final Path actionSetPath = new Path(basePath + "/" + setId);
			hadoopServiceCore.deleteFromHdfs(ClusterName.DM, actionSetPath.toString());
			return isClient.deleteSetProfile(setId);
		} catch (HadoopServiceException e) {
			throw new ActionManagerException(e);
		}
	}

	@Override
	public List<ActionManagerSet> ListSets() throws ActionManagerException {
		return isClient.listSets();
	}

	@Override
	public void notify(final String subscriptionId, final String topic, final String isId, final String message) {
		getNotificationHandler().notified(subscriptionId, topic, isId, message);
	}

	public NotificationHandler getNotificationHandler() {
		return notificationHandler;
	}

	@Required
	public void setNotificationHandler(final NotificationHandler notificationHandler) {
		this.notificationHandler = notificationHandler;
	}

}