<commit_before>package cz.seznam.euphoria.core.client.triggers;

import cz.seznam.euphoria.core.client.dataset.windowing.TimeInterval;
import cz.seznam.euphoria.core.client.operator.state.ValueStorage;
import cz.seznam.euphoria.core.client.operator.state.ValueStorageDescriptor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * A {@link Trigger} that is periodically fired based on given time interval.
 * Used to implement "early triggering" functionality.
 */
public class PeriodicTimeTrigger<T> implements Trigger<T, TimeInterval> {

  private static final Logger LOG = LoggerFactory.getLogger(PeriodicTimeTrigger.class);

  /** Next fire stamp (when merging the lowest timestamp is taken) */
  private final ValueStorageDescriptor<Long> fireTimeDescriptor =
          ValueStorageDescriptor.of("fire-time", Long.class, Long.MAX_VALUE, Math::min);

  private final long interval;

  public PeriodicTimeTrigger(long interval) {
    this.interval = interval;
  }

  @Override
  public TriggerResult onElement(long time, T element, TimeInterval window, TriggerContext ctx) {
    ValueStorage<Long> fireStamp = ctx.getValueStorage(fireTimeDescriptor);

    if (fireStamp.get() == Long.MAX_VALUE) {
      // register first timer aligned with window start
      long start = window.getStartMillis() - (window.getStartMillis() % interval);
      long nextFireTimestamp = start + interval;

      ctx.registerTimer(nextFireTimestamp, window);
      fireStamp.set(nextFireTimestamp);
    }

    return TriggerResult.NOOP;
  }

  @Override
  public TriggerResult onTimeEvent(long time, TimeInterval window, TriggerContext ctx) {
    ValueStorage<Long> fireStamp = ctx.getValueStorage(fireTimeDescriptor);

    if (fireStamp.get().equals(time)) {
      LOG.debug("Firing PeriodicTimeTrigger, time {}, window: {}", time, window);
      ctx.registerTimer(time + interval, window);
      fireStamp.set(time + interval);

      return TriggerResult.FLUSH;
    }

    return TriggerResult.NOOP;
  }

  @Override
  public void onClear(TimeInterval window, TriggerContext ctx) {
    ValueStorage<Long> fireStamp = ctx.getValueStorage(fireTimeDescriptor);
    ctx.deleteTimer(fireStamp.get(), window);
    fireStamp.clear();
  }

  @Override
  public TriggerResult onMerge(TimeInterval window, TriggerContext.TriggerMergeContext ctx) {
    ctx.mergeStoredState(fireTimeDescriptor);
    // register timer according to merged state
    ValueStorage<Long> fireStamp = ctx.getValueStorage(fireTimeDescriptor);
    ctx.registerTimer(fireStamp.get(), window);

    return TriggerResult.NOOP;
  }<commit_msg>#! [euphoria-core] Drop logging in periodic-time-triggering<commit_after>package cz.seznam.euphoria.core.client.triggers;

import cz.seznam.euphoria.core.client.dataset.windowing.TimeInterval;
import cz.seznam.euphoria.core.client.operator.state.ValueStorage;
import cz.seznam.euphoria.core.client.operator.state.ValueStorageDescriptor;

/**
 * A {@link Trigger} that is periodically fired based on given time interval.
 * Used to implement "early triggering" functionality.
 */
public class PeriodicTimeTrigger<T> implements Trigger<T, TimeInterval> {

  /** Next fire stamp (when merging the lowest timestamp is taken) */
  private final ValueStorageDescriptor<Long> fireTimeDescriptor =
          ValueStorageDescriptor.of("fire-time", Long.class, Long.MAX_VALUE, Math::min);

  private final long interval;

  public PeriodicTimeTrigger(long interval) {
    this.interval = interval;
  }

  @Override
  public TriggerResult onElement(long time, T element, TimeInterval window, TriggerContext ctx) {
    ValueStorage<Long> fireStamp = ctx.getValueStorage(fireTimeDescriptor);

    if (fireStamp.get() == Long.MAX_VALUE) {
      // register first timer aligned with window start
      long start = window.getStartMillis() - (window.getStartMillis() % interval);
      long nextFireTimestamp = start + interval;

      ctx.registerTimer(nextFireTimestamp, window);
      fireStamp.set(nextFireTimestamp);
    }

    return TriggerResult.NOOP;
  }

  @Override
  public TriggerResult onTimeEvent(long time, TimeInterval window, TriggerContext ctx) {
    ValueStorage<Long> fireStamp = ctx.getValueStorage(fireTimeDescriptor);

    if (fireStamp.get().equals(time)) {
      ctx.registerTimer(time + interval, window);
      fireStamp.set(time + interval);

      return TriggerResult.FLUSH;
    }

    return TriggerResult.NOOP;
  }

  @Override
  public void onClear(TimeInterval window, TriggerContext ctx) {
    ValueStorage<Long> fireStamp = ctx.getValueStorage(fireTimeDescriptor);
    ctx.deleteTimer(fireStamp.get(), window);
    fireStamp.clear();
  }

  @Override
  public TriggerResult onMerge(TimeInterval window, TriggerContext.TriggerMergeContext ctx) {
    ctx.mergeStoredState(fireTimeDescriptor);
    // register timer according to merged state
    ValueStorage<Long> fireStamp = ctx.getValueStorage(fireTimeDescriptor);
    ctx.registerTimer(fireStamp.get(), window);

    return TriggerResult.NOOP;
  }
{'fraction_non_alphanumeric': 0.09920318725099601, 'fraction_numerical': 0.00039840637450199205, 'mean_word_length': 4.4814410480349345, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 17, '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': '5402176', 'n_tokens_mistral': 1462, 'n_tokens_neox': 1428, 'n_words': 329}