package com.wordcount.example;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.HashMap;
import java.util.Map;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.IRichBolt;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;

public class WordSpitterBolt implements IRichBolt {
	Map<String, Integer> counters;
	private OutputCollector collector;
	Connection conn= null;
	
	public WordSpitterBolt(){
		VerticaCON job=new VerticaCON();
		conn=job.getConnection();		
	}

	@Override
	public void prepare(Map stormConf, TopologyContext context,
			OutputCollector collector) {
		this.counters = new HashMap<String, Integer>();
		this.collector = collector;
	}

	@Override
	public void execute(Tuple input) {
		System.out.println("input value:"+input);
		VerticaInsert vercon=new VerticaInsert();
		try {
			vercon.insertRow(conn,input);
		} catch (SQLException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
/*		try {
			PreparedStatement ps=conn.prepareStatement("INSERT INTO NGOM " + "VALUES ("+input+")");
            int numb_rec=ps.executeUpdate();
			System.out.println("numb_rec:"+numb_rec);
		} catch (SQLException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}*/
		collector.ack(input);
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		//declarer.declare(new Fields("word"));
	}

	@Override
	public void cleanup() {
		System.out.println("CLEAN UP DONE");
	}

	@Override
	public Map<String, Object> getComponentConfiguration() {
		return null;
	}
}