From cb0614019f8f15cb5d7210504ec7c3c57b81e9f9 Mon Sep 17 00:00:00 2001
From: Brianna <bri@songoda.com>
Date: Sat, 18 Apr 2020 05:10:54 -0400
Subject: [PATCH] Replace synced async threading with queued async threading.

---
 .../java/com/songoda/core/SongodaCore.java    |  2 -
 .../core/database/DataManagerAbstract.java    | 63 ++++++++-----------
 2 files changed, 26 insertions(+), 39 deletions(-)

diff --git a/Core/src/main/java/com/songoda/core/SongodaCore.java b/Core/src/main/java/com/songoda/core/SongodaCore.java
index 93df6f44..78657659 100644
--- a/Core/src/main/java/com/songoda/core/SongodaCore.java
+++ b/Core/src/main/java/com/songoda/core/SongodaCore.java
@@ -404,8 +404,6 @@ public class SongodaCore {
             if (pi != null) {
                 registeredPlugins.remove(pi);
             }
-            // Terminate all active threads
-            DataManagerAbstract.terminateAllThreads();
             if (event.getPlugin() == piggybackedPlugin) {
                 // uh-oh! Abandon ship!!
                 Bukkit.getServicesManager().unregisterAll(piggybackedPlugin);
diff --git a/Core/src/main/java/com/songoda/core/database/DataManagerAbstract.java b/Core/src/main/java/com/songoda/core/database/DataManagerAbstract.java
index 2cb906c9..540b68a6 100644
--- a/Core/src/main/java/com/songoda/core/database/DataManagerAbstract.java
+++ b/Core/src/main/java/com/songoda/core/database/DataManagerAbstract.java
@@ -8,17 +8,17 @@ import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.sql.Statement;
 import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
 import java.util.Map;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
 
 public class DataManagerAbstract {
 
     protected final DatabaseConnector databaseConnector;
     protected final Plugin plugin;
 
-    private static Map<String, ScheduledExecutorService> threads = new HashMap<>();
+    private static Map<String, LinkedList<Runnable>> queues = new HashMap<>();
 
     public DataManagerAbstract(DatabaseConnector databaseConnector, Plugin plugin) {
         this.databaseConnector = databaseConnector;
@@ -62,7 +62,6 @@ public class DataManagerAbstract {
 
     /**
      * Queue a task to be run asynchronously. <br>
-     * TODO: This needs to be separated from BukkitScheduler
      *
      * @param runnable task to run
      */
@@ -80,43 +79,33 @@ public class DataManagerAbstract {
     }
 
     /**
-     * Queue a task to be run synchronously on a new thread.
+     * Queue tasks to be ran asynchronously.
      *
-     * @param runnable  task to run on the next server tick
-     * @param threadKey the thread key to run on.
+     * @param runnable task to put into queue.
+     * @param queueKey the queue key to add the runnable to.
      */
-    public void sync(Runnable runnable, String threadKey) {
-        threads.computeIfAbsent(threadKey.toUpperCase(),
-                t -> Executors.newSingleThreadScheduledExecutor()).execute(runnable);
+    public void queueAsync(Runnable runnable, String queueKey) {
+        if (queueKey == null) return;
+        List<Runnable> queue = queues.computeIfAbsent(queueKey, t -> new LinkedList<>());
+        queue.add(runnable);
+        if (queue.size() == 1) runQueue(queueKey);
     }
 
-    /**
-     * Terminate thread once all tasks have been completed.
-     *
-     * @param threadKey the thread key to terminate.
-     */
-    public static void terminateThread(String threadKey) {
-        ScheduledExecutorService service = threads.get(threadKey);
-        if (service != null) {
-            threads.remove(threadKey);
-            try {
-                service.awaitTermination(0, TimeUnit.SECONDS);
-            } catch (InterruptedException e) {
-                e.printStackTrace();
-            }
-        }
+    private void runQueue(String queueKey) {
+        doQueue(queueKey, (s) -> {
+            if (!queues.get(queueKey).isEmpty())
+                runQueue(queueKey);
+        });
     }
 
-    /**
-     * Terminate all active threads.
-     */
-    public static void terminateAllThreads() {
-        for (ScheduledExecutorService service : threads.values()) {
-            try {
-                service.awaitTermination(0, TimeUnit.SECONDS);
-            } catch (InterruptedException e) {
-                e.printStackTrace();
-            }
-        }
+    private void doQueue(String queueKey, Consumer<Boolean> callback) {
+        Runnable runnable = queues.get(queueKey).getFirst();
+        async(() -> {
+            runnable.run();
+            sync(() -> {
+                queues.get(queueKey).remove(runnable);
+                callback.accept(true);
+            });
+        });
     }
 }