/* Copyright 2017 Telstra Open Source
 *
 *   Licensed under the Apache License, Version 2.0 (the "License");
 *   you may not use this file except in compliance with the License.
 *   You may obtain a copy of the License at
 *
 *       http://www.apache.org/licenses/LICENSE-2.0
 *
 *   Unless required by applicable law or agreed to in writing, software
 *   distributed under the License is distributed on an "AS IS" BASIS,
 *   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *   See the License for the specific language governing permissions and
 *   limitations under the License.
 */

package org.openkilda.wfm.topology.opentsdb.bolts;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import org.apache.storm.Config;
import org.apache.storm.Constants;
import org.apache.storm.opentsdb.bolt.TupleOpenTsdbDatapointMapper;
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.openkilda.messaging.info.Datapoint;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


public class OpenTSDBFilterBolt extends BaseRichBolt {

    private static final Logger LOGGER = LoggerFactory.getLogger(OpenTSDBFilterBolt.class);
    private static final long TEN_MINUTES = TimeUnit.SECONDS.convert(10, TimeUnit.MINUTES);
    
    private static final Fields DECLARED_FIELDS =
            new Fields(TupleOpenTsdbDatapointMapper.DEFAULT_MAPPER.getMetricField(),
                    TupleOpenTsdbDatapointMapper.DEFAULT_MAPPER.getTimestampField(),
                    TupleOpenTsdbDatapointMapper.DEFAULT_MAPPER.getValueField(),
                    TupleOpenTsdbDatapointMapper.DEFAULT_MAPPER.getTagsField());

    private Map<Integer, Datapoint> storage = new HashMap<>();
    private OutputCollector collector;

    @Override
    public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
        this.collector = collector;
    }
    
    @Override
	public Map<String, Object> getComponentConfiguration() {
    	Config conf = new Config();
		conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS, TEN_MINUTES);
		return conf;
	}


	@Override
    public void execute(Tuple tuple) {
		
		if (isTickTuple(tuple)) {
			Set<Integer> keys = storage.keySet();
			// opentsdb using current epoch time (date +%s) in seconds
			long now  = System.currentTimeMillis()/1000;
			for (Integer key: keys) {
				storage.compute(key, (k, v) -> now - v.getTime() > TEN_MINUTES ? null: v);
			}
			collector.ack(tuple);
			return;		
		}
		
        if (!tuple.contains("datapoint")) { //TODO: Should make sure tuple comes from correct bolt, ie not TickTuple
            collector.ack(tuple);
            return;
        }

        Datapoint datapoint = (Datapoint) tuple.getValueByField("datapoint");

        if (isUpdateRequired(datapoint)) {
            addDatapoint(datapoint);

            List<Object> stream = Stream.of(datapoint.getMetric(), datapoint.getTime(), datapoint.getValue(),
                    datapoint.getTags()).collect(Collectors.toList());

            LOGGER.debug("emit: " + stream);
            collector.emit(stream);
        }
        collector.ack(tuple);
    }

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

    private void addDatapoint(Datapoint datapoint) {
        LOGGER.debug("adding datapoint: " + datapoint.simpleHashCode());
        LOGGER.debug("storage.size: " + storage.size());
        storage.put(datapoint.simpleHashCode(), datapoint);
    }

    private boolean isUpdateRequired(Datapoint datapoint) {
        boolean update = true;
        if (storage.containsKey(datapoint.simpleHashCode())) {
            Datapoint prevDatapoint = storage.get(datapoint.simpleHashCode());
            update = !prevDatapoint.getValue().equals(datapoint.getValue()) ||
                    datapoint.getTime() - prevDatapoint.getTime() >= TEN_MINUTES;
        }
        return update;
    }
    
    private boolean isTickTuple(Tuple tuple) {
    	String sourceComponent = tuple.getSourceComponent();
    	String sourceStreamId = tuple.getSourceStreamId();
    	
    	return sourceComponent.equals(Constants.SYSTEM_COMPONENT_ID) &&
    			sourceStreamId.equals(Constants.SYSTEM_TICK_STREAM_ID);
    }
}

@@ -2278,17 +2278,20 @@
 verride%0A
-%09
+    
 public M
@@ -2343,17 +2343,20 @@
 ) %7B%0A    
-%09
+    
 Config c
@@ -2375,18 +2375,24 @@
 nfig();%0A
-%09%09
+        
 conf.put
@@ -2444,18 +2444,24 @@
 NUTES);%0A
-%09%09
+        
 return c
@@ -2469,14 +2469,20 @@
 nf;%0A
-%09%7D%0A%0A%0A%09
+    %7D%0A%0A%0A    
 @Ove
@@ -2526,21 +2526,33 @@
 uple) %7B%0A
-%09%09%0A%09%09
+        %0A        
 if (isTi
@@ -2569,19 +2569,28 @@
 ple)) %7B%0A
-%09%09%09
+            
 Set%3CInte
@@ -2619,19 +2619,28 @@
 ySet();%0A
-%09%09%09
+            
 // opent
@@ -2690,19 +2690,28 @@
 seconds%0A
-%09%09%09
+            
 long now
@@ -2747,19 +2747,28 @@
 )/1000;%0A
-%09%09%09
+            
 for (Int
@@ -2789,12 +2789,24 @@
 ) %7B%0A
-%09%09%09%09
+                
 stor
@@ -2880,57 +2880,94 @@
 v);%0A
-%09%09%09%7D%0A%09%09%09collector.ack(tuple);%0A%09%09%09return;%09%09%0A%09%09%7D%0A%09%09
+            %7D%0A            collector.ack(tuple);%0A            return;%0A        %7D%0A        
 %0A   
@@ -4483,25 +4483,28 @@
 uple) %7B%0A    
-%09
+    
 String sourc
@@ -4548,17 +4548,20 @@
 ();%0A    
-%09
+    
 String s
@@ -4611,15 +4611,21 @@
     
-%09%0A
+    %0A   
     
-%09
+ 
 retu
@@ -4692,11 +4692,20 @@
     
-%09%09%09
+            
 sour

replace tab with spaces