package eu.dnetlib.actionmanager.actions;

import java.util.List;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;

import com.google.common.collect.Lists;

import eu.dnetlib.actionmanager.ActionManagerConstants;
import eu.dnetlib.actionmanager.common.Agent;
import eu.dnetlib.actionmanager.common.Operation;
import eu.dnetlib.actionmanager.common.Provenance;
import eu.dnetlib.actionmanager.hbase.HBasePutFactory;

abstract public class AbstractMetaAction<T> extends AbstractAction {

	private Operation operation;

	private T metaContent;

	private ActionFactory actionFactory;

	private static final Log log = LogFactory.getLog(AbstractMetaAction.class); // NOPMD by marko on 11/24/08 5:02 PM

	public AbstractMetaAction(String set, Agent agent, Operation operation, T metaContent, ActionFactory actionFactory) {
		super(set, agent);
		this.setOperation(operation);
		this.metaContent = metaContent;
		this.setActionFactory(actionFactory);
	}

	@Override
	public List<Put> asPutOperations(String parentId, Provenance provenance, String trust, String nsprefix) {
		List<Put> puts = Lists.newArrayList();
		List<String> partIds = Lists.newArrayList();
		try {
			for (AtomicAction action : calculateAtomicActions(provenance, trust, nsprefix)) {
				if (action != null && action.isValid()) {
					puts.addAll(action.asPutOperations(getRowKey(), provenance, trust, nsprefix));
					partIds.add(action.getRowKey());
				}
			}
			puts.add(prepareMetaPut(partIds));
		} catch (Exception e) {
			log.error("Error generating actions", e);
		}

		return puts;
	}

	private Put prepareMetaPut(List<String> partIds) {
		final Put put = HBasePutFactory.createPutOperation(getRowKey(), getSet(), getAgent());

		put.add(ActionManagerConstants.OPERATION_COLFAMILY, Bytes.toBytes(getOperation().toString()), Bytes.toBytes(getOperation().toString()));
		put.add(ActionManagerConstants.ACTION_COLFAMILY, Bytes.toBytes(getTypology().toString()), getBytesContent(metaContent));

		for (String id : partIds) {
			put.add(ActionManagerConstants.RELATION_COLFAMILY, Bytes.toBytes(id), ActionManagerConstants.HASPARTS);
		}
		return put;
	}

	abstract protected List<AtomicAction> calculateAtomicActions(Provenance provenance, String trust, String nsprefix);

	abstract protected byte[] getBytesContent(T content);

	public Operation getOperation() {
		return operation;
	}

	public void setOperation(Operation operation) {
		this.operation = operation;
	}

	public T getMetaContent() {
		return metaContent;
	}

	public void setMetaContent(T metaContent) {
		this.metaContent = metaContent;
	}

	public ActionFactory getActionFactory() {
		return actionFactory;
	}

	public void setActionFactory(ActionFactory actionFactory) {
		this.actionFactory = actionFactory;
	}

}