//package org.apache.storm.starter;
import java.io.FileWriter;   // Import the FileWriter class
import java.util.HashMap;
import java.util.Map;
import java.util.Date;
import java.text.SimpleDateFormat;
import java.io.File;  // Import the File class
import java.io.BufferedWriter;
import java.io.FileWriter;
import java.io.IOException;  // Import the IOException class to handle errors
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.IRichBolt;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Tuple;

public class WordCountBolt implements IRichBolt {

 Map < String, Integer > counters;
 Integer id;
 String name;
 String fileName;

 @Override
 public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {

  this.counters = new HashMap < > ();
  this.name = context.getThisComponentId();
  this.id = context.getThisTaskId();
 }

 @Override
 public void execute(Tuple input) {


  String word = input.getStringByField("word");
  if (!counters.containsKey(word)) {
   counters.put(word, 1);
  } else {
   counters.put(word, counters.get(word) + 1);
  }
 }

 @Override
 public void cleanup() {
    try {	
      BufferedWriter logOutput;
      String timeStamp = new SimpleDateFormat("yyyy.MM.dd.HH.mm.ss").format(new Date());
      String filename = "/tmp/storm-log-"+timeStamp+".log";
      logOutput = new BufferedWriter(new FileWriter(filename, true));
      logOutput.newLine();
      logOutput.newLine();
      logOutput.append("Final word count:::::");
      logOutput.newLine();
      for (Map.Entry < String, Integer > entry: counters.entrySet()) {
           logOutput.append(entry.getKey() + "-" + entry.getValue());
	   logOutput.newLine();
      }
      logOutput.close();
    } catch (IOException e){
	    e.printStackTrace();
    }

 }

 @Override
 public void declareOutputFields(OutputFieldsDeclarer declarer) {
  // TODO Auto-generated method stub
 }

 @Override
 public Map < String, Object > getComponentConfiguration() {
  // TODO Auto-generated method stub
  return null;
 }
}