andrew
/
SimplyScheduled
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Improved scheduling and test reliability.

This commit is contained in:
Andrew Lalis 2021-08-06 01:12:29 +02:00
parent 020d8e2258
commit 27dce051f7
9 changed files with 123 additions and 69 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

33
pom.xml
View File

@ -7,6 +7,33 @@
<groupId>nl.andrewlalis</groupId>
<artifactId>simply-scheduled</artifactId>
<version>1.0-SNAPSHOT</version>
<packaging>jar</packaging>
<name>Simply Scheduled</name>
<description>Lightweight task scheduling library.</description>
<url>https://github.com/andrewlalis/SimplyScheduled</url>
<licenses>
<license>
<name>Apache License, Version 2.0</name>
<url>https://www.apache.org/licenses/LICENSE-2.0.txt</url>
<distribution>repo</distribution>
</license>
</licenses>
<developers>
<developer>
<name>Andrew Lalis</name>
<email>andrewlalisofficial@gmail.com</email>
<timezone>CEST</timezone>
</developer>
</developers>
<scm>
<connection>scm:git:git://github.com/andrewlalis/SimplyScheduled.git</connection>
<developerConnection>scm:git:ssh://github.com:andrewlalis/SimplyScheduled.git</developerConnection>
<url>https://github.com/andrewlalis/SimplyScheduled</url>
</scm>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
@ -20,7 +47,7 @@
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-surefire-plugin</artifactId>
<version>2.22.2</version>
<version>3.0.0-M5</version>
</plugin>
</plugins>
</build>
@ -30,7 +57,7 @@
<dependency>
<groupId>org.junit</groupId>
<artifactId>junit-bom</artifactId>
<version>5.7.1</version>
<version>5.7.2</version>
<type>pom</type>
<scope>import</scope>
</dependency>
@ -42,7 +69,7 @@
<dependency>
<groupId>org.junit.jupiter</groupId>
<artifactId>junit-jupiter</artifactId>
<version>5.7.1</version>
<version>5.7.2</version>
<scope>test</scope>
</dependency>
</dependencies>

1
src/main/java/module-info.java
View File

@ -1,5 +1,6 @@
module simply_scheduled {
exports nl.andrewlalis.simply_scheduled;
exports nl.andrewlalis.simply_scheduled.schedule;
opens nl.andrewlalis.simply_scheduled;
}

19
src/main/java/nl/andrewlalis/simply_scheduled/BasicScheduler.java
View File

@ -8,6 +8,7 @@ import java.util.Optional;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.RejectedExecutionException;
/**
* A simple thread-based scheduler that sleeps until the next task, runs it
@ -47,9 +48,9 @@ public class BasicScheduler extends Thread implements Scheduler {
this.running = true;
while (this.running) {
try {
Task nextTask = this.tasks.take();
Instant now = this.clock.instant();
Optional<Instant> optionalNextExecution = nextTask.getSchedule().getNextExecutionTime(now);
final Task nextTask = this.tasks.take();
final Instant now = this.clock.instant();
final Optional<Instant> optionalNextExecution = nextTask.getSchedule().getNextExecutionTime(now);
if (optionalNextExecution.isEmpty()) {
continue; // Skip if the schedule doesn't have a next execution planned.
}
@ -57,10 +58,18 @@ public class BasicScheduler extends Thread implements Scheduler {
if (waitTime > 0) {
Thread.sleep(waitTime);
}
this.executorService.execute(nextTask.getRunnable());
try {
this.executorService.execute(nextTask.getRunnable());
} catch (RejectedExecutionException e) {
if (!this.executorService.isShutdown()) {
// Only show the stack trace if the executor service is not being shut down.
// We expect the service to reject executions if it is shutting down.
e.printStackTrace();
}
}
nextTask.getSchedule().markExecuted(this.clock.instant());
if (nextTask.getSchedule().isRepeating()) {
this.tasks.put(nextTask); // Put the task back in the queue.
this.tasks.put(nextTask);
}
} catch (InterruptedException e) {
this.setRunning(false);

5
src/main/java/nl/andrewlalis/simply_scheduled/Demo.java
View File

@ -2,7 +2,6 @@ package nl.andrewlalis.simply_scheduled;
import nl.andrewlalis.simply_scheduled.schedule.RepeatingSchedule;
import nl.andrewlalis.simply_scheduled.schedule.Schedule;
import nl.andrewlalis.simply_scheduled.schedule.Task;
import java.time.Instant;
import java.time.temporal.ChronoUnit;
@ -10,9 +9,9 @@ import java.time.temporal.ChronoUnit;
public class Demo {
public static void main(String[] args) {
Scheduler scheduler = new BasicScheduler();
Schedule schedule = new RepeatingSchedule(ChronoUnit.MILLIS, 250);
Schedule schedule = new RepeatingSchedule(ChronoUnit.SECONDS, 5, Instant.now().truncatedTo(ChronoUnit.SECONDS));
Runnable job = () -> System.out.println("Doing task: " + Instant.now().toString());
scheduler.addTask(Task.of(job, schedule));
scheduler.addTask(job, schedule);
scheduler.start();
}
}

17
src/main/java/nl/andrewlalis/simply_scheduled/Scheduler.java
View File

@ -1,5 +1,6 @@
package nl.andrewlalis.simply_scheduled;
import nl.andrewlalis.simply_scheduled.schedule.Schedule;
import nl.andrewlalis.simply_scheduled.schedule.Task;
/**
@ -14,6 +15,15 @@ public interface Scheduler {
*/
void addTask(Task task);
/**
* Adds a task to this scheduler.
* @param runnable The code to run.
* @param schedule The schedule that dictates when the code should run.
*/
default void addTask(Runnable runnable, Schedule schedule) {
addTask(new Task(runnable, schedule));
}
/**
* Starts the scheduler. A scheduler should only execute tasks once it has
* started, and it is up to the implementation to determine whether new
@ -27,4 +37,11 @@ public interface Scheduler {
* any currently-executing tasks are immediately shutdown.
*/
void stop(boolean force);
/**
* Stops the scheduler, and waits for any currently-executing tasks to finish.
*/
default void stop() {
stop(false);
}
}

