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

/**
 *
 * @author <NAME> <<EMAIL>>
 */
public class Alert<T> extends BaseOperator
{
  protected long lastAlertTimeStamp = -1;
  protected long inAlertSince = -1;
  protected long timeout = 5000; // 5 seconds
  protected long alertFrequency = 0;
  protected long levelOneTimeStamp = 0;
  protected long levelTwoTimeStamp = 0;
  protected long levelThreeTimeStamp = 0;
  @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;
      }
      if (lastAlertTimeStamp + alertFrequency < now) {
        if (inAlertSince >= levelOneTimeStamp) {
          alert1.emit(tuple);
        }
        if (inAlertSince >= levelTwoTimeStamp) {
          alert2.emit(tuple);
        }
        if (inAlertSince >= levelThreeTimeStamp) {
          alert3.emit(tuple);
        }
        lastAlertTimeStamp = now;
      }
    }

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

  public void setAlertFrequency(long millis)
  {
    alertFrequency = millis;
  }

  public void setLevelOneAlertTimeStamp(long millis)
  {
    levelOneTimeStamp = millis;
  }

  public void setLevelTwoAlertTimeStamp(long millis)
  {
    levelTwoTimeStamp = millis;

  }

  public void setLevelThreeAlertTimeStamp(long millis)
  {
    levelThreeTimeStamp = millis;
  }

  public void setTimeout(long millis)
  {
    timeout = millis;
  }

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

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

}<commit_msg>alert timeout bug fix<commit_after>/*
 *  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;

/**
 *
 * @author <NAME> <<EMAIL>>
 */
public class Alert<T> extends BaseOperator
{
  protected long lastAlertTimeStamp = -1;
  protected long inAlertSince = -1;
  protected long lastTupleTimeStamp = -1;
  protected long timeout = 5000; // 5 seconds
  protected long alertFrequency = 0;
  protected long levelOneTimeStamp = 0;
  protected long levelTwoTimeStamp = 0;
  protected long levelThreeTimeStamp = 0;
  @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) {
          alert1.emit(tuple);
        }
        if (inAlertSince >= levelTwoTimeStamp) {
          alert2.emit(tuple);
        }
        if (inAlertSince >= levelThreeTimeStamp) {
          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 void setAlertFrequency(long millis)
  {
    alertFrequency = millis;
  }

  public void setLevelOneAlertTimeStamp(long millis)
  {
    levelOneTimeStamp = millis;
  }

  public void setLevelTwoAlertTimeStamp(long millis)
  {
    levelTwoTimeStamp = millis;

  }

  public void setLevelThreeAlertTimeStamp(long millis)
  {
    levelThreeTimeStamp = millis;
  }

  public void setTimeout(long millis)
  {
    timeout = millis;
  }

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

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

}
{'fraction_non_alphanumeric': 0.09808061420345489, 'fraction_numerical': 0.012092130518234165, 'mean_word_length': 3.4500426985482493, 'pattern_counts': {'":': 0, '<': 31, '<?xml version=': 0, '>': 35, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2365767', 'n_tokens_mistral': 1667, 'n_tokens_neox': 1601, 'n_words': 369}