/* 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.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;

import edu.emory.mathcs.backport.java.util.concurrent.TimeUnit;

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() {
    	Map<String, Object> conf = super.getComponentConfiguration();
		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);
    }
}

@@ -675,228 +675,109 @@
 ort 
-java.util.HashMap;%0Aimport java.util.List;%0Aimport java.util.Map;%0Aimport java.util.Set;%0Aimport java.util.stream.Collectors;%0Aimport java.util.stream.Stream;%0A%0Aimport org.apache.storm.Config;%0Aimport org.apache.storm.Constants
+static org.openkilda.messaging.Utils.MAPPER;%0A%0Aimport org.slf4j.LoggerFactory;%0Aimport org.slf4j.Logger
 ;%0Aim
@@ -1167,99 +1167,162 @@
 nt;%0A
+%0A
 import 
-org.slf4j.Logger;%0Aimport org.slf4j.LoggerFactory;%0A%0Aimport edu.emory.mathcs.back
+java.util.*;%0Aimport java.util.concurrent.ConcurrentHashMap;%0Aimport java.util.concurrent.ConcurrentMap;%0Aimport java.util.stream.Collectors;%0Aim
 port
-.
+ 
 java
@@ -1331,27 +1331,21 @@
 til.
-concurrent.TimeUnit
+stream.Stream
 ;%0A%0Ap
@@ -1539,60 +1539,16 @@
 S = 
-TimeUnit.SECONDS.convert(10, TimeUnit.MINUTES);%0A    
+60000L;%0A
 %0A   
@@ -1939,16 +1939,26 @@
 private 
+Concurrent
 Map%3CInte
@@ -1987,16 +1987,26 @@
 e = new 
+Concurrent
 HashMap%3C
@@ -2195,28 +2195,24 @@
 ctor;%0A    %7D%0A
-    
 %0A    @Overri
@@ -2218,605 +2218,46 @@
 ide%0A
-%09public Map%3CString, Object%3E getComponentConfiguration() %7B%0A    %09Map%3CString, Object%3E conf = super.getComponentConfiguration();%0A%09%09conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS, TEN_MINUTES);%0A%09%09return conf;%0A%09%7D%0A%0A%0A%09@Override%0A    public void execute(Tuple tuple) %7B%0A%09%09%0A%09%09if (isTickTuple(tuple)) %7B%0A%09%09%09Set%3CInteger%3E keys = storage.keySet();%0A%09%09%09// opentsdb using current epoch time (date +%25s) in seconds%0A%09%09%09long now  = System.currentTimeMillis()/1000;%0A%09%09%09for (Integer key: keys) %7B%0A%09%09%09%09storage.compute(key, (k, v) -%3E now - v.getTime() %3E TEN_MINUTES ? null: 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
+    public void execute(Tuple tuple) %7B
 %0A   
@@ -2510,24 +2510,25 @@
 tapoint%22);%0A%0A
+%0A
         if (
@@ -3726,316 +3726,6 @@
   %7D%0A
-    %0A    private boolean isTickTuple(Tuple tuple) %7B%0A    %09String sourceComponent = tuple.getSourceComponent();%0A    %09String sourceStreamId = tuple.getSourceStreamId();%0A    %09%0A    %09return sourceComponent.equals(Constants.SYSTEM_COMPONENT_ID) &&%0A    %09%09%09sourceStreamId.equals(Constants.SYSTEM_TICK_STREAM_ID);%0A    %7D%0A
 %7D%0A

Revert "periodically purge cache of data points"