package eu.dnetlib.data.hadoop.utils;

import eu.dnetlib.data.proto.dli.DLIObjectProtos;
import org.apache.commons.io.IOUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.Text;
import org.junit.Before;
import org.junit.Ignore;
import org.junit.Test;
import org.springframework.core.io.ClassPathResource;
import org.springframework.core.io.Resource;

import java.io.IOException;

/**
 * Created by sandro on 2/28/17.
 */
@Ignore
public class ReadSeqFileTest {
    private Resource confIn = new ClassPathResource("eu/dnetlib/data/hadoop/config/hadoop-default.dm.cnr.properties");
    private FileSystem fs;
    private Configuration conf;

    @Before
    public void setUp() throws IOException {

        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());
            }
        }


        fs = FileSystem.get(conf);
    }


    @Test
    public void testRead() throws Exception {

        Path path = new Path("hdfs://nmis-hadoop-cluster/user/dnet/dli/index_data.seq/part-r-00000");
        SequenceFile.Reader reader = new SequenceFile.Reader(fs, path, conf);
        Text key = new Text();
        ImmutableBytesWritable value = new ImmutableBytesWritable();
        boolean next = reader.next(key, value);
        int inputSize = 0;
        int outputSize = 0;
        while (next) {
            DLIObjectProtos.DLIObject dliObject = DLIObjectProtos.DLIObject.parseFrom(value.copyBytes());
            System.out.println("dliObject = " + dliObject);
            break;
        }
        System.out.println("inputSize = " + inputSize);
        System.out.println("outputSize = " + outputSize);
    }

}