package eu.dnetlib.data.actionmanager.blackboard;

import java.util.Map;
import java.util.Set;

import com.google.common.base.Splitter;
import com.google.common.collect.Sets;
import eu.dnetlib.enabling.tools.blackboard.BlackboardJob;
import eu.dnetlib.enabling.tools.blackboard.BlackboardServerAction;
import eu.dnetlib.enabling.tools.blackboard.BlackboardServerHandler;
import org.apache.commons.lang3.StringUtils;

public class PromoteActionManagerActionFromHDFS extends AbstractActionManagerAction implements BlackboardServerAction<ActionManagerActions> {

	@Override
	public void execute(final BlackboardServerHandler handler, final BlackboardJob job) {

		final String setParam = job.getParameters().get("set");
		final String targetTable = job.getParameters().get("targetTable");

		Set<String> sets = null;
		if (StringUtils.isNotBlank(setParam) && !setParam.equalsIgnoreCase(ActionManagerJobLauncher.ALL_SETS)) {
			sets = Sets.newHashSet(Splitter.on(",").omitEmptyStrings().trimResults().split(setParam));
		}

		try {
			getHbaseJobLauncher().executePromoteFromHDFSJob(sets, targetTable, new ActionManagerBlackboardCallback() {

				@Override
				public void onDone(final Map<String, String> properties) {
					job.getParameters().putAll(properties);
					handler.done(job);
				}

				@Override
				public void onFailed(final Throwable e) {
					handler.failed(job, e);
				}

			});
		} catch (Throwable e) {
			handler.failed(job, e);
		}

	}
}