32
src/main/java/nl/andrewlalis/simply_scheduled/schedule/MinutelySchedule.java
View File

@ -1,32 +0,0 @@
package nl.andrewlalis.simply_scheduled.schedule;
import java.time.Instant;
import java.time.ZoneId;
import java.time.ZonedDateTime;
import java.time.temporal.ChronoUnit;
import java.util.Optional;
public class MinutelySchedule implements Schedule {
private final int second;
private final ZoneId zoneId;
public MinutelySchedule(int second, ZoneId zoneId) {
this.second = second;
this.zoneId = zoneId;
}
public MinutelySchedule(int second) {
this(second, ZoneId.systemDefault());
}
@Override
public Optional<Instant> getNextExecutionTime(Instant referenceInstant) {
ZonedDateTime currentTime = referenceInstant.atZone(this.zoneId);
int currentSecond = currentTime.getSecond();
if (currentSecond >= this.second) {
return Optional.of(currentTime.plusMinutes(1).withSecond(this.second).truncatedTo(ChronoUnit.SECONDS).toInstant());
}
return Optional.of(currentTime.withSecond(this.second).truncatedTo(ChronoUnit.SECONDS).toInstant());
}
}

27
src/main/java/nl/andrewlalis/simply_scheduled/schedule/RepeatingSchedule.java
View File

