package eu.dnetlib.data.hadoop.hbase;

import java.util.List;

import eu.dnetlib.data.transform.Column;
import eu.dnetlib.data.transform.Row;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.hbase.client.Mutation;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;

/**
 * The Class HBasePutFeeder performs a batch of Put operations..
 */
public class HBasePutFeeder extends HbaseTableFeeder {

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

	/*
	 * (non-Javadoc)
	 *
	 * @see eu.dnetlib.data.hadoop.hbase.HbaseTableFeeder#addOperation(java.util.List, eu.dnetlib.data.transform.Row)
	 */
	@Override
	protected void addOperation(final List<Mutation> buffer, final Row row) {
		final Put put = new Put(Bytes.toBytes(row.getKey()));
		put.setWriteToWAL(true);

		for (final Column<String, byte[]> col : row) {
			log.debug(String.format("adding value to K: '%s' CF:'%s' Q:'%s'", row.getKey(), row.getColumnFamily(), col.getName()));
			put.add(Bytes.toBytes(row.getColumnFamily()), Bytes.toBytes(col.getName()), col.getValue());
		}

		buffer.add(put);
	}

}