package storm.cookbook.log;

import backtype.storm.Config;
import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Tuple;
import org.elasticsearch.client.Client;
import org.elasticsearch.node.Node;
import org.elasticsearch.node.NodeBuilder;

import java.util.Map;
import java.util.logging.Logger;

/**
 * User: domenicosolazzo
 */
public class IndexerBolt extends BaseRichBolt{
    // Elastic search client
    private Client client;
    // Logger
    private Logger LOG = Logger.getLogger(IndexerBolt.class.toString());

    // Collector
    private OutputCollector collector;

    private static final String INDEX_NAME = "logstorm";
    private static final String INDEX_TYPE = "logentry";
    @Override
    public void prepare(Map stormConf, TopologyContext topologyContext, OutputCollector outputCollector) {
        this.collector = outputCollector;

        Node node;
        if((Boolean)stormConf.get(Config.TOPOLOGY_DEBUG) == true){
            // Topology for local testing
            node = NodeBuilder.nodeBuilder().local(true).node();
        }else{
            String clusterName = (String) stormConf.get(Conf.ELASTIC_CLUSTER_NAME);
            if (clusterName == null){
                clusterName = Conf.DEFAULT_ELASTIC_CLUSTER;
            }
            node = NodeBuilder.nodeBuilder().clusterName(clusterName).node();
        }
        client = node.client();

    }

    @Override
    public void execute(Tuple tuple) {

    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {

    }
}

@@ -271,20 +271,145 @@
 m.tuple.
-Tupl
+Fields;%0Aimport backtype.storm.tuple.Tuple;%0Aimport backtype.storm.tuple.Values;%0Aimport org.elasticsearch.action.index.IndexRespons
 e;%0Aimpor
@@ -521,16 +521,58 @@
 Builder;
+%0Aimport storm.cookbook.log.model.LogEntry;
 %0A%0Aimport
@@ -1764,16 +1764,840 @@
 tuple) %7B
+%0A        LogEntry entry = (LogEntry) tuple.getValueByField(FieldNames.LOG_ENTRY);%0A        if (entry == null)%7B%0A            LOG.fatal(%22The entry is null or invalid.%22);%0A            return;%0A        %7D%0A        String toBeIndexed = entry.toJSON().toJSONString();%0A        IndexResponse response = client.prepareIndex(INDEX_NAME, INDEX_TYPE)%0A                .setSource(toBeIndexed)%0A                .execute().actionGet();%0A        if ( response == null )%7B%0A            LOG.error(%22Error indexing the tuple: %22 + tuple.toString());%0A        %7Delse if (response.getId() == null)%7B%0A            LOG.error(%22Index id is null. Error indexing the tuple: %22 + tuple.toString());%0A%0A        %7Delse%7B%0A            LOG.debug(%22The tuple has been indexed. %22 + input.toString());%0A            this.collector.emit(new Values( entry, response.getId() ));%0A        %7D
 %0A%0A    %7D%0A
@@ -2692,16 +2692,112 @@
 arer) %7B%0A
+        outputFieldsDeclarer.declare(new Fields(FieldNames.LOG_ENTRY, FieldNames.LOG_INDEX_ID));
 %0A    %7D%0A%7D

Update the Indexer bolt. Implemented the methods