@ -11,17 +11,29 @@ import java.util.Optional;
public class RepeatingSchedule implements Schedule {
private final ChronoUnit unit;
private final long multiple;
private Instant lastExecution;
private long elapsedIntervals = 0;
private final Instant start;
/**
* Constructs a new repeating schedule.
* @param unit The unit of time that the interval consists of.
* @param multiple The
* @param multiple The number of units of time that each interval consists of.
* @param start The starting point for this schedule.
*/
public RepeatingSchedule(ChronoUnit unit, long multiple) {
public RepeatingSchedule(ChronoUnit unit, long multiple, Instant start) {
this.unit = unit;
this.multiple = multiple;
this.lastExecution = null;
this.start = start;
}
/**
* Constructs a new repeating schedule, using {@link Instant#now()} as the
* starting point.
* @param unit The unit of time that the interval consists of.
* @param multiple The number of units of time that each interval consists of.
*/
public RepeatingSchedule(ChronoUnit unit, long multiple) {
this(unit, multiple, Instant.now());
}
/**
@ -32,14 +44,11 @@ public class RepeatingSchedule implements Schedule {
*/
@Override
public Optional<Instant> getNextExecutionTime(Instant referenceInstant) {
if (this.lastExecution == null) {
this.lastExecution = referenceInstant;
}
return Optional.of(this.lastExecution.plus(multiple, unit));
return Optional.of(this.start.plus(elapsedIntervals * multiple, unit));
}
@Override
public void markExecuted(Instant instant) {
this.lastExecution = instant;
this.elapsedIntervals++;
}
}

24
src/main/java/nl/andrewlalis/simply_scheduled/schedule/Task.java
View File

@ -14,20 +14,40 @@ public class Task implements Comparable<Task>{
private final Runnable runnable;
private final Schedule schedule;
/**
* Constructs a new task that will run the given runnable according to the
* given schedule. Allows for specifying a {@link Clock}, this is mostly
* useful for testing purposes.
* @param clock The clock to use for time-sensitive operations.
* @param runnable The code to run when the task is executed.
* @param schedule The schedule which determines when the task is executed.
*/
public Task(Clock clock, Runnable runnable, Schedule schedule) {
this.clock = clock;
this.runnable = runnable;
this.schedule = schedule;
}
public static Task of(Runnable runnable, Schedule schedule) {
return new Task(Clock.systemDefaultZone(), runnable, schedule);
/**
* Constructs a new task that will run the given runnable according to the
* given schedule.
* @param runnable The code to run when the task is executed.
* @param schedule The schedule which determines when the task is executed.
*/
public Task(Runnable runnable, Schedule schedule) {
this(Clock.systemDefaultZone(), runnable, schedule);
}
/**
* @return The runnable which will be executed when this task is scheduled.
*/
public Runnable getRunnable() {
return runnable;
}
/**
* @return The schedule for this task.
*/
public Schedule getSchedule() {
return schedule;
}

34
src/test/java/nl/andrewlalis/simply_scheduled/SchedulerTest.java → src/test/java/nl/andrewlalis/simply_scheduled/BasicSchedulerTest.java
View File

@ -1,6 +1,5 @@
package nl.andrewlalis.simply_scheduled;
import nl.andrewlalis.simply_scheduled.schedule.MinutelySchedule;
import nl.andrewlalis.simply_scheduled.schedule.RepeatingSchedule;
import nl.andrewlalis.simply_scheduled.schedule.Schedule;
import nl.andrewlalis.simply_scheduled.schedule.Task;
@ -8,6 +7,7 @@ import org.junit.jupiter.api.Test;
import java.time.Clock;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.time.temporal.ChronoUnit;
import java.util.concurrent.Executors;
@ -16,7 +16,11 @@ import java.util.concurrent.atomic.AtomicInteger;
import static org.junit.jupiter.api.Assertions.*;
public class SchedulerTest {
/**
* Tests the functionality of the {@link BasicScheduler} to reliably execute
* scheduled tasks.
*/
public class BasicSchedulerTest {
@Test
void testSchedule() {
@ -28,7 +32,7 @@ public class SchedulerTest {
flag.set(true);
System.out.println("\tExecuted task.");
};
Task task = new Task(clock, taskRunnable, new MinutelySchedule(3));
Task task = new Task(clock, taskRunnable, new RepeatingSchedule(ChronoUnit.SECONDS, 2, clock.instant().plusSeconds(1)));
scheduler.addTask(task);
scheduler.start();
System.out.println("Now: " + clock.instant().toString());
@ -45,24 +49,24 @@ public class SchedulerTest {
}
@Test
void testRepeatingSchedule() {
Scheduler scheduler = new BasicScheduler(Clock.systemUTC(), Executors.newWorkStealingPool());
Schedule schedule = new RepeatingSchedule(ChronoUnit.SECONDS, 1);
void testRepeatingSchedule() throws InterruptedException {
Scheduler scheduler = new BasicScheduler(Clock.systemUTC(), Executors.newFixedThreadPool(3));
Instant startInstant = Instant.now(Clock.systemUTC());
Schedule schedule = new RepeatingSchedule(ChronoUnit.SECONDS, 1, startInstant);
AtomicInteger value = new AtomicInteger(0);
Runnable taskRunnable = () -> {
value.set(value.get() + 1);
System.out.println("\tExecuted task.");
System.out.println("\tExecuted task at " + LocalDateTime.now());
};
scheduler.addTask(Task.of(taskRunnable, schedule));
scheduler.addTask(new Task(taskRunnable, schedule));
assertEquals(0, value.get());
scheduler.start();
System.out.println("Waiting 3.5 seconds for 3 iterations.");
try {
Thread.sleep(3500);
} catch (InterruptedException e) {
e.printStackTrace();
}
assertEquals(3, value.get());
Thread.sleep(3500);
// We expect the task to have run 4 times:
// at t=0, t=1, t=2, and t=3.
assertEquals(4, value.get());
scheduler.stop(true);
}
}