analytics/src/main/java/cz/senslog/analyzer/analysis/module/ThresholdHandler.java

package cz.senslog.analyzer.analysis.module;

import cz.senslog.analyzer.domain.*;
import cz.senslog.analyzer.core.api.Handler;
import cz.senslog.analyzer.core.api.HandlerContext;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import java.util.*;

import static java.util.Collections.emptyList;

public abstract class ThresholdHandler<T extends Data<?, ?>> extends Handler<T, T> {

    private static final Logger logger = LogManager.getLogger(ThresholdHandler.class);

    /** Map of thresholds rules order by group_id (Map<group_id, List<ThresholdRule>>). */
    private Map<Long, List<Threshold.Rule>> thresholdRules;

    protected abstract List<Threshold> loadThresholdValues();
    protected abstract long getGroupId(T data);

    @Override
    public void init() {
        List<Threshold> thresholds = loadThresholdValues();
        thresholdRules = new HashMap<>(thresholds.size());
        for (Threshold threshold : thresholds) {
            thresholdRules.computeIfAbsent(threshold.getGroupId(), k -> new ArrayList<>())
                    .add(threshold.getRule());
        }
    }

    @Override
    public void handle(HandlerContext<T, T> context) {
        T data = context.data();
        long groupId = getGroupId(context.data());
        List<Threshold.Rule> rules = getThresholdRulesByGroupId(groupId);
        ValidationResult validateResult = data.validate(rules);

        if (validateResult.isNotValid()) {
            context.eventBus().notify(new Alert(
                    groupId, String.format("group(%d)", groupId), data.getTimestamp(),
                    validateResult.getMessages().toArray(new String[0])
            ));
        }

        context.next(data);
    }

    private List<Threshold.Rule> getThresholdRulesByGroupId(long groupId) {
        return thresholdRules.getOrDefault(groupId, emptyList());
    }
}