Implement support for instant data propagation with Redis

common/src/main/java/me/lucko/luckperms/common/LuckPermsPlugin.java

@@ -35,6 +35,7 @@ import me.lucko.luckperms.common.contexts.ContextManager;
 import me.lucko.luckperms.common.core.UuidCache;
 import me.lucko.luckperms.common.data.Importer;
 import me.lucko.luckperms.common.groups.GroupManager;
+import me.lucko.luckperms.common.messaging.RedisMessaging;
 import me.lucko.luckperms.common.storage.Datastore;
 import me.lucko.luckperms.common.tracks.TrackManager;
 import me.lucko.luckperms.common.users.UserManager;
@@ -60,6 +61,7 @@ public interface LuckPermsPlugin {
     TrackManager getTrackManager();
     LPConfiguration getConfiguration();
     Datastore getDatastore();
+    RedisMessaging getRedisMessaging();
     Logger getLog();
     UuidCache getUuidCache();
     ApiProvider getApiProvider();

common/src/main/java/me/lucko/luckperms/common/commands/CommandManager.java

@@ -59,6 +59,7 @@ public class CommandManager {
             .add(new TrackMainCommand())
             .add(new LogMainCommand())
             .add(new SyncCommand())
+            .add(new NetworkSyncCommand())
             .add(new InfoCommand())
             .add(new DebugCommand())
             .add(new ImportCommand())

common/src/main/java/me/lucko/luckperms/common/commands/misc/NetworkSyncCommand.java

@@ -0,0 +1,54 @@
+/*
+ * Copyright (c) 2016 Lucko (Luck) <luck@lucko.me>
+ *
+ *  Permission is hereby granted, free of charge, to any person obtaining a copy
+ *  of this software and associated documentation files (the "Software"), to deal
+ *  in the Software without restriction, including without limitation the rights
+ *  to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ *  copies of the Software, and to permit persons to whom the Software is
+ *  furnished to do so, subject to the following conditions:
+ *
+ *  The above copyright notice and this permission notice shall be included in all
+ *  copies or substantial portions of the Software.
+ *
+ *  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ *  IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ *  FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ *  AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ *  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ *  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ *  SOFTWARE.
+ */
+
+package me.lucko.luckperms.common.commands.misc;
+
+import me.lucko.luckperms.common.LuckPermsPlugin;
+import me.lucko.luckperms.common.commands.CommandResult;
+import me.lucko.luckperms.common.commands.Sender;
+import me.lucko.luckperms.common.commands.SingleMainCommand;
+import me.lucko.luckperms.common.constants.Message;
+import me.lucko.luckperms.common.constants.Permission;
+
+import java.util.List;
+
+public class NetworkSyncCommand extends SingleMainCommand {
+    public NetworkSyncCommand() {
+        super("NetworkSync", "/%s networksync", 0, Permission.SYNC);
+    }
+
+    @Override
+    protected CommandResult execute(LuckPermsPlugin plugin, Sender sender, List<String> args, String label) {
+        Message.UPDATE_TASK_REQUEST.send(sender);
+        plugin.getUpdateTaskBuffer().request().getUnchecked();
+        Message.UPDATE_TASK_COMPLETE_NETWORK.send(sender);
+
+        if (plugin.getRedisMessaging() != null) {
+            plugin.getRedisMessaging().pushUpdate();
+            Message.UPDATE_TASK_PUSH_SUCCESS.send(sender);
+        } else {
+            Message.UPDATE_TASK_PUSH_FAILURE.send(sender);
+        }
+
+        return CommandResult.SUCCESS;
+    }
+}

common/src/main/java/me/lucko/luckperms/common/config/AbstractConfiguration.java

@@ -74,6 +74,9 @@ public abstract class AbstractConfiguration<T extends LuckPermsPlugin> implement
     private String storageMethod;
     private boolean splitStorage;
     private Map<String, String> splitStorageOptions;
+    private boolean redisEnabled;
+    private String redisAddress;
+    private String redisPassword;
 
     public AbstractConfiguration(T plugin, String defaultServerName, boolean defaultIncludeGlobal, String defaultStorage) {
         this.plugin = plugin;
@@ -142,6 +145,10 @@ public abstract class AbstractConfiguration<T extends LuckPermsPlugin> implement
                 .put("uuid", getString("split-storage.methods.uuid", defaultStorage))
                 .put("log", getString("split-storage.methods.log", defaultStorage))
                 .build();
+
+        redisEnabled = getBoolean("redis.enabled", false);
+        redisAddress = getString("redis.address", null);
+        redisPassword = getString("redis.password", "");
         
         if (Patterns.NON_ALPHA_NUMERIC.matcher(getServer()).find()) {
             plugin.getLog().severe("Server name defined in config.yml contains invalid characters. Server names can " +

common/src/main/java/me/lucko/luckperms/common/config/LPConfiguration.java

@@ -92,4 +92,10 @@ public interface LPConfiguration {
 
     Map<String, String> getSplitStorageOptions();
 
+    boolean isRedisEnabled();
+
+    String getRedisAddress();
+
+    String getRedisPassword();
+
 }

common/src/main/java/me/lucko/luckperms/common/constants/Message.java

@@ -110,6 +110,9 @@ public enum Message {
 
     UPDATE_TASK_REQUEST("&bUpdate task scheduled.", true),
     UPDATE_TASK_COMPLETE("&aUpdate task finished.", true),
+    UPDATE_TASK_COMPLETE_NETWORK("&aUpdate task finished. Now attempting to push to other servers.", true),
+    UPDATE_TASK_PUSH_SUCCESS("&aOther servers were notified successfully.", true),
+    UPDATE_TASK_PUSH_FAILURE("&cError whilst pushing changes to other servers. Is Redis enabled?", true),
     INFO(
             "{PREFIX}&2Running &bLuckPerms v{0}&2 by &bLuck&2." + "\n" +
             "{PREFIX}&f-> &3Platform: &f{1}" + "\n" +

common/src/main/java/me/lucko/luckperms/common/messaging/RedisMessaging.java

@@ -0,0 +1,124 @@
+/*
+ * Copyright (c) 2016 Lucko (Luck) <luck@lucko.me>
+ *
+ *  Permission is hereby granted, free of charge, to any person obtaining a copy
+ *  of this software and associated documentation files (the "Software"), to deal
+ *  in the Software without restriction, including without limitation the rights
+ *  to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ *  copies of the Software, and to permit persons to whom the Software is
+ *  furnished to do so, subject to the following conditions:
+ *
+ *  The above copyright notice and this permission notice shall be included in all
+ *  copies or substantial portions of the Software.
+ *
+ *  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ *  IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ *  FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ *  AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ *  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ *  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ *  SOFTWARE.
+ */
+
+package me.lucko.luckperms.common.messaging;
+
+import lombok.RequiredArgsConstructor;
+import me.lucko.luckperms.common.LuckPermsPlugin;
+import redis.clients.jedis.Jedis;
+import redis.clients.jedis.JedisPool;
+import redis.clients.jedis.JedisPoolConfig;
+import redis.clients.jedis.JedisPubSub;
+
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.UUID;
+
+/**
+ * Uses Redis to push/receive changes to/from other servers
+ */
+@RequiredArgsConstructor
+public class RedisMessaging {
+    private static final String CHANNEL = "luckperms";
+
+    private final LuckPermsPlugin plugin;
+    private JedisPool jedisPool;
+    private LPSub sub;
+
+    public void init(String address, String password) {
+        String host = address.substring(0, address.indexOf(':'));
+        int port = Integer.parseInt(address.substring(address.indexOf(":") + 1));
+
+        if (password.equals("")) {
+            jedisPool = new JedisPool(new JedisPoolConfig(), host, port);
+        } else {
+            jedisPool = new JedisPool(new JedisPoolConfig(), host, port, 0, password);
+        }
+
+        plugin.doAsync(() -> {
+            sub = new LPSub(plugin);
+            try (Jedis jedis = jedisPool.getResource()) {
+                jedis.subscribe(sub, CHANNEL);
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+        });
+    }
+
+    public void shutdown() {
+        sub.unsubscribe();
+        jedisPool.destroy();
+    }
+
+    public void pushUpdate() {
+        plugin.doAsync(() -> {
+            UUID id = sub.generateId();
+            plugin.getLog().info("[Redis Messaging] Sending redis ping with id: " + id.toString());
+            try (Jedis jedis = jedisPool.getResource()) {
+                jedis.publish(CHANNEL, "update:" + id.toString());
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+        });
+    }
+
+    @RequiredArgsConstructor
+    private static class LPSub extends JedisPubSub {
+        private final LuckPermsPlugin plugin;
+        private final Set<UUID> receivedMsgs = Collections.synchronizedSet(new HashSet<>());
+
+        public UUID generateId() {
+            UUID uuid = UUID.randomUUID();
+            receivedMsgs.add(uuid);
+            return uuid;
+        }
+
+        @Override
+        public void onMessage(String channel, String msg) {
+            if (!channel.equals(CHANNEL)) {
+                return;
+            }
+
+            if (!msg.startsWith("update:")) {
+                return;
+            }
+
+            String requestId = msg.substring("update:".length());
+            UUID uuid;
+            try {
+                uuid = UUID.fromString(requestId);
+            } catch (IllegalArgumentException e) {
+                e.printStackTrace();
+                return;
+            }
+
+            if (!receivedMsgs.add(uuid)) {
+                return;
+            }
+
+            plugin.getLog().info("[Redis Messaging] Received update ping with id: " + uuid.toString());
+            plugin.getUpdateTaskBuffer().request();
+        }
+    }
+
+}