analytics/src/main/java/cz/senslog/analytics/utils/validator/ThresholdChecker.java

package cz.senslog.analytics.utils.validator;

import cz.senslog.analytics.domain.*;

import java.time.OffsetDateTime;
import java.util.*;
import java.util.function.Consumer;
import java.util.function.Predicate;

import static java.util.Collections.emptyList;

public class ThresholdChecker<DS extends TimeSeriesDatasource> {

    public static <T extends TimeSeriesDatasource> ThresholdChecker<T> disabled() {
        return ThresholdChecker.create(emptyList(), null, null);
    }

    private final Map<Long, List<Threshold>> thresholds;

    private final Validator<DS> validator;

    private final Consumer<ViolationReport> ifViolated;

    private final Map<Long, NotifyTrigger> notifyTriggerMap;

    public static <DS extends TimeSeriesDatasource> ThresholdChecker<DS> create(List<Threshold> thresholds,
                                                           Validator<DS> validator,
                                                           Consumer<ViolationReport> ifViolated)
    {
        return new ThresholdChecker<>(thresholds, validator, ifViolated);
    }

    public ThresholdChecker(List<Threshold> thresholds,
                            Validator<DS> validator,
                            Consumer<ViolationReport> ifViolated
    ) {
        this.validator = validator;
        this.ifViolated = ifViolated;

        this.notifyTriggerMap = new HashMap<>();
        this.thresholds = new HashMap<>();
        for (Threshold th : thresholds) {
            long sourceId = th.groupId();
            if (!notifyTriggerMap.containsKey(sourceId)) {
                this.notifyTriggerMap.put(sourceId, th.notifyTriggerType().createInstance());
            }
            this.thresholds.computeIfAbsent(th.groupId(), k -> new ArrayList<>()).add(th);
        }
    }

    public Predicate<DS> check() {
        return this::validateThreshold;
    }

    private boolean validateThreshold(DS data) {
        long sourceId = data.datasourceId();
        OffsetDateTime timestamp = data.timestamp();

        if (!thresholds.containsKey(sourceId)) {
            return true;
        }
        boolean process = true;
        List<Threshold> rules = thresholds.get(sourceId);

        NotifyTrigger notifier = notifyTriggerMap.get(sourceId);
        for (Threshold th : rules) {
            ValidationResult res = validator.validate(data, th);
            notifier.accept(res);

            if (res.isNotValid() && !th.allowProcess()) {
                process = false;
            }
        }

        if (notifier.shouldNotify()) {
            ifViolated.accept(new ViolationReport(sourceId, timestamp, notifier.resultsToNotify()));
        }

        return process;
    }
}