[edison-jobs]: add local scheduling

edison-jobs/src/main/java/de/otto/edison/jobs/configuration/JobsConfiguration.java

@@ -23,8 +23,10 @@
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.boot.task.ThreadPoolTaskSchedulerBuilder;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.scheduling.TaskScheduler;
 import org.springframework.scheduling.annotation.EnableAsync;
 import org.springframework.scheduling.annotation.EnableScheduling;
 
@@ -80,6 +82,12 @@ public Thread newThread(Runnable r) {
         });
     }
 
+    @Bean
+    @ConditionalOnMissingBean(TaskScheduler.class)
+    public TaskScheduler taskScheduler(ThreadPoolTaskSchedulerBuilder builder) {
+        return builder.build();
+    }
+
     @Bean
     @ConditionalOnMissingBean(JobMetaRepository.class)
     public JobMetaRepository jobMetaRepository() {

edison-jobs/src/main/java/de/otto/edison/jobs/service/LocalJobScheduler.java

@@ -0,0 +1,51 @@
+package de.otto.edison.jobs.service;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.boot.context.event.ApplicationReadyEvent;
+import org.springframework.context.event.EventListener;
+import org.springframework.scheduling.TaskScheduler;
+import org.springframework.scheduling.Trigger;
+import org.springframework.scheduling.TriggerContext;
+import org.springframework.scheduling.support.CronTrigger;
+import org.springframework.scheduling.support.PeriodicTrigger;
+import org.springframework.scheduling.support.SimpleTriggerContext;
+import org.springframework.stereotype.Service;
+
+import java.util.List;
+
+@Service
+@ConditionalOnProperty(value = "edison.jobs.localScheduling.enabled", havingValue = "true")
+public class LocalJobScheduler {
+    private static final Logger LOG = LoggerFactory.getLogger(LocalJobScheduler.class);
+
+    private final List<JobRunnable> jobRunnables;
+    private final JobService jobService;
+    private final TaskScheduler taskScheduler;
+
+    public LocalJobScheduler(List<JobRunnable> jobRunnables, JobService jobService, TaskScheduler taskScheduler) {
+        this.jobRunnables = jobRunnables;
+        this.jobService = jobService;
+        this.taskScheduler = taskScheduler;
+    }
+
+    @EventListener(ApplicationReadyEvent.class)
+    public void schedule() {
+        TriggerContext dummyTriggerContext = new SimpleTriggerContext();
+        jobRunnables.stream()
+                .map(JobRunnable::getJobDefinition)
+                .filter(jobDefinition -> jobDefinition.cron().isPresent() || jobDefinition.fixedDelay().isPresent())
+                .forEach(jobDefinition -> {
+                    Trigger trigger;
+                    if (jobDefinition.cron().isPresent()) {
+                        trigger = new CronTrigger(jobDefinition.cron().get());
+                    } else {
+                        trigger = new PeriodicTrigger(jobDefinition.fixedDelay().get());
+                    }
+
+                    taskScheduler.schedule(() -> jobService.startAsyncJob(jobDefinition.jobType()), trigger);
+                    LOG.info("Scheduled {} for {}", jobDefinition.jobType(), trigger.nextExecution(dummyTriggerContext));
+                });
+    }
+}

edison-jobs/src/test/java/de/otto/edison/jobs/service/LocalJobSchedulerTest.java

@@ -0,0 +1,96 @@
+package de.otto.edison.jobs.service;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mock;
+import org.springframework.scheduling.TaskScheduler;
+import org.springframework.scheduling.Trigger;
+import org.springframework.scheduling.support.CronTrigger;
+import org.springframework.scheduling.support.PeriodicTrigger;
+
+import java.time.Duration;
+import java.util.List;
+import java.util.Optional;
+
+import static de.otto.edison.jobs.definition.DefaultJobDefinition.*;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.*;
+import static org.mockito.MockitoAnnotations.openMocks;
+
+class LocalJobSchedulerTest {
+
+    @Mock
+    private JobRunnable fixedDelayJobRunnable;
+    @Mock
+    private JobRunnable manualJobRunnable;
+    @Mock
+    private JobRunnable cronJobRunnable;
+
+    @Mock
+    private JobService jobService;
+    @Mock
+    private TaskScheduler taskScheduler;
+
+    @BeforeEach
+    public void setUp() {
+        openMocks(this);
+        when(fixedDelayJobRunnable.getJobDefinition()).thenReturn(
+                fixedDelayJobDefinition("FIXED", "", "", Duration.ofSeconds(2), 0, Optional.empty())
+        );
+        when(manualJobRunnable.getJobDefinition()).thenReturn(
+                manuallyTriggerableJobDefinition("MANUAL", "", "", 0, Optional.empty())
+        );
+        when(cronJobRunnable.getJobDefinition()).thenReturn(
+                cronJobDefinition("CRON", "", "", "0 0 * * * *", 0, Optional.empty())
+        );
+    }
+
+    @Test
+    public void shouldScheduleRunnable() {
+        // given
+        LocalJobScheduler localJobScheduler = new LocalJobScheduler(List.of(fixedDelayJobRunnable, cronJobRunnable), jobService, taskScheduler);
+
+        // when
+        localJobScheduler.schedule();
+
+        // then
+        ArgumentCaptor<Trigger> triggerCaptor = ArgumentCaptor.forClass(Trigger.class);
+        verify(taskScheduler, times(2)).schedule(any(), triggerCaptor.capture());
+
+        assertTrue(triggerCaptor.getAllValues().stream().anyMatch(trigger -> trigger instanceof CronTrigger));
+        assertTrue(triggerCaptor.getAllValues().stream().anyMatch(trigger -> trigger instanceof PeriodicTrigger));
+    }
+
+    @Test
+    public void shouldStartJobFromRunnable() {
+        // given
+        doAnswer(invocation -> {
+            Runnable runnable = invocation.getArgument(0);
+            runnable.run();
+            return null;
+        }).when(taskScheduler).schedule(any(), (Trigger) any());
+        LocalJobScheduler localJobScheduler = new LocalJobScheduler(List.of(fixedDelayJobRunnable, cronJobRunnable), jobService, taskScheduler);
+
+        // when
+        localJobScheduler.schedule();
+
+        // then
+        ArgumentCaptor<String> stringCaptor = ArgumentCaptor.forClass(String.class);
+        verify(jobService, times(2)).startAsyncJob(stringCaptor.capture());
+        assertTrue(stringCaptor.getAllValues().contains("FIXED"));
+        assertTrue(stringCaptor.getAllValues().contains("CRON"));
+    }
+
+    @Test
+    public void shouldFilterManuallyTriggeredJobs() {
+        // given
+        LocalJobScheduler localJobScheduler = new LocalJobScheduler(List.of(fixedDelayJobRunnable, cronJobRunnable, manualJobRunnable), jobService, taskScheduler);
+
+        // when
+        localJobScheduler.schedule();
+
+        // then
+        verify(taskScheduler, times(2)).schedule(any(), (Trigger) any());
+    }
+}