package com.enow.storm.TriggerTopology;

import java.util.Map;

import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;
import org.bson.Document;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.collect.Lists;
import com.mongodb.MongoClient;
import com.mongodb.WriteConcern;
import com.mongodb.client.FindIterable;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.MongoDatabase;

import java.util.List;

public class StagingBolt extends BaseRichBolt {
    protected static final Logger LOG = LoggerFactory.getLogger(CallingKafkaBolt.class);
    private OutputCollector collector;
    private TopicStructure ts;
    boolean machineIdCheck = false;
    boolean phaseRoadMapIdCheck = false;

    @Override

    public void prepare(Map MongoConf, TopologyContext context, OutputCollector collector) {
        this.collector = collector;
        ts = new TopicStructure();
    }

    @Override
    public void execute(Tuple input) {

        if (null == input.getValueByField("topicStucture")) {
            return;
        } else if ((null == input.getStringByField("msg") || input.getStringByField("msg").length() == 0)) {
            return;
        }

        ts = (TopicStructure) input.getValueByField("topicStucture");
        final String msg = input.getStringByField("msg");

        MongoClient mongoClient = new MongoClient("52.193.56.228", 9092);

        mongoClient.setWriteConcern(WriteConcern.ACKNOWLEDGED);
        MongoDatabase dbWrite = mongoClient.getDatabase("enow");
        MongoCollection<Document> deviceListCollection = dbWrite.getCollection("device");

        if (deviceListCollection.count(new Document("deviceId", ts.getDeviceId())) == 0) {
            machineIdCheck = false;
        } else if (deviceListCollection.count(new Document("deviceId", ts.getDeviceId())) == 1) {
            machineIdCheck = true;
        } else {
//			machineIdCheck = "device id : now we have a problem";
            LOG.debug("There are more than two machine ID on MongoDB");
        }

        MongoCollection<Document> phaseRoadMapCollection = dbWrite.getCollection("phaseRoadMap");

        if (phaseRoadMapCollection.count(new Document("phaseRoadMapId", ts.getPhaseRoadMapId())) == 0) {
            phaseRoadMapIdCheck = false;
        } else if (phaseRoadMapCollection.count(new Document("phaseRoadMapId", ts.getPhaseRoadMapId())) == 1) {
            phaseRoadMapIdCheck = true;
        } else {
//			phaseRoadMapIdCheck = "phase road map id : now we have a problem";
            LOG.debug("There are more than two Phase Roadmap Id on MongoDB");
        }
        collector.emit(new Values(ts, msg, machineIdCheck, phaseRoadMapIdCheck));

        try {
            LOG.debug("input = [" + input + "]");
            collector.ack(input);
        } catch (Exception e) {
            collector.fail(input);
        }
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("topicStucture", "msg", "machineIdCheck", "phaseRoadMapIdCheck"));
    }
}
@@ -911,17 +911,52 @@
 ucture t
-s
+opicStructure;%0A    String msg = null
 ;%0A    bo
@@ -1180,17 +1180,29 @@
        t
-s
+opicStructure
  = new T
@@ -1278,25 +1278,24 @@
 nput) %7B%0A
-%0A
         
 if (null
@@ -1290,19 +1290,41 @@
     
-if (null ==
+topicStructure = (TopicStructure)
  inp
@@ -1358,16 +1358,97 @@
 ucture%22)
+;%0A        msg = input.getStringByField(%22msg%22);%0A        if (null == topicStructure
 ) %7B%0A    
@@ -1495,70 +1495,18 @@
  == 
-input.getStringByField(%22
 msg
-%22)
  %7C%7C 
-input.getStringByField(%22
 msg
-%22)
 .len
@@ -1550,25 +1550,24 @@
       %7D%0A
-%0A
         
 ts = (To
@@ -1562,128 +1562,26 @@
     
-ts = (TopicStructure) input.getValueByField(%22topicStucture%22);%0A        final String msg = input.getStringByField(%22msg%22);%0A
+// check Device ID
 %0A   
@@ -1929,33 +1929,45 @@
 nt(%22deviceId%22, t
-s
+opicStructure
 .getDeviceId()))
@@ -2079,25 +2079,37 @@
 deviceId%22, t
-s
+opicStructure
 .getDeviceId
@@ -2173,21 +2173,22 @@
  else %7B%0A
-//
 %09%09%09
+// 
 machineI
@@ -2311,24 +2311,58 @@
 ;%0A        %7D%0A
+        // check Phase Road-map ID
 %0A        Mon
@@ -2514,33 +2514,45 @@
 aseRoadMapId%22, t
-s
+opicStructure
 .getPhaseRoadMap
@@ -2687,17 +2687,29 @@
 apId%22, t
-s
+opicStructure
 .getPhas
@@ -2792,13 +2792,23 @@
 e %7B%0A
-//%09%09%09
+            // 
 phas
@@ -2993,17 +2993,29 @@
 Values(t
-s
+opicStructure
 , msg, m
@@ -3428,8 +3428,9 @@
 %0A    %7D%0A%7D
+%0A

clear valuables name