package com.microsoft.azure.sre.datahub;

import java.text.BreakIterator;

import org.apache.storm.topology.BasicOutputCollector;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseBasicBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;
import org.apache.storm.Constants;

// For logging
import org.apache.logging.log4j.Logger;
import org.apache.logging.log4j.LogManager;

//There are a variety of bolt types. In this case, use BaseBasicBolt
public class TransformEchoBolt extends BaseBasicBolt {
  //Create logger for this class
  private static final Logger logger = LogManager.getLogger(TransformEchoBolt.class);
  private int counter = 0;

  protected static boolean isTickTuple(Tuple tuple) {
    return tuple.getSourceComponent().equals(Constants.SYSTEM_COMPONENT_ID)
         && tuple.getSourceStreamId().equals(Constants.SYSTEM_TICK_STREAM_ID);
  }

  //Execute is called to process tuples
  @Override
  public void execute(Tuple tuple, BasicOutputCollector collector) {
    if (isTickTuple(tuple)) 
    {
      //logger.info("Java Tick!");
      return;
    }

    counter++;
    if ((counter % 10000) == 0)
    {
      logger.info("Received record # " + counter);
      //Get the sentence content from the tuple
      String service = tuple.getString(0);
      String environment = tuple.getString(1);
      String plane = tuple.getString(2);
      String metric = tuple.getString(3);
      String timeEmittedMin = tuple.getString(4);
      String aggregationLevel = tuple.getString(5);
      String region = tuple.getString(6);
      String scaleUnit = tuple.getString(7);
      String scenario = tuple.getString(8);
      String operation = tuple.getString(9);
      Double value = tuple.getDouble(10);

      //logger.info("# " + counter + " Service " + service + " region " + region + " scenario " + scenario + " at time " + timeEmittedMin + " metric: " + metric + " value: " + value + " operation " + operation + " scaleunit " + scaleUnit);
    }
  }

  //Declare that emitted tuples contain a word field
  @Override
  public void declareOutputFields(OutputFieldsDeclarer declarer) {
  }
}