X-Git-Url: https://git.pterodactylus.net/?a=blobdiff_plain;ds=sidebyside;f=src%2Fmain%2Fjava%2Fnet%2Fpterodactylus%2Frhynodge%2Fengine%2FEngine.java;h=8ff70201734f934946e03c286b79e78069ab00d4;hb=cf2f4441705f9a5941022fb0837d08c95a1d0633;hp=423eaafcaf59119ab35970c4a96540ce85a6fcd0;hpb=bf289dc06386026dc59f6d10ce421447fe6dd5aa;p=rhynodge.git
diff --git a/src/main/java/net/pterodactylus/rhynodge/engine/Engine.java b/src/main/java/net/pterodactylus/rhynodge/engine/Engine.java
index 423eaaf..8ff7020 100644
--- a/src/main/java/net/pterodactylus/rhynodge/engine/Engine.java
+++ b/src/main/java/net/pterodactylus/rhynodge/engine/Engine.java
@@ -17,54 +17,42 @@
package net.pterodactylus.rhynodge.engine;
-import static com.google.common.base.Optional.absent;
-import static com.google.common.base.Optional.of;
-import static com.google.common.collect.Maps.newTreeMap;
-import static java.lang.String.format;
+import static java.lang.System.currentTimeMillis;
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
-import java.util.HashMap;
import java.util.Map;
-import java.util.Map.Entry;
-import java.util.SortedMap;
+import java.util.Optional;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Future;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+
+import javax.inject.Inject;
+import javax.inject.Singleton;
-import net.pterodactylus.rhynodge.Filter;
-import net.pterodactylus.rhynodge.Query;
import net.pterodactylus.rhynodge.Reaction;
-import net.pterodactylus.rhynodge.Trigger;
-import net.pterodactylus.rhynodge.states.AbstractState;
-import net.pterodactylus.rhynodge.states.FailedState;
+import net.pterodactylus.rhynodge.actions.EmailAction;
import net.pterodactylus.rhynodge.states.StateManager;
-import com.google.common.base.Optional;
-import com.google.common.util.concurrent.AbstractExecutionThreadService;
-import org.apache.commons.lang3.tuple.Pair;
-import org.apache.log4j.Logger;
-
/**
* Rhynodge main engine.
*
* @author David âBombeâ Roden
*/
-public class Engine extends AbstractExecutionThreadService {
-
- /** The logger. */
- private static final Logger logger = Logger.getLogger(Engine.class);
+@Singleton
+public class Engine {
- /** The state manager. */
private final StateManager stateManager;
+ private final ScheduledExecutorService executorService;
+ private final Map> scheduledFutures = new ConcurrentHashMap<>();
+ private final EmailAction errorEmailAction;
- /** All defined reactions. */
- /* synchronize on itself. */
- private final Map reactions = new HashMap();
-
- /**
- * Creates a new engine.
- *
- * @param stateManager
- * The state manager
- */
- public Engine(StateManager stateManager) {
+ @Inject
+ public Engine(StateManager stateManager, EmailAction errorEmailAction) {
this.stateManager = stateManager;
+ this.errorEmailAction = errorEmailAction;
+ executorService = new ScheduledThreadPoolExecutor(1);
}
//
@@ -75,181 +63,31 @@ public class Engine extends AbstractExecutionThreadService {
* Adds the given reaction to this engine.
*
* @param name
- * The name of the reaction
+ * The name of the reaction
* @param reaction
- * The reaction to add to this engine
+ * The reaction to add to this engine
*/
public void addReaction(String name, Reaction reaction) {
- synchronized (reactions) {
- reactions.put(name, reaction);
- reactions.notifyAll();
- }
+ ReactionState reactionState = new ReactionState(stateManager, name);
+ Optional lastState = reactionState.loadLastState();
+ long lastExecutionTime = lastState.map(net.pterodactylus.rhynodge.State::time).orElse(0L);
+ long nextExecutionTime = lastExecutionTime + reaction.updateInterval();
+ ReactionRunner reactionRunner = new ReactionRunner(reaction, reactionState, errorEmailAction);
+ ScheduledFuture> future = executorService.scheduleWithFixedDelay(reactionRunner, nextExecutionTime - currentTimeMillis(), reaction.updateInterval(), MILLISECONDS);
+ scheduledFutures.put(name, future);
}
/**
* Removes the reaction with the given name.
*
* @param name
- * The name of the reaction to remove
+ * The name of the reaction to remove
*/
public void removeReaction(String name) {
- synchronized (reactions) {
- if (!reactions.containsKey(name)) {
- return;
- }
- reactions.remove(name);
- reactions.notifyAll();
- }
- }
-
- //
- // ABSTRACTSERVICE METHODS
- //
-
- /**
- * {@inheritDoc}
- */
- @Override
- public void run() {
- while (isRunning()) {
- Optional nextReaction = getNextReaction();
- if (!nextReaction.isPresent()) {
- continue;
- }
-
- String reactionName = nextReaction.get().getKey();
- logger.debug(format("Next Reaction: %s.", reactionName));
-
- /* wait until the next reaction has to run. */
- Optional lastState = stateManager.loadLastState(reactionName);
- long lastStateTime = lastState.isPresent() ? lastState.get().time() : 0;
- int lastStateFailCount = lastState.isPresent() ? lastState.get().failCount() : 0;
- long waitTime = (lastStateTime + nextReaction.get().getReaction().updateInterval()) - System.currentTimeMillis();
- logger.debug(format("Time to wait for next Reaction: %d millseconds.", waitTime));
- if (waitTime > 0) {
- synchronized (reactions) {
- try {
- logger.info(format("Waiting until %tc.", lastStateTime + nextReaction.get().getReaction().updateInterval()));
- reactions.wait(waitTime);
- } catch (InterruptedException ie1) {
- /* weâre looping! */
- }
- }
-
- /* re-start loop to check for new reactions. */
- continue;
- }
-
- /* run reaction. */
- logger.info(format("Running Query for %s...", reactionName));
- Query query = nextReaction.get().getReaction().query();
- net.pterodactylus.rhynodge.State state;
- try {
- logger.debug("Querying system...");
- state = query.state();
- if (state == null) {
- state = FailedState.INSTANCE;
- }
- logger.debug("System queried.");
- } catch (Throwable t1) {
- logger.warn("Querying system failed!", t1);
- state = new AbstractState(t1) {
- /* no further state. */
- };
- }
- logger.debug(format("State is %s.", state));
-
- /* convert states. */
- for (Filter filter : nextReaction.get().getReaction().filters()) {
- if (state.success()) {
- net.pterodactylus.rhynodge.State newState = filter.filter(state);
- //logger.debug(String.format("Old state is %s, new state is %s.", state, newState));
- state = newState;
- }
- }
- if (!state.success()) {
- state.setFailCount(lastStateFailCount + 1);
- }
- Optional lastSuccessfulState = stateManager.loadLastSuccessfulState(reactionName);
-
- /* merge states. */
- boolean triggerHit = false;
- Trigger trigger = nextReaction.get().getReaction().trigger();
- if (lastSuccessfulState.isPresent() && lastSuccessfulState.get().success() && state.success()) {
- net.pterodactylus.rhynodge.State newState = trigger.mergeStates(lastSuccessfulState.get(), state);
-
- /* save new state. */
- stateManager.saveState(reactionName, newState);
-
- triggerHit = trigger.triggers();
- } else {
- /* save first or error state. */
- stateManager.saveState(reactionName, state);
- }
-
- /* run action if trigger was hit. */
- logger.debug(format("Trigger was hit: %s.", triggerHit));
- if (triggerHit) {
- logger.info("Executing Action...");
- nextReaction.get().getReaction().action().execute(trigger.output(nextReaction.get().getReaction()));
- }
-
+ if (!scheduledFutures.containsKey(name)) {
+ return;
}
- }
-
- private Optional getNextReaction() {
- while (isRunning()) {
- synchronized (reactions) {
- if (reactions.isEmpty()) {
- logger.debug("Sleeping while no Reactions available.");
- try {
- reactions.wait();
- } catch (InterruptedException ie1) {
- /* ignore, weâre looping anyway. */
- }
- continue;
- }
- }
-
- /* find next reaction. */
- SortedMap> nextReactions = newTreeMap();
- synchronized (reactions) {
- for (Entry reactionEntry : reactions.entrySet()) {
- Optional state = stateManager.loadLastState(reactionEntry.getKey());
- long stateTime = state.isPresent() ? state.get().time() : 0;
- nextReactions.put(stateTime + reactionEntry.getValue().updateInterval(), Pair.of(reactionEntry.getKey(), reactionEntry.getValue()));
- }
- Pair keyReaction = nextReactions.get(nextReactions.firstKey());
- return of(new NextReaction(keyReaction.getKey(), keyReaction.getValue(), nextReactions.firstKey()));
- }
- }
- return absent();
- }
-
- private static class NextReaction {
-
- private final String key;
- private final Reaction reaction;
- private final long nextTime;
-
- private NextReaction(String key, Reaction reaction, long nextTime) {
- this.key = key;
- this.reaction = reaction;
- this.nextTime = nextTime;
- }
-
- public String getKey() {
- return key;
- }
-
- public Reaction getReaction() {
- return reaction;
- }
-
- public long getNextTime() {
- return nextTime;
- }
-
+ scheduledFutures.remove(name).cancel(true);
}
}