Make scheduler platform dependent

bukkit/src/main/java/com/github/games647/fastlogin/bukkit/BukkitScheduler.java

@@ -0,0 +1,27 @@
+package com.github.games647.fastlogin.bukkit;
+
+import com.github.games647.fastlogin.core.AsyncScheduler;
+
+import java.util.concurrent.Executor;
+import java.util.concurrent.ThreadFactory;
+
+import org.bukkit.Bukkit;
+import org.bukkit.plugin.Plugin;
+import org.slf4j.Logger;
+
+public class BukkitScheduler extends AsyncScheduler {
+
+    private final Plugin plugin;
+    private final Executor syncExecutor;
+
+    public BukkitScheduler(Plugin plugin, Logger logger, ThreadFactory threadFactory) {
+        super(logger, threadFactory);
+        this.plugin = plugin;
+
+        syncExecutor = r -> Bukkit.getScheduler().runTask(plugin, r);
+    }
+
+    public Executor getSyncExecutor() {
+        return syncExecutor;
+    }
+}

bukkit/src/main/java/com/github/games647/fastlogin/bukkit/FastLoginBukkit.java

@@ -41,17 +41,24 @@ public class FastLoginBukkit extends JavaPlugin implements PlatformPlugin<Comman
 
     //1 minutes should be enough as a timeout for bad internet connection (Server, Client and Mojang)
     private final ConcurrentMap<String, BukkitLoginSession> loginSession = CommonUtil.buildCache(1, -1);
-    private final Logger logger = CommonUtil.createLoggerFromJDK(getLogger());
     private final Map<UUID, PremiumStatus> premiumPlayers = new ConcurrentHashMap<>();
+    private final Logger logger;
 
     private boolean serverStarted;
     private boolean bungeeCord;
+    private final BukkitScheduler scheduler;
     private FastLoginCore<Player, CommandSender, FastLoginBukkit> core;
 
