/* Copyright 2018 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.portstate.spout;

import static java.lang.String.format;
import static org.openkilda.messaging.Utils.PAYLOAD;

import org.openkilda.messaging.Destination;
import org.openkilda.messaging.Message;
import org.openkilda.messaging.Utils;
import org.openkilda.messaging.command.CommandMessage;
import org.openkilda.messaging.command.discovery.PortsCommandData;

import com.fasterxml.jackson.core.JsonProcessingException;
import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichSpout;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;
import java.util.UUID;

public class SwitchPortsSpout extends BaseRichSpout {

    private static final Logger logger = LoggerFactory.getLogger(SwitchPortsSpout.class);
    private static final String CRON_TUPLE = "cron.tuple";
    private static final int DEFAULT_FREQUENCY = 600;
    private static final String REQUESTER = SwitchPortsSpout.class.getSimpleName();
    private final int frequency;
    private SpoutOutputCollector collector;

    public SwitchPortsSpout() {
        this(DEFAULT_FREQUENCY);
    }

    public SwitchPortsSpout(int frequency) {
        this.frequency = frequency;
    }

    private static long now() {
        return System.currentTimeMillis();
    }

    @Override
    public void open(Map map, TopologyContext context, SpoutOutputCollector collector) {
        this.collector = collector;
    }

    @Override
    public void nextTuple() {
        Message message = buildPortsCommand(REQUESTER);
        logger.debug("emitting portsCommand: {}", message.toString());

        try {
            Values values = new Values(PAYLOAD, Utils.MAPPER.writeValueAsString(message));
            collector.emit(values);
        } catch (JsonProcessingException e) {
            logger.error("Error sleeping");
        }

        // Note that no tupleId which means this is an untracked tuple which is
        // required for the sleep
        try {
            Thread.sleep(frequency * 1000);
        } catch (InterruptedException e) {
            logger.error("Error sleeping");
        }
    }

    private Message buildPortsCommand(String requester) {
        String correlationId = format("SwitchPortsSpout-%s", UUID.randomUUID().toString());
        return new CommandMessage(new PortsCommandData(requester), now(), correlationId,
                Destination.CONTROLLER);
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("key", "message"));
    }
}

@@ -707,61 +707,8 @@
 mat;
-%0Aimport static org.openkilda.messaging.Utils.PAYLOAD;
 %0A%0Aim
@@ -1755,19 +1755,20 @@
 e final 
-int
+long
  frequen
@@ -1814,16 +1814,51 @@
 llector;
+%0A    private long lastTickTime = 0;
 %0A%0A    pu
@@ -1998,24 +1998,32 @@
  = frequency
+ * 1000L
 ;%0A    %7D%0A%0A   
@@ -2287,24 +2287,106 @@
 xtTuple() %7B%0A
+        final long now = now();%0A        if (now - lastTickTime %3E frequency) %7B%0A    
         Mess
@@ -2417,25 +2417,16 @@
 Command(
-REQUESTER
 );%0A     
@@ -2424,24 +2424,28 @@
 ();%0A        
+    
 logger.debug
@@ -2455,17 +2455,17 @@
 mitting 
-p
+P
 ortsComm
@@ -2467,16 +2467,20 @@
 sCommand
+Data
 : %7B%7D%22, m
@@ -2489,23 +2489,15 @@
 sage
-.toString());%0A%0A
+);%0A    
     
@@ -2498,32 +2498,36 @@
           try %7B%0A
+    
             Valu
@@ -2553,15 +2553,12 @@
 ues(
-PAYLOAD
+null
 , Ut
@@ -2594,24 +2594,28 @@
 (message));%0A
+    
             
@@ -2638,32 +2638,36 @@
 alues);%0A        
+    
 %7D catch (JsonPro
@@ -2692,32 +2692,36 @@
 ) %7B%0A            
+    
 logger.error(%22Er
@@ -2728,144 +2728,194 @@
 ror 
-sleeping%22);%0A        %7D%0A%0A        // Note that no tupleId which means this is an untracked tuple which is%0A        // required for the sleep
+on json serialization%22, e);%0A            %7D%0A%0A            if (now - lastTickTime %3E frequency * 2) %7B%0A                logger.warn(%22long tick for PortsCommandData - %7B%7Dms%22, now - lastTickTime);
 %0A   
@@ -2911,37 +2911,38 @@
 kTime);%0A        
-try %7B
+    %7D%0A
 %0A            Thr
@@ -2942,136 +2942,84 @@
     
-Thread.sleep(frequency * 1000);%0A        %7D catch (InterruptedException e) %7B%0A            logger.error(%22Error sleeping%22);%0A        %7D
+lastTickTime = now;%0A        %7D%0A        org.apache.storm.utils.Utils.sleep(1);
 %0A   
@@ -3064,24 +3064,8 @@
 and(
-String requester
 ) %7B%0A
@@ -3215,17 +3215,17 @@
 ata(
-requester
+REQUESTER
 ), n

Remove invalid logic of sleeping in spout nextTuple in PortStats