//package org.apache.hadoop.nobida;
//
//import java.io.IOException;
//import java.nio.ByteBuffer;
//import java.nio.charset.Charset;
//import java.text.ParseException;
//import java.text.SimpleDateFormat;
//import java.util.Arrays;
//import java.util.LinkedList;
//import java.util.List;
//import java.util.Random;
//import java.util.concurrent.CountDownLatch;
//import java.util.concurrent.ExecutorService;
//import java.util.concurrent.Executors;
//import java.util.concurrent.locks.Condition;
//import java.util.concurrent.locks.ReentrantLock;
//
//import org.apache.hadoop.conf.Configuration;
//import org.apache.hadoop.fs.Path;
//import org.apache.hadoop.hbase.HColumnDescriptor;
//import org.apache.hadoop.hbase.HTableDescriptor;
//import org.apache.hadoop.hbase.client.Durability;
//import org.apache.hadoop.hbase.client.HBaseAdmin;
//import org.apache.hadoop.hbase.client.HTable;
//import org.apache.hadoop.hbase.client.HTableInterface;
//import org.apache.hadoop.hbase.client.HTablePool;
//import org.apache.hadoop.hbase.client.Put;
//import org.apache.hadoop.hbase.util.Bytes;
//import org.apache.zookeeper.CreateMode;
//import org.slf4j.Logger;
//import org.slf4j.LoggerFactory;
//
//public class NobidaBuffer {
//
//  final byte[] a = { (byte) 'a' };
//  final byte[] s = { 's' };
//  final byte[] sp = { (byte) 's', (byte) 'p' };
//  final byte[] d = { 'd' };
//  final byte[] dp = { (byte) 'd', (byte) 'p' };
//  final byte[] u = { 'u' };
//  final byte[] z = { 'z' };
//  final byte[] metaFamily = { 'b' };
//  final byte[] dataFamily = { 'd' };
//
//  final static Logger logger = LoggerFactory.getLogger(NobidaBuffer.class);
//  final static BufferNodeConfig cfg = ConfigFactory.create(
//      BufferNodeConfig.class, System.getProperties());
//  final private HTablePool tablePool;
//  private byte[] listName = "hehe".getBytes();
//  private String tableName = cfg.tableName();
//  private long writeBufferSize = cfg.writeBufferSize();
//  private int workerNum = cfg.workerNum();
//  private HBasePutter[] workers = new HBasePutter[workerNum];
//  private int batchSize = cfg.batchSize();
//  private int bufferSize = cfg.bufferSize();
//  private int sessionTimeout = cfg.sessionTimeout();
//  private String hbaseConfigFile = cfg.hbaseConfigFile();
//  private String zookeeperAddress = cfg.zookeeperAddress();
//  private String rootDir = "redis";
//  private Charset charset = Charset.forName("UTF-8");
//  private Jedis jedis;
//  private String redisAddress = cfg.redisAddress();
//  private int id = cfg.id();
//  private String redisCommand = cfg.redisCommand(id + "");
//  private int regionNum = cfg.regionNum();
//  private String[] columnFamilies = cfg.columnFamilies();
//  private int redisNum = cfg.redisNum();
//  private ExecutorService exec = Executors.newCachedThreadPool();
//  private CountDownLatch cdl = new CountDownLatch(workerNum);
//  private CuratorFramework client;
//  private NodeCache nodeCache;
//  private CountDownLatch shutdownSignal = new CountDownLatch(1);
//  private volatile boolean isSlave;
//  private boolean debug = cfg.debug();
//
//  public NobidaBuffer() {
//    String[] tmp = redisAddress.split(":");
//    this.jedis = new Jedis(tmp[0], Integer.parseInt(tmp[1]));
//
//    Configuration conf = new Configuration();
//    conf.addResource(new Path(hbaseConfigFile));
//    tablePool = new HTablePool(conf, Integer.MAX_VALUE);
//    HBaseAdmin admin = null;
//    try {
//      admin = new HBaseAdmin(conf);
//      HTableDescriptor des = new HTableDescriptor(tableName);
//      if (!admin.tableExists(tableName)) {
//        byte[][] splits = null;
//        if (regionNum > 1) {
//          splits = regionSplitter(regionNum);
//        }
//        for (int i = 0; i < columnFamilies.length; i++) {
//          HColumnDescriptor hcd = new HColumnDescriptor(columnFamilies[i]);
//          des.addFamily(hcd);
//        }
//        if (splits != null)
//          admin.createTable(des, splits);
//        else
//          admin.createTable(des);
//      }
//      // zqx
//      createIndexTable();
//    } catch (IOException e) {
//      logger.error("Caught an error: ", e);
//    } finally {
//      if (admin != null)
//        try {
//          admin.close();
//        } catch (IOException e) {
//          // TODO Auto-generated catch block
//          logger.error("Caught an error: ", e);
//        }
//    }
//  }
//
//  // zqx
//  public void createIndexTable() throws IOException {
//
//    HBaseAdmin admin = null;
//    Configuration conf = new Configuration();
//    conf.addResource(new Path(hbaseConfigFile));
//    admin = new HBaseAdmin(conf);
//
//    HTableDescriptor desIndex = new HTableDescriptor(tableName + "_index");
//    HColumnDescriptor hcdIndex = new HColumnDescriptor(Bytes.toBytes("i"));
//    if (!admin.tableExists(tableName + "_index")) {
//      byte[][] splitsIndex = null;
//      splitsIndex = regionSplitter(4);
//
//      desIndex.addFamily(hcdIndex);
//
//      if (splitsIndex != null)
//        admin.createTable(desIndex, splitsIndex);
//      else
//        admin.createTable(desIndex);
//
//      try {
//        admin.close();
//      } catch (IOException e) {
//        // TODO Auto-generated catch block
//        e.printStackTrace();
//      }
//    }
//  }
//  class HBasePutter implements Runnable {
//    // zqx
//    private HTable indexTable = null;
//    private HTableInterface hTable = tablePool.getTable(tableName);
//    private int workerId;
//    private byte[] rowkey = new byte[9];// [0:regionHeader,1..8:time stamp]
//    private int hdrlen;// ip packet head length
//    // zqx
//    private Random random = new Random();
//    /*
//     * reuse these buffer to construct byte arrays needed by hbase client
//     * library
//     */
//    private ByteBuffer ipsrc = ByteBuffer.allocate(bufferSize);
//    private ByteBuffer ipdes = ByteBuffer.allocate(bufferSize);
//    private ByteBuffer portsrc = ByteBuffer.allocate(bufferSize);
//    private ByteBuffer portdes = ByteBuffer.allocate(bufferSize);
//    private ByteBuffer type = ByteBuffer.allocate(bufferSize);
//    private long sec;
//    private long usec;
//    private ByteBuffer ts = ByteBuffer.allocate(8);
//    private SimpleDateFormat sdf = new SimpleDateFormat();
//    private StringBuilder date = new StringBuilder();
//    private List<byte[]> packets;
//    private ReentrantLock lock = new ReentrantLock();
//    private Condition condition = lock.newCondition();
//    private int bits;
//    private int hour;
//    private int day;
//    private int minute;
//    private int second;
//    private int milisecond;
//    private int nanosecond;
//    private int microsecond;
//
//    public HBasePutter(int workerId) {
//      // zqx
//      Configuration conf = new Configuration();
//      conf.addResource(new Path(hbaseConfigFile));
//      indexTable = new HTable(conf, tableName + "_index");
//
//      this.workerId = workerId;
//      sdf.applyPattern("yyyy-MM-dd-HH:mm:ss");
//      hTable.setAutoFlush(false);
//      // zqx
//      indexTable.setAutoFlush(false);
//      try {
//        hTable.setWriteBufferSize(writeBufferSize);
//        // zqx
//        indexTable.setWriteBufferSize(writeBufferSize);
//      } catch (IOException e) {
//        logger.error("Caught an error: ", e);
//      }
//    }
//
//    public void setPackets(List<byte[]> packets) {
//      this.packets = packets;
//    }
//
//    public boolean tryLock() {
//      return lock.tryLock();
//    }
//
//    public void unlock() {
//      lock.unlock();
//    }
//
//    public void signal() {
//      condition.signal();
//    }
//
//    @Override
//    public void run() {
//
//      lock.lock();
//      /*
//       * ensure that every worker is started before the main thread trying to
//       * signal them
//       */
//      cdl.countDown();
//      while (true) {
//        try {
//          condition.await();
//        } catch (InterruptedException e) {
//          logger.error("", e);
//          if (Thread.interrupted()) {
//            lock.unlock();
//            break;
//          }
//        }
//        try {
//
//          for (byte[] packet : packets) {
//            /*
//             * may get a null packet from redis
//             */
//            if (packet == null || packet.length == 0) {
//              continue;
//            }
//            rowkey[0] = (byte) (workerId + workerNum * id + redisNum
//                * workerNum * packet[4]);
//
//            calcTs(packet, 0);
//
//            // sec = (packet[0] & 0xff) + (packet[1] & 0xff) * 256
//            // + (packet[2] & 0xff) * 65536
//            // + (packet[3] & 0xff) * 16777216;
//            // usec = (packet[8] & 0xff) + (packet[9] & 0xff) * 256
//            // + (packet[10] & 0xff) * 65536
//            // + (packet[11] & 0xff) * 16777216;
//            ts.rewind();
//            ts.putLong(sec * 1000000000 + usec * 1000 + nanosecond);
//            System.arraycopy(ts.array(), 0, rowkey, 1, 8);
//            type.rewind();
//            ipsrc.rewind();
//            ipdes.rewind();
//            portsrc.rewind();
//            portdes.rewind();
//            int len;
//            /*
//             * dissecting the big packet
//             */
//            for (int start = 0; start < packet.length; start += len) {
//              len = calcLen(packet, start);
//              calcTs(packet, start);
//              packet[start + 4] = 0;
//              hdrlen = (packet[start + 38] & 0x0f) * 4;
//
//              if (packet[start + 47] == 6 || packet[start + 47] == 17) {
//                type.put(packet[start + 47]);
//              } else {
//                type.put((byte) '*');
//              }
//              type.put((byte) '*');
//              type.put((byte) '*');
//              type.put((byte) '*');
//              ipsrc.put(packet[start + 50]);
//              ipsrc.put(packet[start + 51]);
//              ipsrc.put(packet[start + 52]);
//              ipsrc.put(packet[start + 53]);
//              ipdes.put(packet[start + 54]);
//              ipdes.put(packet[start + 55]);
//              ipdes.put(packet[start + 56]);
//              ipdes.put(packet[start + 57]);
//
//              portsrc.put(packet[start + 38 + hdrlen]);
//              portsrc.put(packet[start + 39 + hdrlen]);
//              portsrc.put((byte) '*');
//              portsrc.put((byte) '*');
//
//              portdes.put(packet[start + 40 + hdrlen]);
//              portdes.put(packet[start + 41 + hdrlen]);
//              portdes.put((byte) '*');
//              portdes.put((byte) '*');
//              // zqx
//              Put indexPut = indexPut(rowkey, ipsrc.array(), ipdes.array());
//              indexTable.put(indexPut);
//            }
//            Put put = new Put(rowkey);
//            put.setDurability(Durability.SKIP_WAL);
//            put.add(metaFamily, a, Long.MAX_VALUE, type.array(), 0,
//                type.position());
//            put.add(metaFamily, d, Long.MAX_VALUE, ipdes.array(), 0,
//                ipdes.position());
//            put.add(metaFamily, dp, Long.MAX_VALUE, portdes.array(), 0,
//                portdes.position());
//            put.add(metaFamily, s, Long.MAX_VALUE, ipsrc.array(), 0,
//                ipsrc.position());
//            put.add(metaFamily, sp, Long.MAX_VALUE, portsrc.array(), 0,
//                portsrc.position());
//
//            // need to truncate b[4-7] and b[12-15] to be standard
//            // pcap format
//            put.add(dataFamily, z, Long.MAX_VALUE, packet);
//
//            if (!debug) {
//              try {
//                hTable.put(put);
//
//              } catch (Exception e) {
//                logger.error("Caught an error: ", e);
//              }
//            }
//          }
//
//        } catch (Exception e1) {
//          logger.error("Caught an error: ", e1);
//        } finally {
//          // workers never release their lock except using
//          // conditioners
//          // lock.unlock();
//        }
//      }
//
//    }
//
//    // zqx
//    public Put indexPut(byte[] rowKey, byte[] ipsrc, byte[] ipdes) {
//      byte[] time = Arrays.copyOfRange(rowKey, 1, rowKey.length);
//      byte[] indexKey = Bytes.add(ipsrc, time, ipsrc);
//      byte[] preKey = new byte[] { (byte) random.nextInt(4) };
//      indexKey = Bytes.add(preKey, indexKey);
//      Put put = new Put(indexKey);
//      put.add(Bytes.toBytes("i"), Bytes.toBytes("i"), rowKey);
//      return put;
//    }
//    /*
//     * calculate the packet size started at position 'start'.
//     */
//    private int calcLen(byte[] b, int start) {
//      int len = (b[start + 16] & 0xff) + (b[start + 17] & 0xff) * 256
//          + (b[start + 18] & 0xff) * 65536 + 24;
//      return len;
//    }
//
//    private void calcTs(byte[] packet, int start) throws ParseException {
//      date = new StringBuilder();
//      date.append((packet[start + 27] << 8) + (packet[start + 28] & 0xff));
//      date.append('-');
//      date.append(packet[start + 29] >>> 4 & 0xf);
//      date.append('-');
//      bits = (packet[start + 29] << 6) + (packet[start + 30] >>> 2 & 0x3f);
//      hour = bits & 0x1f;
//      bits >>>= 5;
//      day = bits & 0x1f;
//      date.append(day);
//      date.append('-');
//      date.append(hour);
//      bits = (packet[start + 30] << 4) + (packet[start + 31] >>> 4 & 0xf);
//      minute = bits & 0x3f;
//      date.append(':');
//      date.append(minute);
//      bits = (packet[start + 31] << 2) + (packet[start + 32] >>> 6 & 0x3);
//      second = bits & 0x3f;
//      date.append(':');
//      date.append(second);
//      bits = (packet[start + 32] << 4) + (packet[start + 33] >>> 4 & 0xf);
//      milisecond = bits & 0x3ff;
//      bits = (packet[start + 33] << 6) + (packet[start + 34] >>> 2 & 0x3f);
//      microsecond = bits & 0x3ff;
//      bits = (packet[start + 34] << 8) + (packet[start + 35] & 0xff);
//      nanosecond = bits & 0x3ff;
//      sec = sdf.parse(date.toString()).getTime() / 1000;
//      ts.rewind();
//      ts.putLong(sec);
//      packet[start + 0] = ts.get(7);
//      packet[start + 1] = ts.get(6);
//      packet[start + 2] = ts.get(5);
//      packet[start + 3] = ts.get(4);
//      usec = milisecond * 1000 + microsecond;
//      ts.rewind();
//      ts.putLong(usec);
//      packet[start + 8] = ts.get(7);
//      packet[start + 9] = ts.get(6);
//      packet[start + 10] = ts.get(5);
//      packet[start + 11] = ts.get(4);
//    }
//  }
//
//  private class RedisReceiver implements Runnable {
//
//    private List<byte[]> packets;
//
//    public void run() {
//
//      /*
//       * wait until all workers have started
//       */
//      try {
//        cdl.await();
//      } catch (InterruptedException e) {
//        logger.error("", e);
//        if (Thread.interrupted()) {
//          return;
//        }
//      }
//      int idx = 0;// index of workers
//      while (true) {
//        /*
//         * regard all exceptions as redis exceptions
//         */
//        try {
//          packets = new LinkedList<byte[]>();
//          for (int i = 0; i < batchSize; i++) {
//            try {
//              byte[] packet = jedis.rpop(listName);
//              if (packet != null && packet.length != 0) {
//                packets.add(packet);
//              }
//            } catch (JedisDataException e) {
//              /*
//               * what exception can we tell that redis becomes a slave
//               */
//              logger.error("", e);
//              if (isSlave)
//                return;
//            }
//          }
//        } catch (Exception e) {
//          logger.error("", e);
//          /*
//           * check if redis is ok
//           */
//          if (reconnect()) {
//            continue;
//          } else {
//            logger.error("failed reconnecting to redis " + id);
//            shutdownSignal.countDown();
//            break;
//          }
//        }
//
//        idx = (idx + 1) % workerNum;// move to the next worker
//
//        /*
//         * repeat choosing a waiting worker until we get one
//         */
//        for (;; idx = (idx + 1) % workerNum) {
//          if (workers[idx].tryLock()) {
//            workers[idx].setPackets(packets);
//            workers[idx].signal();
//            workers[idx].unlock();
//            break;
//          }
//        }
//
//      }
//    }
//  }
//
//  /*
//   * to LinShuai when ntpdate started, reconnect is triggered
//   */
//  private boolean reconnect() {
//    while (true) {
//      try {
//        jedis.disconnect();
//        jedis.connect();
//        logger.info("Reconnect a redis");
//        return true;
//      } catch (Exception e) {
//        try {
//          Thread.sleep(1000l);
//        } catch (InterruptedException e1) {
//        }
//      }
//    }
//  }
//
//  public void start() {
//
//    try {
//      /*
//       * if this program collapes, shutdown the relevant redis
//       */
//      JVMShutdownHook jvmShutdownHook = new JVMShutdownHook();
//      Runtime.getRuntime().addShutdownHook(jvmShutdownHook);
//      String[] arg = redisCommand.split("\\s");
//      Runtime.getRuntime().exec(arg);
//    } catch (Exception e) {
//      logger.error("Caught an error: ", e);
//      System.exit(1);
//    }
//
//    for (int i = 0; i < workerNum; i++) {
//      workers[i] = new HBasePutter(i);
//      exec.execute(workers[i]);
//    }
//
//    client = CuratorFrameworkFactory.builder().connectString(zookeeperAddress)
//        .namespace(rootDir).sessionTimeoutMs(sessionTimeout)
//        .retryPolicy(new RetryNTimes(Integer.MAX_VALUE, 1000))
//        .connectionTimeoutMs(5000).build();
//    client.start();
//    nodeCache = new NodeCache(client, "/" + id);
//    nodeCache.getListenable().addListener(new NodeListener());
//    try {
//      nodeCache.start();
//    } catch (Exception e) {
//      logger.error("", e);
//    }
//    promoteToMaster();
//
//    while (true) {
//      try {
//        shutdownSignal.await();
//        exec.shutdownNow();
//        break;
//      } catch (Exception e) {
//        logger.error("", e);
//      }
//    }
//  }
//
//  /*
//   * pre-split regions
//   */
//  private byte[][] regionSplitter(int num) {
//    byte[][] ret = new byte[num - 1][];
//    int start = 0;
//    byte[] rowkey = new byte[9];
//    Arrays.fill(rowkey, 1, 8, (byte) 0);
//    for (int i = 1; i < num; i++) {
//      rowkey[0] = (byte) (start + i);
//      ret[i - 1] = Arrays.copyOf(rowkey, rowkey.length);
//    }
//    return ret;
//  }
//
//  public static void main(String args[]) {
//
//    new NobidaBuffer().start();
//
//  }
//
//  private void promoteToMaster() {
//
//    try {
//      client.create().withMode(CreateMode.EPHEMERAL)
//          .forPath("/" + id, redisAddress.getBytes());
//      jedis.slaveofNoOne();
//      exec.execute(new RedisReceiver());
//      isSlave = false;
//      logger.info("promoteToMaster");
//    } catch (Exception e) {
//    }
//  }
//
//  private void slaveOfMaster(String masterAddress) {
//    String[] tmp = masterAddress.split(":");
//    String ip = tmp[0];
//    int port = Integer.parseInt(tmp[1]);
//    jedis.slaveof(ip, port);
//    isSlave = true;
//    logger.info("Become a slave of redis " + masterAddress);
//  }
//
//  private class NodeListener implements NodeCacheListener {
//
//    @Override
//    public void nodeChanged() {
//      ChildData data = nodeCache.getCurrentData();
//      if (data == null) {
//        if (isSlave) {
//          promoteToMaster();
//        }
//      } else {
//        String masterAddress = new String(data.getData(), charset);
//        if (masterAddress.equals(redisAddress)) {
//          return;
//        } else {
//          slaveOfMaster(masterAddress);
//        }
//      }
//
//    }
//
//  }
//
//  private class JVMShutdownHook extends Thread {
//    public void run() {
//      jedis.connect();
//      jedis.shutdown();
//    }
//  }
// }