+    public FastLoginBukkit() {
+        this.logger = CommonUtil.createLoggerFromJDK(getLogger());
+        this.scheduler = new BukkitScheduler(this, logger, getThreadFactory());
+    }
+
     @Override
     public void onEnable() {
         core = new FastLoginCore<>(this);
         core.load();
+
         try {
             bungeeCord = Class.forName("org.spigotmc.SpigotConfig").getDeclaredField("bungee").getBoolean(null);
         } catch (ClassNotFoundException notFoundEx) {
@@ -72,7 +79,7 @@ public class FastLoginBukkit extends JavaPlugin implements PlatformPlugin<Comman
             setServerStarted();
 
             // check for incoming messages from the bungeecord version of this plugin
-            String forceChannel = new NamespaceKey(getName(), LoginActionMessage.FORCE_CHANNEL).getCombinedName();
+            String forceChannel = NamespaceKey.getCombined(getName(), LoginActionMessage.FORCE_CHANNEL);
             getServer().getMessenger().registerIncomingPluginChannel(this, forceChannel, new BungeeListener(this));
 
             // outgoing
@@ -195,6 +202,11 @@ public class FastLoginBukkit extends JavaPlugin implements PlatformPlugin<Comman
         return logger;
     }
 
+    @Override
+    public BukkitScheduler getScheduler() {
+        return scheduler;
+    }
+
     @Override
     public void sendMessage(CommandSender receiver, String message) {
         receiver.sendMessage(message);

bukkit/src/main/java/com/github/games647/fastlogin/bukkit/command/CrackedCommand.java

@@ -43,7 +43,7 @@ public class CrackedCommand extends ToggleCommand {
 
                 profile.setPremium(false);
                 profile.setId(null);
-                plugin.getCore().getAsyncScheduler().runAsync(() -> {
+                plugin.getScheduler().runAsync(() -> {
                     plugin.getCore().getStorage().save(profile);
                 });
             } else {
@@ -75,7 +75,7 @@ public class CrackedCommand extends ToggleCommand {
             plugin.getCore().sendLocaleMessage("remove-premium", sender);
 
             profile.setPremium(false);
-            plugin.getCore().getAsyncScheduler().runAsync(() -> {
+            plugin.getScheduler().runAsync(() -> {
                 plugin.getCore().getStorage().save(profile);
             });
         }

bukkit/src/main/java/com/github/games647/fastlogin/bukkit/command/PremiumCommand.java

@@ -55,7 +55,7 @@ public class PremiumCommand extends ToggleCommand {
             } else {
                 //todo: resolve uuid
                 profile.setPremium(true);
-                plugin.getCore().getAsyncScheduler().runAsync(() -> {
+                plugin.getScheduler().runAsync(() -> {
                     plugin.getCore().getStorage().save(profile);
                 });
 
@@ -84,7 +84,7 @@ public class PremiumCommand extends ToggleCommand {
         } else {
             //todo: resolve uuid
             profile.setPremium(true);
-            plugin.getCore().getAsyncScheduler().runAsync(() -> {
+            plugin.getScheduler().runAsync(() -> {
                 plugin.getCore().getStorage().save(profile);
             });
 

bukkit/src/main/java/com/github/games647/fastlogin/bukkit/listener/protocollib/ProtocolLibListener.java

@@ -66,7 +66,7 @@ public class ProtocolLibListener extends PacketAdapter {
 
         packetEvent.getAsyncMarker().incrementProcessingDelay();
         Runnable verifyTask = new VerifyResponseTask(plugin, packetEvent, sender, sharedSecret, keyPair);
-        plugin.getCore().getAsyncScheduler().runAsync(verifyTask);
+        plugin.getScheduler().runAsync(verifyTask);
     }
 
     private void onLogin(PacketEvent packetEvent, Player player) {
@@ -84,6 +84,6 @@ public class ProtocolLibListener extends PacketAdapter {
 
         packetEvent.getAsyncMarker().incrementProcessingDelay();
         Runnable nameCheckTask = new NameCheckTask(plugin, packetEvent, random, player, username, keyPair.getPublic());
-        plugin.getCore().getAsyncScheduler().runAsync(nameCheckTask);
+        plugin.getScheduler().runAsync(nameCheckTask);
     }
 }

bungee/src/main/java/com/github/games647/fastlogin/bungee/FastLoginBungee.java

@@ -3,6 +3,7 @@ package com.github.games647.fastlogin.bungee;
 import com.github.games647.fastlogin.bungee.hook.BungeeAuthHook;
 import com.github.games647.fastlogin.bungee.listener.ConnectListener;
 import com.github.games647.fastlogin.bungee.listener.PluginMessageListener;
+import com.github.games647.fastlogin.core.AsyncScheduler;
 import com.github.games647.fastlogin.core.CommonUtil;
 import com.github.games647.fastlogin.core.message.ChangePremiumMessage;
 import com.github.games647.fastlogin.core.message.ChannelMessage;
@@ -37,11 +38,13 @@ public class FastLoginBungee extends Plugin implements PlatformPlugin<CommandSen
     private final ConcurrentMap<PendingConnection, BungeeLoginSession> session = new MapMaker().weakKeys().makeMap();
 
     private FastLoginCore<ProxiedPlayer, CommandSender, FastLoginBungee> core;
+    private AsyncScheduler scheduler;
     private Logger logger;
 
     @Override
     public void onEnable() {
         logger = CommonUtil.createLoggerFromJDK(getLogger());
+        scheduler = new AsyncScheduler(logger, getThreadFactory());
 
         core = new FastLoginCore<>(this);
         core.load();
@@ -54,8 +57,8 @@ public class FastLoginBungee extends Plugin implements PlatformPlugin<CommandSen
         getProxy().getPluginManager().registerListener(this, new PluginMessageListener(this));
 
         //this is required to listen to incoming messages from the server
-        getProxy().registerChannel(new NamespaceKey(getName(), ChangePremiumMessage.CHANGE_CHANNEL).getCombinedName());
+        getProxy().registerChannel(NamespaceKey.getCombined(getName(), ChangePremiumMessage.CHANGE_CHANNEL));
-        getProxy().registerChannel(new NamespaceKey(getName(), SuccessMessage.SUCCESS_CHANNEL).getCombinedName());
+        getProxy().registerChannel(NamespaceKey.getCombined(getName(), SuccessMessage.SUCCESS_CHANNEL));
 
         registerHook();
     }
@@ -123,4 +126,9 @@ public class FastLoginBungee extends Plugin implements PlatformPlugin<CommandSen
                 .setThreadFactory(new GroupedThreadFactory(this, getName()))
                 .build();
     }
+
+    @Override
+    public AsyncScheduler getScheduler() {
+        return scheduler;
+    }
 }

bungee/src/main/java/com/github/games647/fastlogin/bungee/listener/ConnectListener.java

@@ -47,7 +47,7 @@ public class ConnectListener implements Listener {
 
         PendingConnection connection = preLoginEvent.getConnection();
         Runnable asyncPremiumCheck = new AsyncPremiumCheck(plugin, preLoginEvent, connection);
-        plugin.getCore().getAsyncScheduler().runAsync(asyncPremiumCheck);
+        plugin.getScheduler().runAsync(asyncPremiumCheck);
     }
 
     @EventHandler(priority = EventPriority.LOWEST)
@@ -100,7 +100,7 @@ public class ConnectListener implements Listener {
         Server server = serverConnectedEvent.getServer();
 
         Runnable loginTask = new ForceLoginTask(plugin.getCore(), player, server);
-        plugin.getCore().getAsyncScheduler().runAsync(loginTask);
+        plugin.getScheduler().runAsync(loginTask);
     }
 
     @EventHandler

bungee/src/main/java/com/github/games647/fastlogin/bungee/listener/PluginMessageListener.java

@@ -55,7 +55,7 @@ public class PluginMessageListener implements Listener {
         byte[] data = Arrays.copyOf(pluginMessageEvent.getData(), pluginMessageEvent.getData().length);
         ProxiedPlayer forPlayer = (ProxiedPlayer) pluginMessageEvent.getReceiver();
 
-        plugin.getCore().getAsyncScheduler().runAsync(() -> readMessage(forPlayer, channel, data));
+        plugin.getScheduler().runAsync(() -> readMessage(forPlayer, channel, data));
     }
 
     private void readMessage(ProxiedPlayer forPlayer, String channel, byte[] data) {
@@ -81,10 +81,10 @@ public class PluginMessageListener implements Listener {
 
                 core.getPendingConfirms().remove(forPlayer.getUniqueId());
                 Runnable task = new AsyncToggleMessage(core, forPlayer, playerName, true, isSourceInvoker);
-                plugin.getCore().getAsyncScheduler().runAsync(task);
+                plugin.getScheduler().runAsync(task);
             } else {
                 Runnable task = new AsyncToggleMessage(core, forPlayer, playerName, false, isSourceInvoker);
-                plugin.getCore().getAsyncScheduler().runAsync(task);
+                plugin.getScheduler().runAsync(task);
             }
         }
     }

core/src/main/java/com/github/games647/fastlogin/core/AsyncScheduler.java

@@ -1,10 +1,16 @@
 package com.github.games647.fastlogin.core;
 
+import com.google.common.util.concurrent.MoreExecutors;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 
+import org.slf4j.Logger;
+
 /**
  * This limits the number of threads that are used at maximum. Thread creation can be very heavy for the CPU and
  * context switching between threads too. However we need many threads for blocking HTTP and database calls.
@@ -14,27 +20,50 @@ import java.util.concurrent.TimeUnit;
  */
 public class AsyncScheduler {
 
+    private static final int MAX_CAPACITY = 1024;
+
+    //todo: single thread for delaying and scheduling tasks
+    private final Logger logger;
+
     // 30 threads are still too many - the optimal solution is to separate into processing and blocking threads
     // where processing threads could only be max number of cores while blocking threads could be minimized using
     // non-blocking I/O and a single event executor
-    private final ThreadPoolExecutor executorService;
+    private final ExecutorService processingPool;
 
-    public AsyncScheduler(ThreadFactory threadFactory) {
+    /*
-        executorService = new ThreadPoolExecutor(5, 30,
+    private final ExecutorService databaseExecutor = new ThreadPoolExecutor(1, 10,
+            0L, TimeUnit.MILLISECONDS,
+            new LinkedBlockingQueue<>(MAX_CAPACITY));
+     */
+
+    public AsyncScheduler(Logger logger, ThreadFactory threadFactory) {
+        this.logger = logger;
+        processingPool = new ThreadPoolExecutor(6, 32,
                 60L, TimeUnit.SECONDS,
-                new LinkedBlockingQueue<>(1024), threadFactory);
+                new LinkedBlockingQueue<>(MAX_CAPACITY), threadFactory);
     }
 
-    public void runAsync(Runnable task) {
+    /*
-        executorService.execute(task);
+    public <R> CompletableFuture<R> runDatabaseTask(Supplier<R> databaseTask) {
+        return CompletableFuture.supplyAsync(databaseTask, databaseExecutor)
+                .exceptionally(error -> {
+                    logger.warn("Error occurred on thread pool", error);
+                    return null;
+                })
+                // change context to the processing pool
+                .thenApplyAsync(r -> r, processingPool);
+    }
+     */
+
+    public CompletableFuture<Void> runAsync(Runnable task) {
+        return CompletableFuture.runAsync(task, processingPool).exceptionally(error -> {
+            logger.warn("Error occurred on thread pool", error);
+            return null;
+        });
     }
 
     public void shutdown() {
-        executorService.shutdown();
+        MoreExecutors.shutdownAndAwaitTermination(processingPool, 1, TimeUnit.MINUTES);
-        try {
+        //MoreExecutors.shutdownAndAwaitTermination(databaseExecutor, 1, TimeUnit.MINUTES);
-            executorService.awaitTermination(1, TimeUnit.MINUTES);
-        } catch (InterruptedException interruptEx) {
-            Thread.currentThread().interrupt();
-        }
     }
 }

core/src/main/java/com/github/games647/fastlogin/core/shared/FastLoginCore.java

@@ -4,7 +4,6 @@ import com.github.games647.craftapi.resolver.MojangResolver;
 import com.github.games647.craftapi.resolver.http.RotatingProxySelector;
 import com.github.games647.fastlogin.core.AuthStorage;
 import com.github.games647.fastlogin.core.CommonUtil;
-import com.github.games647.fastlogin.core.AsyncScheduler;
 import com.github.games647.fastlogin.core.hooks.AuthPlugin;
 import com.github.games647.fastlogin.core.hooks.DefaultPasswordGenerator;
 import com.github.games647.fastlogin.core.hooks.PasswordGenerator;
@@ -52,7 +51,6 @@ public class FastLoginCore<P extends C, C, T extends PlatformPlugin<C>> {
     private final T plugin;
 
     private final MojangResolver resolver = new MojangResolver();
-    private final AsyncScheduler scheduler;
 
     private Configuration config;
     private AuthStorage storage;
@@ -61,7 +59,6 @@ public class FastLoginCore<P extends C, C, T extends PlatformPlugin<C>> {
 
     public FastLoginCore(T plugin) {
         this.plugin = plugin;
-        this.scheduler = new AsyncScheduler(plugin.getThreadFactory());
     }
 
     public void load() {
@@ -242,13 +239,9 @@ public class FastLoginCore<P extends C, C, T extends PlatformPlugin<C>> {
         }
     }
 
-    public AsyncScheduler getAsyncScheduler() {
-        return scheduler;
-    }
-
     public void close() {
         plugin.getLog().info("Safely shutting down scheduler. This could take up to one minute.");
-        scheduler.shutdown();
+        plugin.getScheduler().shutdown();
 
         if (storage != null) {
             storage.close();

core/src/main/java/com/github/games647/fastlogin/core/shared/PlatformPlugin.java

@@ -1,5 +1,6 @@
 package com.github.games647.fastlogin.core.shared;
 
+import com.github.games647.fastlogin.core.AsyncScheduler;
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
 
 import java.nio.file.Path;
@@ -17,6 +18,8 @@ public interface PlatformPlugin<C> {
 
     void sendMessage(C receiver, String message);
 
+    AsyncScheduler getScheduler();
+
     default void sendMultiLineMessage(C receiver, String message) {
         for (String line : message.split("%nl%")) {
             sendMessage(receiver, line);