import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.Date;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;

public class Populator {

    public static void main(String[] args) throws IOException {
        System.out.println("HBase Start");

        // The configuration object tells the client where to connect.
        // It reads from hbase-site.xml and hbase-default.xml (from hbase-0.94.3/conf)
        Configuration config = HBaseConfiguration.create();
        //config.set("hbase.zookeeper.quorum", "localhost");  // Here we are running zookeeper locally

        //gotta create the table before we do anything
        HBaseAdmin admin = new HBaseAdmin(config);
        HTableDescriptor messageTable = new HTableDescriptor(Bytes.toBytes("messages"));

        //defining the column families of the table
        HColumnDescriptor idsFamily = new HColumnDescriptor(Bytes.toBytes("ids"));
        messageTable.addFamily(idsFamily);
        HColumnDescriptor timeFamily = new HColumnDescriptor(Bytes.toBytes("timestamp"));
        messageTable.addFamily(timeFamily);
        HColumnDescriptor msgFamily = new HColumnDescriptor(Bytes.toBytes("text"));
        messageTable.addFamily(msgFamily);

        admin.createTable(messageTable);

        // We initialize an HTable object to connect to our messages table.
        HTable table = new HTable(config, "messages");

        //----Now the table is created, everything under here involves populating the table----

        //create 500 users and 10 different emails
        String[] users = new String[500];
        String[] emails = {"@gmail.com",
                "@hotmail.com",
                "@aol.com",
                "@yahoo.com",
                "@msn.com",
                "@protonmail.com",
                "@live.com",
                "@asu.edu",
                "@mail.com",
                "@inbox.com",};

        //every 10 users will have the same email domain (just to keep track & to throw in a little variety)
        for (int i=0;i<500;i++)
        {
            users[i] = "human" + i + emails[i%10];
        }

        //the various date formats we will use
        SimpleDateFormat sdfKey = new SimpleDateFormat(":yyyyMMddHHmmssSSS");
        SimpleDateFormat sdfDate = new SimpleDateFormat("yyyy-MM-dd");
        SimpleDateFormat sdfTime = new SimpleDateFormat("HH:mm:ss.SSS");
        Date resultdate = new Date(System.currentTimeMillis());


        //for every user, that user will send out 10 emails to the next 10 users on the array (so user 10 will send to users 11-20)
        //%500 is used here so it wraps back - human495 will send to 496,497,498,499,1,2,3,4,5
        for (int i=0;i<500;i++)
        {
            for (int j=0;j<10;j++)
            {
                //redefine resultdate
                //set row key (user[i]+timestamp)
                //do all the put stuff :)
                resultdate = new Date(System.currentTimeMillis()); //reassign resultdate
                System.out.println(users[i] + " Sent to " + users[(i+j+1)%500] + sdfKey.format(resultdate) + " Email #" + j);
            }

        }



        //Defining our row keys: the ID number and the current system time in milliseconds to make it unique
        //when scanning we can just scan for id+*
        resultdate = new Date(System.currentTimeMillis());
        byte[] rowKey1 = Bytes.add(Bytes.toBytes("1234"), Bytes.toBytes(sdfKey.format(resultdate).toString()));
        byte[] rowKey2 = Bytes.add(Bytes.toBytes("7890"), Bytes.toBytes(sdfKey.format(resultdate).toString()));

        Put p = new Put(Bytes.toBytes(Bytes.toString(rowKey1)));       //value here is the row key(title)
        p.add(Bytes.toBytes("ids"), Bytes.toBytes("toID"), Bytes.toBytes("1234"));
        p.add(Bytes.toBytes("ids"), Bytes.toBytes("fromID"), Bytes.toBytes("7890"));
        resultdate = new Date(System.currentTimeMillis());
        String currentDate = sdfDate.format(resultdate).toString();
        String currentTime = sdfTime.format(resultdate).toString();
        p.add(Bytes.toBytes("timestamp"), Bytes.toBytes("date"), Bytes.toBytes(currentDate));
        p.add(Bytes.toBytes("timestamp"), Bytes.toBytes("time"), Bytes.toBytes(currentTime));
        p.add(Bytes.toBytes("text"), Bytes.toBytes("title"), Bytes.toBytes("Pressing Information"));
        p.add(Bytes.toBytes("text"), Bytes.toBytes("body"), Bytes.toBytes("This is regarding your dog. He's a good boy."));
        table.put(p);

        // Adding row post3
        p = new Put(Bytes.toBytes(Bytes.toString(rowKey2)));
        p.add(Bytes.toBytes("ids"), Bytes.toBytes("toID"), Bytes.toBytes("7890"));
        p.add(Bytes.toBytes("ids"), Bytes.toBytes("fromID"), Bytes.toBytes("1234"));
        p.add(Bytes.toBytes("timestamp"), Bytes.toBytes("date"), Bytes.toBytes("Feb 3"));
        p.add(Bytes.toBytes("timestamp"), Bytes.toBytes("time"), Bytes.toBytes("14:54"));
        p.add(Bytes.toBytes("text"), Bytes.toBytes("title"), Bytes.toBytes("Inquiry"));
        p.add(Bytes.toBytes("text"), Bytes.toBytes("body"), Bytes.toBytes("r u dum lol"));
        table.put(p);

        // We can retrieve the data of a row using Get. Get returns an array of byte
        Get g = new Get(Bytes.toBytes("1234"));
        Result r = table.get(g);
        byte [] value1 = r.getValue(Bytes.toBytes("ids"), Bytes.toBytes("fromID"));
        byte [] value2 = r.getValue(Bytes.toBytes("text"), Bytes.toBytes("title"));

        // We convert to Strings
        String valueStr1 = Bytes.toString(value1);
        String valueStr2 = Bytes.toString(value2);
        System.out.println("GET: " + valueStr1 + " " + valueStr2);

        // We can retrieve the data of multiple rows using Scan
        //Retrieving only column title of all the rows
        Scan s = new Scan();
        s.addColumn(Bytes.toBytes("text"), Bytes.toBytes("title"));
        ResultScanner scanner = table.getScanner(s);
        try {
            for (Result rr : scanner)
              System.out.println("Row data: " + rr);
        }
        finally {
          scanner.close();
        }

        //Retrieving all columns of all the rows
        /*
        s = new Scan();
        s.addFamily(Bytes.toBytes("post"));
        scanner = table.getScanner(s);
        try {
            for (Result rr : scanner)
              System.out.println("Row data: " + rr);
        }
        finally {
          scanner.close();
        }
        */

        table.close();//Releases held resources
        System.out.println("HBase End");
    }
}