Add cleanup process which triggers every 5 minutes to remove state ma…

…chines in a completed state for longer than 5 minutes

- Add state machine listener to set a timestamp when it lands on the FINISHED state
- Add scheduled task to cleanup any machines which have a completed-timestamp older than 5 mins

src/main/java/ch/uzh/ifi/access/student/config/SchedulingConfig.java

@@ -0,0 +1,34 @@
+package ch.uzh.ifi.access.student.config;
+
+import ch.uzh.ifi.access.student.evaluation.process.EvalMachineRepoService;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.scheduling.annotation.EnableScheduling;
+import org.springframework.scheduling.annotation.Scheduled;
+
+import java.time.Instant;
+import java.time.temporal.ChronoUnit;
+
+@Configuration
+@EnableScheduling
+public class SchedulingConfig {
+
+    private static Logger logger = LoggerFactory.getLogger(SchedulingConfig.class);
+
+    private static final long FIXED_DELAY_IN_MINUTES = 5;
+
+    private EvalMachineRepoService machineRepository;
+
+    public SchedulingConfig(EvalMachineRepoService machineRepository) {
+        this.machineRepository = machineRepository;
+    }
+
+    @Scheduled(fixedDelay = FIXED_DELAY_IN_MINUTES * 60000)
+    public void cleanUpRepo() {
+        Instant threshold = Instant.now().minus(5, ChronoUnit.MINUTES);
+        logger.debug("Starting state machine cleanup. Repository size {}, removing machine older than {}", machineRepository.count(), threshold);
+        machineRepository.removeMachinesOlderThan(threshold);
+        logger.debug("Completed cleanup. Repository size {}", machineRepository.count());
+    }
+}

src/main/java/ch/uzh/ifi/access/student/evaluation/process/EvalMachineFactory.java

@@ -16,6 +16,7 @@ public class EvalMachineFactory {
     public static final String EXTENDED_VAR_SUBMISSION_ID = "submissionId";
     public static final String EXTENDED_VAR_NEXT_STEP = "nextStep";
     public static final String EXTENDED_VAR_NEXT_STEP_DELAY = "nextStepDelay";
+    public static final String EXTENDED_VAR_COMPLETION_TIME = "completionTime";
 
     public static StateMachine<EvalMachine.States, EvalMachine.Events> initSMForSubmission(String submissionId) throws Exception {
 
@@ -57,6 +58,7 @@ public static StateMachine<EvalMachine.States, EvalMachine.Events> initSMForSubm
 
         StateMachine<EvalMachine.States, EvalMachine.Events> machine = builder.build();
         machine.getExtendedState().getVariables().put(EXTENDED_VAR_SUBMISSION_ID, submissionId);
+        machine.addStateListener(new StateMachineEventListener(machine));
         return machine;
     }
 

src/main/java/ch/uzh/ifi/access/student/evaluation/process/EvalMachineRepoService.java

@@ -3,6 +3,7 @@
 import org.springframework.statemachine.StateMachine;
 import org.springframework.stereotype.Service;
 
+import java.time.Instant;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 
@@ -27,4 +28,17 @@ public void store(String key, StateMachine machine) {
         machines.put(key, machine);
     }
 
+    public long count() {
+        return machines.size();
+    }
+
+    public void removeMachinesOlderThan(Instant threshold) {
+        for (String machineKey : machines.keySet()) {
+            StateMachine machine = machines.get(machineKey);
+            Instant completionTime = (Instant) machine.getExtendedState().getVariables().get(EvalMachineFactory.EXTENDED_VAR_COMPLETION_TIME);
+            if (completionTime.isBefore(threshold)) {
+                machines.remove(machineKey);
+            }
+        }
+    }
 }

src/main/java/ch/uzh/ifi/access/student/evaluation/process/StateMachineEventListener.java

