fix(folia): route async tasks through platform schedulers

Author: Xephi

authme-core/src/main/java/fr/xephi/authme/AuthMe.java

@@ -166,7 +166,8 @@ public void onEnable() {
 
         // Schedule clean up task
         CleanupTask cleanupTask = injector.getSingleton(CleanupTask.class);
-        cleanupTask.runTaskTimerAsynchronously(this, CLEANUP_INTERVAL, CLEANUP_INTERVAL);
+        injector.getSingleton(BukkitService.class)
+            .runTaskTimerAsynchronously(cleanupTask, CLEANUP_INTERVAL, CLEANUP_INTERVAL);
 
         // Do a backup on start
         backupService.doBackup(BackupService.BackupCause.START);

authme-core/src/main/java/fr/xephi/authme/initialization/OnStartupTasks.java

@@ -18,7 +18,6 @@
 import org.bstats.charts.SimplePie;
 import org.bukkit.Bukkit;
 import org.bukkit.entity.Player;
-import org.bukkit.scheduler.BukkitRunnable;
 
 import javax.inject.Inject;
 import java.util.List;
@@ -98,19 +97,16 @@ public void scheduleRecallEmailTask() {
         if (!settings.getProperty(RECALL_PLAYERS)) {
             return;
         }
-        bukkitService.runTaskTimerAsynchronously(new BukkitRunnable() {
-            @Override
-            public void run() {
-                List<String> loggedPlayersWithEmptyMail = dataSource.getLoggedPlayersWithEmptyMail();
-                bukkitService.runTask(() -> {
-                    for (String playerWithoutMail : loggedPlayersWithEmptyMail) {
-                        Player player = bukkitService.getPlayerExact(playerWithoutMail);
-                        if (player != null) {
-                            messages.send(player, MessageKey.ADD_EMAIL_MESSAGE);
-                        }
+        bukkitService.runTaskTimerAsynchronously(() -> {
+            List<String> loggedPlayersWithEmptyMail = dataSource.getLoggedPlayersWithEmptyMail();
+            bukkitService.runTask(() -> {
+                for (String playerWithoutMail : loggedPlayersWithEmptyMail) {
+                    Player player = bukkitService.getPlayerExact(playerWithoutMail);
+                    if (player != null) {
+                        messages.send(player, MessageKey.ADD_EMAIL_MESSAGE);
                     }
-                });
-            }
+                }
+            });
         }, 1, (long) TICKS_PER_MINUTE * settings.getProperty(EmailSettings.DELAY_RECALL));
     }
 }

authme-core/src/main/java/fr/xephi/authme/platform/AbstractSpigotPlatformAdapter.java

@@ -58,6 +58,22 @@ public void run() {
         return wrapTask(bukkitRunnable.runTaskTimer(plugin, delay, period));
     }
 
+    @Override
+    public CancellableTask runAsyncTask(AuthMe plugin, Runnable task) {
+        return wrapTask(Bukkit.getScheduler().runTaskAsynchronously(plugin, task));
+    }
+
+    @Override
+    public CancellableTask runAsyncTaskTimer(AuthMe plugin, Runnable task, long delay, long period) {
+        BukkitRunnable bukkitRunnable = new BukkitRunnable() {
+            @Override
+            public void run() {
+                task.run();
+            }
+        };
+        return wrapTask(bukkitRunnable.runTaskTimerAsynchronously(plugin, delay, period));
+    }
+
     @Override
     public void runOnGlobalThread(AuthMe plugin, Runnable task) {
         Bukkit.getScheduler().runTask(plugin, task);

authme-core/src/main/java/fr/xephi/authme/platform/SchedulingAdapter.java

@@ -54,6 +54,26 @@ public interface SchedulingAdapter {
      */
     CancellableTask runAtFixedRateOnEntityThread(AuthMe plugin, Entity entity, Runnable task, long delay, long period);
 
+    /**
+     * Runs a one-off background task that is not tied to a global or entity-owned thread.
+     *
+     * @param plugin the plugin scheduling the task
+     * @param task the task to run
+     * @return the scheduled task handle
+     */
+    CancellableTask runAsyncTask(AuthMe plugin, Runnable task);
+
+    /**
+     * Runs a repeating background task that is not tied to a global or entity-owned thread.
+     *
+     * @param plugin the plugin scheduling the task
+     * @param task the task to run
+     * @param delay the initial delay in ticks
+     * @param period the repeat period in ticks
+     * @return the scheduled task handle
+     */
+    CancellableTask runAsyncTaskTimer(AuthMe plugin, Runnable task, long delay, long period);
+
     /**
      * Runs a task on the platform's global execution context.
      *

authme-core/src/main/java/fr/xephi/authme/service/BukkitService.java

@@ -216,12 +216,12 @@ public void runTaskOptionallyAsync(Runnable task) {
      * Returns a task that will run asynchronously.
      *
      * @param task the task to be run
-     * @return a BukkitTask that contains the id number
+     * @return the scheduled task handle
      * @throws IllegalArgumentException if plugin is null
      * @throws IllegalArgumentException if task is null
      */
-    public BukkitTask runTaskAsynchronously(Runnable task) {
-        return Bukkit.getScheduler().runTaskAsynchronously(authMe, task);
+    public CancellableTask runTaskAsynchronously(Runnable task) {
+        return schedulingAdapter.runAsyncTask(authMe, task);
     }
 
     /**
@@ -235,12 +235,11 @@ public BukkitTask runTaskAsynchronously(Runnable task) {
      * @param delay the ticks to wait before running the task for the first
      *     time
      * @param period the ticks to wait between runs
-     * @return a BukkitTask that contains the id number
+     * @return the scheduled task handle
      * @throws IllegalArgumentException if task is null
-     * @throws IllegalStateException if this was already scheduled
      */
-    public BukkitTask runTaskTimerAsynchronously(BukkitRunnable task, long delay, long period) {
-        return task.runTaskTimerAsynchronously(authMe, delay, period);
+    public CancellableTask runTaskTimerAsynchronously(Runnable task, long delay, long period) {
+        return schedulingAdapter.runAsyncTaskTimer(authMe, task, delay, period);
     }
 
     /**

authme-core/src/main/java/fr/xephi/authme/task/CleanupTask.java

@@ -2,14 +2,13 @@
 
 import ch.jalu.injector.factory.SingletonStore;
 import fr.xephi.authme.initialization.HasCleanup;
-import org.bukkit.scheduler.BukkitRunnable;
 
 import javax.inject.Inject;
 
 /**
  * Task run periodically to invoke the cleanup task on services.
  */
-public class CleanupTask extends BukkitRunnable {
+public class CleanupTask implements Runnable {
 
     @Inject
     private SingletonStore<HasCleanup> hasCleanupStore;

authme-core/src/main/java/fr/xephi/authme/task/purge/PurgeService.java

@@ -4,6 +4,7 @@
 import fr.xephi.authme.datasource.DataSource;
 import fr.xephi.authme.output.ConsoleLoggerFactory;
 import fr.xephi.authme.permission.PermissionsManager;
+import fr.xephi.authme.service.CancellableTask;
 import fr.xephi.authme.service.BukkitService;
 import fr.xephi.authme.settings.Settings;
 import fr.xephi.authme.settings.properties.PurgeSettings;
@@ -99,7 +100,8 @@ public void purgePlayers(CommandSender sender, Set<String> names, OfflinePlayer[
 
         isPurging = true;
         PurgeTask purgeTask = new PurgeTask(this, permissionsManager, sender, names, players);
-        bukkitService.runTaskTimerAsynchronously(purgeTask, 0, 1);
+        CancellableTask taskHandle = bukkitService.runTaskTimerAsynchronously(purgeTask, 0, 1);
+        purgeTask.setTaskHandle(taskHandle);
     }
 
     /**

authme-core/src/main/java/fr/xephi/authme/task/purge/PurgeTask.java

@@ -9,14 +9,14 @@
 import org.bukkit.OfflinePlayer;
 import org.bukkit.command.CommandSender;
 import org.bukkit.entity.Player;
-import org.bukkit.scheduler.BukkitRunnable;
+import fr.xephi.authme.service.CancellableTask;
 
 import java.util.HashSet;
 import java.util.Locale;
 import java.util.Set;
 import java.util.UUID;
 
-class PurgeTask extends BukkitRunnable {
+class PurgeTask implements Runnable {
 
     //how many players we should check for each tick
     private static final int INTERVAL_CHECK = 5;
@@ -31,6 +31,8 @@ class PurgeTask extends BukkitRunnable {
     private final int totalPurgeCount;
 
     private int currentPage = 0;
+    private CancellableTask taskHandle;
+    private boolean cancellationRequested;
 
     /**
      * Constructor.
@@ -108,7 +110,7 @@ public void run() {
     }
 
     private void finish() {
-        cancel();
+        cancelTask();
 
         // Show a status message
         sendMessage(ChatColor.GREEN + "[AuthMe] Database has been purged successfully");
@@ -117,6 +119,16 @@ private void finish() {
         purgeService.setPurging(false);
     }
 
+    void setTaskHandle(CancellableTask taskHandle) {
+        synchronized (this) {
+            if (cancellationRequested) {
+                taskHandle.cancel();
+            } else {
+                this.taskHandle = taskHandle;
+            }
+        }
+    }
+
     private void sendMessage(String message) {
         if (sender == null) {
             Bukkit.getConsoleSender().sendMessage(message);
@@ -127,4 +139,14 @@ private void sendMessage(String message) {
             }
         }
     }
+
+    private void cancelTask() {
+        synchronized (this) {
+            if (taskHandle != null) {
+                taskHandle.cancel();
+            } else {
+                cancellationRequested = true;
+            }
+        }
+    }
 }

authme-core/src/test/java/fr/xephi/authme/service/BukkitServiceTest.java

@@ -217,7 +217,7 @@ public void shouldRunTaskInAsync() {
         // given
         Runnable task = mock(Runnable.class);
         BukkitService spy = Mockito.spy(bukkitService);
-        doReturn(null).when(spy).runTaskAsynchronously(task);
+        doReturn(mock(CancellableTask.class)).when(spy).runTaskAsynchronously(task);
 
         // when
         spy.runTaskOptionallyAsync(task);
@@ -247,35 +247,32 @@ public void shouldRunTaskDirectlyIfConfigured() {
     public void shouldRunTaskAsynchronously() {
         // given
         Runnable task = () -> {/* noop */};
-        BukkitTask bukkitTask = mock(BukkitTask.class);
-        given(scheduler.runTaskAsynchronously(authMe, task)).willReturn(bukkitTask);
+        CancellableTask cancellableTask = mock(CancellableTask.class);
+        given(schedulingAdapter.runAsyncTask(authMe, task)).willReturn(cancellableTask);
 
         // when
-        BukkitTask resultingTask = bukkitService.runTaskAsynchronously(task);
+        CancellableTask resultingTask = bukkitService.runTaskAsynchronously(task);
 
         // then
-        assertThat(resultingTask, equalTo(bukkitTask));
-        verify(scheduler, only()).runTaskAsynchronously(authMe, task);
+        assertThat(resultingTask, equalTo(cancellableTask));
+        verify(schedulingAdapter, only()).runAsyncTask(authMe, task);
     }
 
     @Test
     public void shouldRunTaskTimerAsynchronously() {
         // given
-        BukkitRunnable task = new BukkitRunnable() {
-            @Override
-            public void run() {
-            }
-        };
+        Runnable task = mock(Runnable.class);
         long delay = 20L;
         long period = 4000L;
-        BukkitTask bukkitTask = mock(BukkitTask.class);
-        given(task.runTaskTimerAsynchronously(authMe, delay, period)).willReturn(bukkitTask);
+        CancellableTask cancellableTask = mock(CancellableTask.class);
+        given(schedulingAdapter.runAsyncTaskTimer(authMe, task, delay, period)).willReturn(cancellableTask);
 
         // when
-        BukkitTask resultingTask = bukkitService.runTaskTimerAsynchronously(task, delay, period);
+        CancellableTask resultingTask = bukkitService.runTaskTimerAsynchronously(task, delay, period);
 
         // then
-        assertThat(resultingTask, equalTo(bukkitTask));
+        assertThat(resultingTask, equalTo(cancellableTask));
+        verify(schedulingAdapter, only()).runAsyncTaskTimer(authMe, task, delay, period);
     }
 
     @Test

authme-core/src/test/java/fr/xephi/authme/task/purge/PurgeServiceTest.java

@@ -68,7 +68,7 @@ public class PurgeServiceTest {
     @Captor
     private ArgumentCaptor<Long> longCaptor;
     @Captor
-    private ArgumentCaptor<PurgeTask> purgeTaskCaptor;
+    private ArgumentCaptor<Runnable> purgeTaskCaptor;
 
     @BeforeAll
     public static void initLogger() {
@@ -196,7 +196,7 @@ private void assertCorrectPurgeTimestamp(long timestamp, int configuredDays) {
 
     private void verifyScheduledPurgeTask(UUID senderUuid, Set<String> names) {
         verify(bukkitService).runTaskTimerAsynchronously(purgeTaskCaptor.capture(), eq(0L), eq(1L));
-        PurgeTask task = purgeTaskCaptor.getValue();
+        PurgeTask task = (PurgeTask) purgeTaskCaptor.getValue();
 
         Object senderInTask = ReflectionTestUtils.getFieldValue(PurgeTask.class, task, "sender");
         Set<String> namesInTask = ReflectionTestUtils.getFieldValue(PurgeTask.class, task, "toPurge");