package ru.kirillius.pf.sdn.service; import jakarta.annotation.PostConstruct; import jakarta.annotation.PreDestroy; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Service; import ru.kirillius.java.utils.events.EventListener; import ru.kirillius.pf.sdn.api.Networking.Subscriptions.Subscription; import ru.kirillius.pf.sdn.api.flow.Flow; import ru.kirillius.pf.sdn.api.flow.FlowAction; import ru.kirillius.pf.sdn.api.flow.FlowFunction; import ru.kirillius.pf.sdn.api.flow.TriggerType; import ru.kirillius.pf.sdn.entity.ActionConfig; import ru.kirillius.pf.sdn.entity.FlowConfig; import ru.kirillius.pf.sdn.entity.SubscriptionGroup; import ru.kirillius.pf.sdn.flow.DummyFunction; import ru.kirillius.pf.sdn.repository.PipelineConfigRepository; import java.util.Arrays; import java.util.Map; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutorService; import java.util.regex.Pattern; @Slf4j @Service @RequiredArgsConstructor public class FlowService { private final PipelineConfigRepository configRepository; private final FlowExecutorService flowExecutorService; private final SubscriptionService subscriptionService; private final ExecutorService executorService; private final Map flows = new ConcurrentHashMap<>(); private final Map> functions = new ConcurrentHashMap<>(); private EventListener executedListener; private EventListener subscriptionUpdateListener; private EventListener subscriptionSetUpdateListener; public void registerFunction(Class functionClass, String id) { if (functions.containsKey(id)) { throw new IllegalStateException("Function with id '" + id + "' already exists"); } functions.put(id, functionClass); } public Set getFunctionNames() { return functions.keySet(); } public void trigger(long id) { flows.keySet().stream().filter(config -> config.getId().equals(id)).findFirst().ifPresent(this::trigger); } public void trigger(FlowConfig flowConfig) { if (flows.containsKey(flowConfig)) { trigger(flows.get(flowConfig)); } } public void trigger(Flow pipeline) { var matchedConditions = pipeline.getConfig().getStartConditions().stream().filter(c -> c.getTrigger() == TriggerType.Manual).toList(); if (matchedConditions.isEmpty()) { log.error("Unable to manual start pipeline {} because it has no manual trigger", pipeline.getConfig().getName()); return; } matchedConditions.forEach(c -> { if (c.isDontStartIfRunning() && pipeline.isRunning()) { return; } flowExecutorService.triggerExecute(pipeline); }); } public void loadFlow(Flow flow) { loadFlow(flow.getConfig()); } public void loadFlow(FlowConfig flowConfig) { if (configRepository.existsById(flowConfig.getId())) { reloadFlowInternal(flowConfig); } else { flowExecutorService.cancel(flows.get(flowConfig)); flows.remove(flowConfig); } } private void reloadFlowInternal(FlowConfig flowConfig) { if (flows.containsKey(flowConfig)) { flowExecutorService.cancel(flows.get(flowConfig)); } flows.put(flowConfig, new Flow(flowConfig, this::buildAction)); } private FlowAction buildAction(ActionConfig config) { return new FlowAction(functions.getOrDefault(config.getFunctionId(), DummyFunction.class), config); } private void checkForIntervalExecution(Flow pipeline) { var matchedConditions = pipeline.getConfig().getStartConditions().stream().filter(c -> c.getTrigger() == TriggerType.Interval).toList(); if (matchedConditions.isEmpty()) { return; } matchedConditions.forEach(c -> { var properties = c.getProperties(); var timer = properties.getOrDefault("interval", null); if (timer == null) { return; } var delay = Long.parseLong(timer); if (delay <= 0) { return; } executorService.submit(() -> { try { Thread.sleep(delay * 60000L); } catch (InterruptedException e) { return; } if (c.isDontStartIfRunning() && pipeline.isRunning()) { return; } flowExecutorService.triggerExecute(pipeline); }); }); } @PostConstruct private void initialize() { subscriptionUpdateListener = subscriptionService.getUpdateEvent().add(this::subscriptionUpdate); subscriptionSetUpdateListener = subscriptionService.getSetUpdateEvent().add(this::subscriptionSetUpdate); executedListener = flowExecutorService.getOnExecuted().add(this::checkForIntervalExecution); registerFunction(DummyFunction.class, "Error:fallback"); configRepository.findAll().forEach(this::reloadFlowInternal); flows.values() .stream() .filter(p -> p .getConfig() .getStartConditions() .stream() .anyMatch(c -> c.getTrigger() == TriggerType.OnStart)) .forEach(flowExecutorService::triggerExecute); } private void subscriptionSetUpdate(SubscriptionGroup setEntry) { flows.values().forEach(pipeline -> { var matchedConditions = pipeline.getConfig().getStartConditions().stream().filter(c -> c.getTrigger() == TriggerType.OnSubscriptionGroupUpdate).toList(); if (matchedConditions.isEmpty()) { return; } matchedConditions.forEach(c -> { if (c.isDontStartIfRunning() && pipeline.isRunning()) { return; } var properties = c.getProperties(); var names = properties.getOrDefault("names", null); if (names == null || names.isEmpty()) { flowExecutorService.triggerExecute(pipeline); return; } if (Arrays.stream(names.split(Pattern.quote(","))).anyMatch(s -> setEntry.getName().equals(s))) { flowExecutorService.triggerExecute(pipeline); } }); }); } private void subscriptionUpdate(Subscription subscription) { flows.values().forEach(pipeline -> { var matchedConditions = pipeline.getConfig().getStartConditions().stream().filter(c -> c.getTrigger() == TriggerType.OnSubscriptionUpdate).toList(); if (matchedConditions.isEmpty()) { return; } matchedConditions.forEach(c -> { if (c.isDontStartIfRunning() && pipeline.isRunning()) { return; } var properties = c.getProperties(); var names = properties.getOrDefault("names", null); if (names == null || names.isEmpty()) { flowExecutorService.triggerExecute(pipeline); return; } if (Arrays.stream(names.split(Pattern.quote(","))).anyMatch(s -> subscription.getName().equals(s))) { flowExecutorService.triggerExecute(pipeline); } }); }); } @PreDestroy private void destroy() { if (executedListener != null) { flowExecutorService.getOnExecuted().remove(executedListener); executedListener = null; } if (subscriptionUpdateListener != null) { subscriptionService.getUpdateEvent().remove(subscriptionUpdateListener); subscriptionUpdateListener = null; } if (subscriptionSetUpdateListener != null) { subscriptionService.getSetUpdateEvent().remove(subscriptionSetUpdateListener); subscriptionSetUpdateListener = null; } } }