@@ -0,0 +1,24 @@
+package ch.uzh.ifi.access.student.evaluation.process;
+
+import org.springframework.statemachine.StateMachine;
+import org.springframework.statemachine.listener.StateMachineListenerAdapter;
+import org.springframework.statemachine.state.State;
+
+import java.time.Instant;
+
+public class StateMachineEventListener
+        extends StateMachineListenerAdapter<EvalMachine.States, EvalMachine.Events> {
+
+    private StateMachine<EvalMachine.States, EvalMachine.Events> machine;
+
+    public StateMachineEventListener(StateMachine<EvalMachine.States, EvalMachine.Events> machine) {
+        this.machine = machine;
+    }
+
+    @Override
+    public void stateEntered(State<EvalMachine.States, EvalMachine.Events> state) {
+        if (EvalMachine.States.FINISHED.equals(state.getId())) {
+            machine.getExtendedState().getVariables().put(EvalMachineFactory.EXTENDED_VAR_COMPLETION_TIME, Instant.now());
+        }
+    }
+}

src/test/java/ch/uzh/ifi/access/student/evaluation/process/EvalMachineRepoServiceTest.java

@@ -0,0 +1,66 @@
+package ch.uzh.ifi.access.student.evaluation.process;
+
+import org.assertj.core.api.Assertions;
+import org.junit.Test;
+import org.springframework.statemachine.StateMachine;
+
+import java.time.Instant;
+import java.time.temporal.ChronoUnit;
+import java.util.UUID;
+
+public class EvalMachineRepoServiceTest {
+
+    @Test
+    public void cleanUpNoMachines() {
+        EvalMachineRepoService repo = new EvalMachineRepoService();
+        repo.removeMachinesOlderThan(Instant.now());
+    }
+
+    @Test
+    public void cleanUp() throws Exception {
+        String id1 = UUID.randomUUID().toString();
+        String id2 = UUID.randomUUID().toString();
+        StateMachine<EvalMachine.States, EvalMachine.Events> m1 = EvalMachineFactory.initSMForSubmission("123");
+        StateMachine<EvalMachine.States, EvalMachine.Events> m2 = EvalMachineFactory.initSMForSubmission("345");
+
+        m1.getExtendedState().getVariables().put(EvalMachineFactory.EXTENDED_VAR_COMPLETION_TIME, Instant.now().minus(30, ChronoUnit.MINUTES));
+        m2.getExtendedState().getVariables().put(EvalMachineFactory.EXTENDED_VAR_COMPLETION_TIME, Instant.now().minus(1, ChronoUnit.MINUTES));
+
+        EvalMachineRepoService repo = new EvalMachineRepoService();
+        repo.store(id1, m1);
+        repo.store(id2, m2);
+
+        Assertions.assertThat(repo.get(id1)).isNotNull();
+        Assertions.assertThat(repo.get(id2)).isNotNull();
+
+        Instant fiveMinutesAgo = Instant.now().minus(5, ChronoUnit.MINUTES);
+        repo.removeMachinesOlderThan(fiveMinutesAgo);
+
+        Assertions.assertThat(repo.get(id1)).isNull();
+        Assertions.assertThat(repo.get(id2)).isNotNull();
+    }
+
+    @Test
+    public void noMachinesToClean() throws Exception {
+        String id1 = UUID.randomUUID().toString();
+        String id2 = UUID.randomUUID().toString();
+        StateMachine<EvalMachine.States, EvalMachine.Events> m1 = EvalMachineFactory.initSMForSubmission("123");
+        StateMachine<EvalMachine.States, EvalMachine.Events> m2 = EvalMachineFactory.initSMForSubmission("345");
+
+        m1.getExtendedState().getVariables().put(EvalMachineFactory.EXTENDED_VAR_COMPLETION_TIME, Instant.now().minus(1, ChronoUnit.MINUTES));
+        m2.getExtendedState().getVariables().put(EvalMachineFactory.EXTENDED_VAR_COMPLETION_TIME, Instant.now().minus(1, ChronoUnit.MINUTES));
+
+        EvalMachineRepoService repo = new EvalMachineRepoService();
+        repo.store(id1, m1);
+        repo.store(id2, m2);
+
+        Assertions.assertThat(repo.get(id1)).isNotNull();
+        Assertions.assertThat(repo.get(id2)).isNotNull();
+
+        Instant fiveMinutesAgo = Instant.now().minus(5, ChronoUnit.MINUTES);
+        repo.removeMachinesOlderThan(fiveMinutesAgo);
+
+        Assertions.assertThat(repo.get(id1)).isNotNull();
+        Assertions.assertThat(repo.get(id2)).isNotNull();
+    }
+}