package org.openkilda.wfm.topology.portstate.bolt;

import com.google.common.collect.HashBasedTable;
import com.google.common.collect.Table;
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.Tuple;
import org.openkilda.messaging.Utils;
import org.openkilda.messaging.info.Datapoint;
import org.openkilda.messaging.info.event.PortChangeType;
import org.openkilda.messaging.info.event.PortInfoData;
import org.openkilda.wfm.topology.AbstractTopology;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class ParsePortInfoBolt extends BaseRichBolt {
    private static final Logger logger = LoggerFactory.getLogger(ParsePortInfoBolt.class);
    private static final String METRIC_NAME = "pen.switch.state";
    private OutputCollector collector;
    private Table<String, Integer, Map<String, String>> tagsTable;

    @Override
    public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {
        this.collector = outputCollector;
        tagsTable = HashBasedTable.create();
    }

    @Override
    public void execute(Tuple tuple) {
        try {
            if (tuple.getValueByField(TopoDiscoParseBolt.FIELD_NAME) instanceof PortInfoData) {
                try {
                    List<Object> result = makeTsdbDatapoint(
                            (PortInfoData) tuple.getValueByField(TopoDiscoParseBolt.FIELD_NAME));
                    logger.debug("Emitting: {}", result);
                    collector.emit(result);
                } catch (IOException e) {
                    logger.error("Error creating tsdbDatapoint for: {}", tuple.toString(), e);
                }
            }
        } finally {
            collector.ack(tuple);
        }
    }

    private List<Object> makeTsdbDatapoint(PortInfoData data) throws IOException {
        return tsdbTuple(METRIC_NAME,
                data.getTimestamp(),
                getStateAsInt(data),
                getTags(data));
    }

    private int getStateAsInt(PortInfoData data) {
        return data.getState() == PortChangeType.UP ? 1 : 0;
    }

    private static List<Object> tsdbTuple(String metric, long timestamp, Number value, Map<String, String> tag)
            throws IOException {
        Datapoint datapoint = new Datapoint(metric, timestamp, tag, value);
        return Collections.singletonList(Utils.MAPPER.writeValueAsString(datapoint));
    }

    private Map<String, String> getTags(PortInfoData data) {
        Map<String, String> tag = tagsTable.get(data.getSwitchId(), data.getPortNo());
        if (tag == null) {
            tag = new HashMap<>();
            tag.put("switchid", data.getSwitchId().replaceAll(":", ""));
            tag.put("port", String.valueOf(data.getPortNo()));
            tagsTable.put(data.getSwitchId(), data.getPortNo(), tag);
        }
        return tag;
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(AbstractTopology.fieldMessage);
    }
}

@@ -1545,48 +1545,48 @@
     
-List%3CObject%3E result = makeTsdbDatapoint(
+PortInfoData port = (PortInfoData) tuple
 %0A   
@@ -1610,36 +1610,16 @@
         
-(PortInfoData) tuple
 .getValu
@@ -1657,19 +1657,146 @@
 LD_NAME)
-)
 ;%0A
+%0A                    if (getStateAsInt(port) %3E= 0) %7B%0A                        List%3CObject%3E result = makeTsdbDatapoint(port);%0A    
         
@@ -1857,32 +1857,36 @@
                 
+    
 collector.emit(r
@@ -1885,32 +1885,261 @@
 r.emit(result);%0A
+                    %7D%0A                    else if(logger.isDebugEnabled()) %7B%0A                        List%3CObject%3E result = makeTsdbDatapoint(port);%0A                        logger.debug(%22Skip: %7B%7D%22, result);%0A                    %7D%0A%0A
                 
@@ -2647,24 +2647,25 @@
 data) %7B%0A
+%0A
         
 return d
@@ -2660,60 +2660,333 @@
     
-return data.getState() == PortChangeType.UP ? 1 : 0;
+PortChangeType state = data.getState();%0A%0A        if (state.equals(PortChangeType.UP) %7C%7C state.equals(PortChangeType.ADD)) %7B%0A            return 1;%0A        %7D%0A%0A        if (state.equals(PortChangeType.DOWN) %7C%7C state.equals(PortChangeType.DELETE)) %7B%0A            return 0;%0A        %7D%0A%0A        return -1; // We skip others state here
 %0A   

Fix event choise UD\DOWN for portstate