package eu.dnetlib.data.hadoop.utils;

import java.io.IOException;
import java.io.StringReader;
import java.util.Map;

import com.google.common.collect.Maps;
import org.apache.commons.io.IOUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;
import org.dom4j.Document;
import org.dom4j.DocumentException;
import org.dom4j.io.SAXReader;
import org.junit.Before;
import org.junit.Ignore;
import org.junit.Test;
import org.springframework.core.io.ClassPathResource;
import org.springframework.core.io.Resource;

/**
 * Created by claudio on 01/02/2017.
 */
@Ignore
public class ScanTest {

	private static final String TABLE_NAME = "db_dli";
	private Resource confIn = new ClassPathResource("eu/dnetlib/data/hadoop/config/hadoop-default.dm.cnr.properties");

	private HTable table;

	@Before
	public void setUp() throws IOException {

		final Configuration conf = new Configuration();

		for(final String line : IOUtils.readLines(confIn.getInputStream())) {
			System.out.println("line = " + line);
			if (!line.trim().isEmpty() && !line.startsWith("#")) {
				final String[] split = line.split("=");
				conf.set(split[0].trim(), split[1].trim());
			}
		}

		table = new HTable(conf, Bytes.toBytes(TABLE_NAME));
	}

	@Test
	public void testScan() throws IOException, DocumentException {
		final String scanString = "<SCAN>\n"
				+ "<FILTERS operator=\"MUST_PASS_ALL\">\n"
				+ "<FILTER type=\"prefix\" param=\"entityTypeId\"/>\n"
				+ "</FILTERS>\n"
				+ "<FAMILIES>\n"
				+ "<FAMILY value=\"rels\" qualifier=\"person_person_isSimilarTo\" />\n"
				+ "<FAMILY value=\"rels\" qualifier=\"dataset_dataset_isSimilarTo\" />\n"
				+ "<FAMILY value=\"rels\" qualifier=\"publication_publication_isSimilarTo\" />\n"
				+ "<FAMILY value=\"rels\" qualifier=\"organization_organization_isSimilarTo\" />\n"
				+ "</FAMILIES>\n"
				+ "</SCAN>";

		final Map<String, String> map = Maps.newHashMap();
		map.put("entityTypeId", "50");
		final SAXReader reader = new SAXReader();
		final Document doc = reader.read(new StringReader(scanString));

		final ScanFactory scanFactory = new ScanFactory();
		final ScanProperties scanProperties = scanFactory.parseScanProperties(doc, map);
		final Scan scan = scanFactory.getScanObject(scanProperties);

		final ResultScanner rs = table.getScanner(scan);
		rs.forEach(r -> {
			r.list().forEach(kv -> {
				System.out.println(String.format("cf: %s, q: %s", new String(kv.getFamily()), new String(kv.getQualifier())));
			});
		});
		rs.close();
	}

}