/*
 *  Copyright (c) 2012-2013 Malhar, Inc.
 *  All Rights Reserved.
 */
package com.malhartech.lib.util;

import com.malhartech.annotation.InputPortFieldAnnotation;
import com.malhartech.annotation.OutputPortFieldAnnotation;
import com.malhartech.api.BaseOperator;
import com.malhartech.api.DefaultInputPort;
import com.malhartech.api.DefaultOutputPort;
import javax.validation.constraints.Min;

/**
 *
 * @param <T>
 * @author David Yan <davidyan@malhar-inc.com>
 */
public class Alert<T> extends BaseOperator
{
  protected long lastAlertTimeStamp = -1;
  protected long inAlertSince = -1;
  protected long lastTupleTimeStamp = -1;

  @Min(0)
  protected long timeout = 5000; // 5 seconds

  @Min(0)
  protected long alertFrequency = 0;
  protected long levelOneTimeStamp = 0;
  protected long levelTwoTimeStamp = 0;
  protected long levelThreeTimeStamp = 0;

  boolean alertOn = true;

  @InputPortFieldAnnotation(name = "in", optional = false)
  public final transient DefaultInputPort<T> in = new DefaultInputPort<T>(this)
  {
    @Override
    public void process(T tuple)
    {
      long now = System.currentTimeMillis();
      if (inAlertSince < 0) {
        inAlertSince = now;
      }
      lastTupleTimeStamp = now;
      if (lastAlertTimeStamp + alertFrequency < now) {
        if (inAlertSince >= levelOneTimeStamp) {
          if (alertOn) {
            alert1.emit(tuple);
          }
        }
        if (inAlertSince >= levelTwoTimeStamp) {
          if (alertOn) {
            alert2.emit(tuple);
          }
        }
        if (inAlertSince >= levelThreeTimeStamp) {
          if (alertOn) {
            alert3.emit(tuple);
          }
        }
        lastAlertTimeStamp = now;
      }
    }

  };
  @OutputPortFieldAnnotation(name = "alert1", optional = false)
  public final transient DefaultOutputPort<T> alert1 = new DefaultOutputPort<T>(this);
  @OutputPortFieldAnnotation(name = "alert2", optional = true)
  public final transient DefaultOutputPort<T> alert2 = new DefaultOutputPort<T>(this);
  @OutputPortFieldAnnotation(name = "alert3", optional = true)
  public final transient DefaultOutputPort<T> alert3 = new DefaultOutputPort<T>(this);

  public long getTimeout()
  {
    return timeout;
  }

  public void setTimeout(long timeout)
  {
    this.timeout = timeout;
  }

  public long getAlertFrequency()
  {
    return alertFrequency;
  }

  public void setAlertFrequency(long alertFrequency)
  {
    this.alertFrequency = alertFrequency;
  }

  public long getLevelOneAlertTime()
  {
    return levelOneAlertTime;
  }

  public void setLevelOneAlertTime(long levelOneAlertTime)
  {
    this.levelOneAlertTime = levelOneAlertTime;
  }

  public long getLevelTwoAlertTime()
  {
    return levelTwoAlertTime;
  }

  public void setLevelTwoAlertTime(long levelTwoAlertTime)
  {
    this.levelTwoAlertTime = levelTwoAlertTime;
  }

  public long getLevelThreeAlertTime()
  {
    return levelThreeAlertTime;
  }

  public void setLevelThreeAlertTime(long levelThreeAlertTime)
  {
    this.levelThreeAlertTime = levelThreeAlertTime;
  }

  public boolean isActivated()
  {
    return activated;
  }

  public void setActivated(boolean activated)
  {
    this.activated = activated;
  }

  public void setAlertOn(boolean flag)
  {
    alertOn = flag;
  }

  @Override
  public void endWindow()
  {
    checkTimeout();
  }

  protected void checkTimeout()
  {
    if (System.currentTimeMillis() - lastTupleTimeStamp > timeout) {
      inAlertSince = -1;
      lastAlertTimeStamp = -1;
    }
  }

}

@@ -351,49 +351,8 @@
 ort;
-%0Aimport javax.validation.constraints.Min;
 %0A%0A/*
@@ -585,27 +585,16 @@
 mp = -1;
-%0A%0A  @Min(0)
 %0A  prote
@@ -635,19 +635,8 @@
 onds
-%0A%0A  @Min(0)
 %0A  p
@@ -694,25 +694,25 @@
 levelOne
-TimeStamp
+AlertTime
  = 0;%0A  
@@ -734,25 +734,25 @@
 levelTwo
-TimeStamp
+AlertTime
  = 0;%0A  
@@ -780,25 +780,34 @@
 hree
-TimeStamp = 0;%0A%0A 
+AlertTime = 0;%0A  protected
  boo
@@ -812,22 +812,24 @@
 oolean a
-lertOn
+ctivated
  = true;
@@ -829,17 +829,16 @@
 = true;%0A
-%0A
   @Input
@@ -1178,16 +1178,29 @@
     if (
+activated && 
 lastAler
@@ -1276,48 +1276,21 @@
 lOne
-TimeStamp) %7B%0A          if (alertOn
+AlertTime
 ) %7B%0A
-  
     
@@ -1311,36 +1311,24 @@
 mit(tuple);%0A
-          %7D%0A
         %7D%0A  
@@ -1365,48 +1365,21 @@
 lTwo
-TimeStamp) %7B%0A          if (alertOn
+AlertTime
 ) %7B%0A
-  
     
@@ -1400,36 +1400,24 @@
 mit(tuple);%0A
-          %7D%0A
         %7D%0A  
@@ -1456,48 +1456,21 @@
 hree
-TimeStamp) %7B%0A          if (alertOn
+AlertTime
 ) %7B%0A
-  
     
@@ -1499,20 +1499,8 @@
 e);%0A
-          %7D%0A
     
@@ -3058,76 +3058,8 @@
  %7D%0A%0A
-  public void setAlertOn(boolean flag)%0A  %7B%0A    alertOn = flag;%0A  %7D%0A%0A
   @O

Put back David's code for activation