Implement BungeeCord & LilyPad messaging services - closes #142

This commit is contained in:
Luck 2017-01-22 21:46:22 +00:00
parent 0f8c334de8
commit 327c8b83be
No known key found for this signature in database
GPG Key ID: EFA9B3EC5FD90F8B
19 changed files with 628 additions and 109 deletions

View File

@ -108,6 +108,13 @@
<version>${project.version}</version> <version>${project.version}</version>
<scope>compile</scope> <scope>compile</scope>
</dependency> </dependency>
<!-- LilyPad - used as a messaging service -->
<dependency>
<groupId>lilypad.client.connect</groupId>
<artifactId>api</artifactId>
<version>0.0.1-SNAPSHOT</version>
<scope>provided</scope>
</dependency>
<!-- HikariCP - needed as source --> <!-- HikariCP - needed as source -->
<dependency> <dependency>
<groupId>com.zaxxer</groupId> <groupId>com.zaxxer</groupId>

View File

@ -33,6 +33,8 @@ import me.lucko.luckperms.api.PlatformType;
import me.lucko.luckperms.api.context.ContextSet; import me.lucko.luckperms.api.context.ContextSet;
import me.lucko.luckperms.api.context.MutableContextSet; import me.lucko.luckperms.api.context.MutableContextSet;
import me.lucko.luckperms.bukkit.inject.Injector; import me.lucko.luckperms.bukkit.inject.Injector;
import me.lucko.luckperms.bukkit.messaging.BungeeMessagingService;
import me.lucko.luckperms.bukkit.messaging.LilyPadMessagingService;
import me.lucko.luckperms.bukkit.model.ChildPermissionProvider; import me.lucko.luckperms.bukkit.model.ChildPermissionProvider;
import me.lucko.luckperms.bukkit.model.DefaultsProvider; import me.lucko.luckperms.bukkit.model.DefaultsProvider;
import me.lucko.luckperms.bukkit.model.LPPermissible; import me.lucko.luckperms.bukkit.model.LPPermissible;
@ -61,6 +63,7 @@ import me.lucko.luckperms.common.managers.UserManager;
import me.lucko.luckperms.common.managers.impl.GenericGroupManager; import me.lucko.luckperms.common.managers.impl.GenericGroupManager;
import me.lucko.luckperms.common.managers.impl.GenericTrackManager; import me.lucko.luckperms.common.managers.impl.GenericTrackManager;
import me.lucko.luckperms.common.managers.impl.GenericUserManager; import me.lucko.luckperms.common.managers.impl.GenericUserManager;
import me.lucko.luckperms.common.messaging.AbstractMessagingService;
import me.lucko.luckperms.common.messaging.RedisMessaging; import me.lucko.luckperms.common.messaging.RedisMessaging;
import me.lucko.luckperms.common.storage.Storage; import me.lucko.luckperms.common.storage.Storage;
import me.lucko.luckperms.common.storage.StorageFactory; import me.lucko.luckperms.common.storage.StorageFactory;
@ -114,7 +117,7 @@ public class LPBukkitPlugin extends JavaPlugin implements LuckPermsPlugin {
private GroupManager groupManager; private GroupManager groupManager;
private TrackManager trackManager; private TrackManager trackManager;
private Storage storage; private Storage storage;
private RedisMessaging redisMessaging = null; private AbstractMessagingService messagingService = null;
private UuidCache uuidCache; private UuidCache uuidCache;
private BukkitListener listener; private BukkitListener listener;
private ApiProvider apiProvider; private ApiProvider apiProvider;
@ -197,17 +200,40 @@ public class LPBukkitPlugin extends JavaPlugin implements LuckPermsPlugin {
// initialise datastore // initialise datastore
storage = StorageFactory.getInstance(this, StorageType.H2); storage = StorageFactory.getInstance(this, StorageType.H2);
// initialise redis // initialise messaging
if (getConfiguration().get(ConfigKeys.REDIS_ENABLED)) { String messagingType = getConfiguration().get(ConfigKeys.MESSAGING_SERVICE).toLowerCase();
if (messagingType.equals("redis")) {
getLog().info("Loading redis..."); getLog().info("Loading redis...");
redisMessaging = new RedisMessaging(this); if (getConfiguration().get(ConfigKeys.REDIS_ENABLED)) {
try { RedisMessaging redis = new RedisMessaging(this);
redisMessaging.init(getConfiguration().get(ConfigKeys.REDIS_ADDRESS), getConfiguration().get(ConfigKeys.REDIS_PASSWORD)); try {
getLog().info("Loaded redis successfully..."); redis.init(getConfiguration().get(ConfigKeys.REDIS_ADDRESS), getConfiguration().get(ConfigKeys.REDIS_PASSWORD));
} catch (Exception e) { getLog().info("Loaded redis successfully...");
getLog().info("Couldn't load redis...");
e.printStackTrace(); messagingService = redis;
} catch (Exception e) {
getLog().warn("Couldn't load redis...");
e.printStackTrace();
}
} else {
getLog().warn("Messaging Service was set to redis, but redis is not enabled!");
} }
} else if (messagingType.equals("bungee")) {
getLog().info("Loading bungee messaging service...");
BungeeMessagingService bungeeMessaging = new BungeeMessagingService(this);
bungeeMessaging.init();
messagingService = bungeeMessaging;
} else if (messagingType.equals("lilypad")) {
getLog().info("Loading LilyPad messaging service...");
if (getServer().getPluginManager().getPlugin("LilyPad-Connect") == null) {
getLog().warn("LilyPad-Connect plugin not present.");
} else {
LilyPadMessagingService lilyPadMessaging = new LilyPadMessagingService(this);
lilyPadMessaging.init();
messagingService = lilyPadMessaging;
}
} else if (!messagingType.equals("none")) {
getLog().warn("Messaging service '" + messagingType + "' not recognised.");
} }
// setup the update task buffer // setup the update task buffer
@ -341,9 +367,9 @@ public class LPBukkitPlugin extends JavaPlugin implements LuckPermsPlugin {
getLog().info("Closing datastore..."); getLog().info("Closing datastore...");
storage.shutdown(); storage.shutdown();
if (redisMessaging != null) { if (messagingService != null) {
getLog().info("Closing redis..."); getLog().info("Closing messaging service...");
redisMessaging.shutdown(); messagingService.close();
} }
getLog().info("Unregistering API..."); getLog().info("Unregistering API...");
@ -374,7 +400,7 @@ public class LPBukkitPlugin extends JavaPlugin implements LuckPermsPlugin {
groupManager = null; groupManager = null;
trackManager = null; trackManager = null;
storage = null; storage = null;
redisMessaging = null; messagingService = null;
uuidCache = null; uuidCache = null;
listener = null; listener = null;
apiProvider = null; apiProvider = null;

View File

@ -0,0 +1,95 @@
/*
* 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.bukkit.messaging;
import com.google.common.collect.Iterables;
import com.google.common.io.ByteArrayDataInput;
import com.google.common.io.ByteArrayDataOutput;
import com.google.common.io.ByteStreams;
import me.lucko.luckperms.bukkit.LPBukkitPlugin;
import me.lucko.luckperms.common.messaging.AbstractMessagingService;
import org.bukkit.entity.Player;
import org.bukkit.plugin.messaging.PluginMessageListener;
import org.bukkit.scheduler.BukkitRunnable;
import java.util.Collection;
/**
* An implementation of {@link me.lucko.luckperms.api.MessagingService} using the plugin messaging channels.
*/
public class BungeeMessagingService extends AbstractMessagingService implements PluginMessageListener {
private final LPBukkitPlugin plugin;
public BungeeMessagingService(LPBukkitPlugin plugin) {
super(plugin, "Bungee");
this.plugin = plugin;
}
public void init() {
plugin.getServer().getMessenger().registerOutgoingPluginChannel(plugin, CHANNEL);
plugin.getServer().getMessenger().registerIncomingPluginChannel(plugin, CHANNEL, this);
}
@Override
public void close() {
plugin.getServer().getMessenger().unregisterIncomingPluginChannel(plugin, CHANNEL);
plugin.getServer().getMessenger().unregisterOutgoingPluginChannel(plugin, CHANNEL);
}
@Override
protected void sendMessage(String channel, String message) {
new BukkitRunnable() {
@Override
public void run() {
Collection<? extends Player> players = plugin.getServer().getOnlinePlayers();
if (players.isEmpty()) {
return;
}
Player p = Iterables.getFirst(players, null);
ByteArrayDataOutput out = ByteStreams.newDataOutput();
out.writeUTF(message);
byte[] data = out.toByteArray();
p.sendPluginMessage(plugin, channel, data);
cancel();
}
}.runTaskTimer(plugin, 1L, 100L);
}
@Override
public void onPluginMessageReceived(String s, Player player, byte[] bytes) {
if (!s.equals(CHANNEL)) {
return;
}
ByteArrayDataInput in = ByteStreams.newDataInput(bytes);
String msg = in.readUTF();
onMessage(s, msg, null);
}
}

View File

@ -0,0 +1,90 @@
/*
* 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.bukkit.messaging;
import me.lucko.luckperms.bukkit.LPBukkitPlugin;
import me.lucko.luckperms.common.messaging.AbstractMessagingService;
import lilypad.client.connect.api.Connect;
import lilypad.client.connect.api.event.EventListener;
import lilypad.client.connect.api.event.MessageEvent;
import lilypad.client.connect.api.request.RequestException;
import lilypad.client.connect.api.request.impl.MessageRequest;
import java.io.UnsupportedEncodingException;
import java.util.Collections;
/**
* An implementation of {@link me.lucko.luckperms.api.MessagingService} using LilyPad.
*/
public class LilyPadMessagingService extends AbstractMessagingService {
private final LPBukkitPlugin plugin;
private Connect connect;
public LilyPadMessagingService(LPBukkitPlugin plugin) {
super(plugin, "LilyPad");
this.plugin = plugin;
}
public void init() {
connect = plugin.getServer().getServicesManager().getRegistration(Connect.class).getProvider();
connect.registerEvents(this);
}
@Override
public void close() {
connect.unregisterEvents(this);
}
@Override
protected void sendMessage(String channel, String message) {
MessageRequest request;
try {
request = new MessageRequest(Collections.emptyList(), channel, message);
} catch (UnsupportedEncodingException e) {
e.printStackTrace();
return;
}
try {
connect.request(request);
} catch (RequestException e) {
e.printStackTrace();
}
}
@EventListener
public void onMessage(MessageEvent event) {
plugin.doAsync(() -> {
try {
String channel = event.getChannel();
String message = event.getMessageAsString();
onMessage(channel, message, null);
} catch (Exception e) {
e.printStackTrace();
}
});
}
}

View File

@ -213,15 +213,23 @@ data:
# e.g. if you're using sqlite or flatfile, this can be set to -1 to save resources. # e.g. if you're using sqlite or flatfile, this can be set to -1 to save resources.
sync-minutes: 3 sync-minutes: 3
# Settings for Redis. # Settings for the messaging service
# #
# If enabled and configured, LuckPerms will use the Redis PubSub system to inform other # If enabled and configured, LuckPerms will use the messaging system to inform other
# connected servers of changes. Use the command "/luckperms networksync" to push changes. # connected servers of changes. Use the command "/luckperms networksync" to push changes.
# Data is NOT stored on redis. It is only used as a messaging platform. # Data is NOT stored using this service. It is only used as a messaging platform.
# #
# If you decide to enable this feature, you should set "sync-minutes" to -1, as there is no need for LuckPerms # If you decide to enable this feature, you should set "sync-minutes" to -1, as there is no need for LuckPerms
# to poll the database for changes. # to poll the database for changes.
# #
# Available options:
# bungee ==> uses the plugin messaging channels. Must be enabled on all connected servers to work.
# lilypad ==> uses lilypad pub sub to push changes. You need to have the LilyPad-Connect plugin installed.
# redis ==> uses redis pub sub to push changes. Your redis server must be configured below.
# none ==> nothing
messaging-service: none
# Settings for Redis.
# Port 6379 is used by default; set address to "host:port" if differs # Port 6379 is used by default; set address to "host:port" if differs
redis: redis:
enabled: false enabled: false

View File

@ -11,6 +11,7 @@ load: STARTUP
# It in turn fixes issues where plugins using Vault cache the provided instance when their plugin enables, or # It in turn fixes issues where plugins using Vault cache the provided instance when their plugin enables, or
# when they check for the presence of a service provider, before LuckPerms has enabled. # when they check for the presence of a service provider, before LuckPerms has enabled.
loadbefore: [Vault] loadbefore: [Vault]
softdepend: [LilyPad-Connect]
commands: commands:
luckperms: luckperms:

View File

@ -29,6 +29,7 @@ import me.lucko.luckperms.api.Logger;
import me.lucko.luckperms.api.PlatformType; import me.lucko.luckperms.api.PlatformType;
import me.lucko.luckperms.api.context.ContextSet; import me.lucko.luckperms.api.context.ContextSet;
import me.lucko.luckperms.api.context.MutableContextSet; import me.lucko.luckperms.api.context.MutableContextSet;
import me.lucko.luckperms.bungee.messaging.BungeeMessagingService;
import me.lucko.luckperms.common.LuckPermsPlugin; import me.lucko.luckperms.common.LuckPermsPlugin;
import me.lucko.luckperms.common.api.ApiHandler; import me.lucko.luckperms.common.api.ApiHandler;
import me.lucko.luckperms.common.api.ApiProvider; import me.lucko.luckperms.common.api.ApiProvider;
@ -53,6 +54,7 @@ import me.lucko.luckperms.common.managers.UserManager;
import me.lucko.luckperms.common.managers.impl.GenericGroupManager; import me.lucko.luckperms.common.managers.impl.GenericGroupManager;
import me.lucko.luckperms.common.managers.impl.GenericTrackManager; import me.lucko.luckperms.common.managers.impl.GenericTrackManager;
import me.lucko.luckperms.common.managers.impl.GenericUserManager; import me.lucko.luckperms.common.managers.impl.GenericUserManager;
import me.lucko.luckperms.common.messaging.AbstractMessagingService;
import me.lucko.luckperms.common.messaging.RedisMessaging; import me.lucko.luckperms.common.messaging.RedisMessaging;
import me.lucko.luckperms.common.storage.Storage; import me.lucko.luckperms.common.storage.Storage;
import me.lucko.luckperms.common.storage.StorageFactory; import me.lucko.luckperms.common.storage.StorageFactory;
@ -91,7 +93,7 @@ public class LPBungeePlugin extends Plugin implements LuckPermsPlugin {
private GroupManager groupManager; private GroupManager groupManager;
private TrackManager trackManager; private TrackManager trackManager;
private Storage storage; private Storage storage;
private RedisMessaging redisMessaging = null; private AbstractMessagingService messagingService = null;
private UuidCache uuidCache; private UuidCache uuidCache;
private ApiProvider apiProvider; private ApiProvider apiProvider;
private Logger log; private Logger log;
@ -129,17 +131,31 @@ public class LPBungeePlugin extends Plugin implements LuckPermsPlugin {
// initialise datastore // initialise datastore
storage = StorageFactory.getInstance(this, StorageType.H2); storage = StorageFactory.getInstance(this, StorageType.H2);
// initialise redis // initialise messaging
if (getConfiguration().get(ConfigKeys.REDIS_ENABLED)) { String messagingType = getConfiguration().get(ConfigKeys.MESSAGING_SERVICE).toLowerCase();
if (messagingType.equals("redis")) {
getLog().info("Loading redis..."); getLog().info("Loading redis...");
redisMessaging = new RedisMessaging(this); if (getConfiguration().get(ConfigKeys.REDIS_ENABLED)) {
try { RedisMessaging redis = new RedisMessaging(this);
redisMessaging.init(getConfiguration().get(ConfigKeys.REDIS_ADDRESS), getConfiguration().get(ConfigKeys.REDIS_PASSWORD)); try {
getLog().info("Loaded redis successfully..."); redis.init(getConfiguration().get(ConfigKeys.REDIS_ADDRESS), getConfiguration().get(ConfigKeys.REDIS_PASSWORD));
} catch (Exception e) { getLog().info("Loaded redis successfully...");
getLog().info("Couldn't load redis...");
e.printStackTrace(); messagingService = redis;
} catch (Exception e) {
getLog().warn("Couldn't load redis...");
e.printStackTrace();
}
} else {
getLog().warn("Messaging Service was set to redis, but redis is not enabled!");
} }
} else if (messagingType.equals("bungee")) {
getLog().info("Loading bungee messaging service...");
BungeeMessagingService bungeeMessaging = new BungeeMessagingService(this);
bungeeMessaging.init();
messagingService = bungeeMessaging;
} else if (!messagingType.equals("none")) {
getLog().warn("Messaging service '" + messagingType + "' not recognised.");
} }
// setup the update task buffer // setup the update task buffer
@ -215,9 +231,9 @@ public class LPBungeePlugin extends Plugin implements LuckPermsPlugin {
getLog().info("Closing datastore..."); getLog().info("Closing datastore...");
storage.shutdown(); storage.shutdown();
if (redisMessaging != null) { if (messagingService != null) {
getLog().info("Closing redis..."); getLog().info("Closing messaging service...");
redisMessaging.shutdown(); messagingService.close();
} }
getLog().info("Unregistering API..."); getLog().info("Unregistering API...");

View File

@ -0,0 +1,92 @@
/*
* 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.bungee.messaging;
import com.google.common.io.ByteArrayDataInput;
import com.google.common.io.ByteArrayDataOutput;
import com.google.common.io.ByteStreams;
import me.lucko.luckperms.bungee.LPBungeePlugin;
import me.lucko.luckperms.common.messaging.AbstractMessagingService;
import net.md_5.bungee.api.config.ServerInfo;
import net.md_5.bungee.api.connection.ProxiedPlayer;
import net.md_5.bungee.api.event.PluginMessageEvent;
import net.md_5.bungee.api.plugin.Listener;
import net.md_5.bungee.event.EventHandler;
/**
* An implementation of {@link me.lucko.luckperms.api.MessagingService} using the plugin messaging channels.
*/
public class BungeeMessagingService extends AbstractMessagingService implements Listener {
private final LPBungeePlugin plugin;
public BungeeMessagingService(LPBungeePlugin plugin) {
super(plugin, "Bungee");
this.plugin = plugin;
}
public void init() {
plugin.getProxy().getPluginManager().registerListener(plugin, this);
plugin.getProxy().registerChannel(CHANNEL);
}
@Override
public void close() {
plugin.getProxy().unregisterChannel(CHANNEL);
}
@Override
protected void sendMessage(String channel, String message) {
ByteArrayDataOutput out = ByteStreams.newDataOutput();
out.writeUTF(message);
byte[] data = out.toByteArray();
for (ServerInfo server : plugin.getProxy().getServers().values()) {
server.sendData(channel, data, true);
}
}
@EventHandler
public void onPluginMessage(PluginMessageEvent e) {
if (!e.getTag().equals(CHANNEL)) {
return;
}
e.setCancelled(true);
if (e.getSender() instanceof ProxiedPlayer) {
return;
}
ByteArrayDataInput in = ByteStreams.newDataInput(e.getData());
String msg = in.readUTF();
onMessage(e.getTag(), msg, u -> {
// Forward to other servers
plugin.doAsync(() -> sendMessage(CHANNEL, "update:" + u.toString()));
});
}
}

View File

@ -155,15 +155,22 @@ data:
# e.g. if you're using sqlite or flatfile, this can be set to -1 to save resources. # e.g. if you're using sqlite or flatfile, this can be set to -1 to save resources.
sync-minutes: 3 sync-minutes: 3
# Settings for Redis. # Settings for the messaging service
# #
# If enabled and configured, LuckPerms will use the Redis PubSub system to inform other # If enabled and configured, LuckPerms will use the messaging system to inform other
# connected servers of changes. Use the command "/luckpermsbungee networksync" to push changes. # connected servers of changes. Use the command "/luckpermsbungee networksync" to push changes.
# Data is NOT stored on redis. It is only used as a messaging platform. # Data is NOT stored using this service. It is only used as a messaging platform.
# #
# If you decide to enable this feature, you should set "sync-minutes" to -1, as there is no need for LuckPerms # If you decide to enable this feature, you should set "sync-minutes" to -1, as there is no need for LuckPerms
# to poll the database for changes. # to poll the database for changes.
# #
# Available options:
# bungee ==> uses the plugin messaging channels. Must be enabled on all connected servers to work.
# redis ==> uses redis pub sub to push changes. Your redis server must be configured below.
# none ==> nothing
messaging-service: none
# Settings for Redis.
# Port 6379 is used by default; set address to "host:port" if differs # Port 6379 is used by default; set address to "host:port" if differs
redis: redis:
enabled: false enabled: false

View File

@ -41,7 +41,7 @@ import me.lucko.luckperms.common.locale.LocaleManager;
import me.lucko.luckperms.common.managers.GroupManager; import me.lucko.luckperms.common.managers.GroupManager;
import me.lucko.luckperms.common.managers.TrackManager; import me.lucko.luckperms.common.managers.TrackManager;
import me.lucko.luckperms.common.managers.UserManager; import me.lucko.luckperms.common.managers.UserManager;
import me.lucko.luckperms.common.messaging.RedisMessaging; import me.lucko.luckperms.common.messaging.AbstractMessagingService;
import me.lucko.luckperms.common.storage.Storage; import me.lucko.luckperms.common.storage.Storage;
import me.lucko.luckperms.common.utils.BufferedRequest; import me.lucko.luckperms.common.utils.BufferedRequest;
import me.lucko.luckperms.common.utils.DebugHandler; import me.lucko.luckperms.common.utils.DebugHandler;
@ -103,7 +103,7 @@ public interface LuckPermsPlugin {
* *
* @return the redis messaging service * @return the redis messaging service
*/ */
RedisMessaging getRedisMessaging(); AbstractMessagingService getMessagingService();
/** /**
* Gets a wrapped logger instance for the platform. * Gets a wrapped logger instance for the platform.

View File

@ -135,7 +135,7 @@ public class ApiProvider implements LuckPermsApi {
@Override @Override
public Optional<MessagingService> getMessagingService() { public Optional<MessagingService> getMessagingService() {
return Optional.ofNullable(plugin.getRedisMessaging()); return Optional.ofNullable(plugin.getMessagingService());
} }
@Override @Override

View File

@ -44,8 +44,8 @@ public class NetworkSyncCommand extends SingleCommand {
plugin.getUpdateTaskBuffer().request().join(); plugin.getUpdateTaskBuffer().request().join();
Message.UPDATE_TASK_COMPLETE_NETWORK.send(sender); Message.UPDATE_TASK_COMPLETE_NETWORK.send(sender);
if (plugin.getRedisMessaging() != null) { if (plugin.getMessagingService() != null) {
plugin.getRedisMessaging().pushUpdate(); plugin.getMessagingService().pushUpdate();
Message.UPDATE_TASK_PUSH_SUCCESS.send(sender); Message.UPDATE_TASK_PUSH_SUCCESS.send(sender);
} else { } else {
Message.UPDATE_TASK_PUSH_FAILURE.send(sender); Message.UPDATE_TASK_PUSH_FAILURE.send(sender);

View File

@ -139,6 +139,7 @@ public class ConfigKeys {
.put("log", c.getString("split-storage.methods.log", "h2")) .put("log", c.getString("split-storage.methods.log", "h2"))
.build(); .build();
})); }));
public static final ConfigKey<String> MESSAGING_SERVICE = EnduringKey.wrap(StringKey.of("messaging-service", "none"));
public static final ConfigKey<Boolean> REDIS_ENABLED = EnduringKey.wrap(BooleanKey.of("redis.enabled", false)); public static final ConfigKey<Boolean> REDIS_ENABLED = EnduringKey.wrap(BooleanKey.of("redis.enabled", false));
public static final ConfigKey<String> REDIS_ADDRESS = EnduringKey.wrap(StringKey.of("redis.address", null)); public static final ConfigKey<String> REDIS_ADDRESS = EnduringKey.wrap(StringKey.of("redis.address", null));
public static final ConfigKey<String> REDIS_PASSWORD = EnduringKey.wrap(StringKey.of("redis.password", "")); public static final ConfigKey<String> REDIS_PASSWORD = EnduringKey.wrap(StringKey.of("redis.password", ""));

View File

@ -0,0 +1,103 @@
/*
* 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.api.MessagingService;
import me.lucko.luckperms.common.LuckPermsPlugin;
import java.util.Collections;
import java.util.HashSet;
import java.util.Set;
import java.util.UUID;
import java.util.function.Consumer;
/**
* An abstract implementation of {@link me.lucko.luckperms.api.MessagingService}.
*/
@RequiredArgsConstructor
public abstract class AbstractMessagingService implements MessagingService {
public static final String CHANNEL = "lpuc";
private final LuckPermsPlugin plugin;
private final String name;
private final Set<UUID> receivedMsgs = Collections.synchronizedSet(new HashSet<>());
public abstract void close();
protected abstract void sendMessage(String channel, String message);
protected void onMessage(String channel, String msg, Consumer<UUID> callback) {
if (!channel.equals(CHANNEL)) {
return;
}
UUID uuid = parseUpdateMessage(msg);
if (uuid == null) {
return;
}
if (!receivedMsgs.add(uuid)) {
return;
}
plugin.getLog().info("[" + name + " Messaging] Received update ping with id: " + uuid.toString());
plugin.getUpdateTaskBuffer().request();
if (callback != null) {
callback.accept(uuid);
}
}
@Override
public void pushUpdate() {
plugin.doAsync(() -> {
UUID id = generateId();
plugin.getLog().info("[" + name + " Messaging] Sending ping with id: " + id.toString());
sendMessage(CHANNEL, "update:" + id.toString());
});
}
private UUID generateId() {
UUID uuid = UUID.randomUUID();
receivedMsgs.add(uuid);
return uuid;
}
private static UUID parseUpdateMessage(String msg) {
if (!msg.startsWith("update:")) {
return null;
}
String requestId = msg.substring("update:".length());
try {
return UUID.fromString(requestId);
} catch (IllegalArgumentException e) {
return null;
}
}
}

View File

@ -24,7 +24,6 @@ package me.lucko.luckperms.common.messaging;
import lombok.RequiredArgsConstructor; import lombok.RequiredArgsConstructor;
import me.lucko.luckperms.api.MessagingService;
import me.lucko.luckperms.common.LuckPermsPlugin; import me.lucko.luckperms.common.LuckPermsPlugin;
import redis.clients.jedis.Jedis; import redis.clients.jedis.Jedis;
@ -32,22 +31,19 @@ import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig; import redis.clients.jedis.JedisPoolConfig;
import redis.clients.jedis.JedisPubSub; 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 * An implementation of {@link me.lucko.luckperms.api.MessagingService} using Redis.
*/ */
@RequiredArgsConstructor public class RedisMessaging extends AbstractMessagingService {
public class RedisMessaging implements MessagingService {
private static final String CHANNEL = "luckperms";
private final LuckPermsPlugin plugin; private final LuckPermsPlugin plugin;
private JedisPool jedisPool; private JedisPool jedisPool;
private LPSub sub; private LPSub sub;
public RedisMessaging(LuckPermsPlugin plugin) {
super(plugin, "Redis");
this.plugin = plugin;
}
public void init(String address, String password) { public void init(String address, String password) {
String[] addressSplit = address.split(":"); String[] addressSplit = address.split(":");
String host = addressSplit[0]; String host = addressSplit[0];
@ -60,7 +56,7 @@ public class RedisMessaging implements MessagingService {
} }
plugin.doAsync(() -> { plugin.doAsync(() -> {
sub = new LPSub(plugin); sub = new LPSub(this);
try (Jedis jedis = jedisPool.getResource()) { try (Jedis jedis = jedisPool.getResource()) {
jedis.subscribe(sub, CHANNEL); jedis.subscribe(sub, CHANNEL);
} catch (Exception e) { } catch (Exception e) {
@ -69,60 +65,28 @@ public class RedisMessaging implements MessagingService {
}); });
} }
public void shutdown() { @Override
public void close() {
sub.unsubscribe(); sub.unsubscribe();
jedisPool.destroy(); jedisPool.destroy();
} }
@Override @Override
public void pushUpdate() { protected void sendMessage(String channel, String message) {
plugin.doAsync(() -> { try (Jedis jedis = jedisPool.getResource()) {
UUID id = sub.generateId(); jedis.publish(CHANNEL, message);
plugin.getLog().info("[Redis Messaging] Sending redis ping with id: " + id.toString()); } catch (Exception e) {
try (Jedis jedis = jedisPool.getResource()) { e.printStackTrace();
jedis.publish(CHANNEL, "update:" + id.toString()); }
} catch (Exception e) {
e.printStackTrace();
}
});
} }
@RequiredArgsConstructor @RequiredArgsConstructor
private static class LPSub extends JedisPubSub { private static class LPSub extends JedisPubSub {
private final LuckPermsPlugin plugin; private final RedisMessaging parent;
private final Set<UUID> receivedMsgs = Collections.synchronizedSet(new HashSet<>());
private UUID generateId() {
UUID uuid = UUID.randomUUID();
receivedMsgs.add(uuid);
return uuid;
}
@Override @Override
public void onMessage(String channel, String msg) { public void onMessage(String channel, String msg) {
if (!channel.equals(CHANNEL)) { parent.onMessage(channel, msg, null);
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();
} }
} }

View File

@ -85,5 +85,9 @@
<id>placeholderapi</id> <id>placeholderapi</id>
<url>http://repo.extendedclip.com/content/repositories/placeholderapi/</url> <url>http://repo.extendedclip.com/content/repositories/placeholderapi/</url>
</repository> </repository>
<repository>
<id>lilypad-repo</id>
<url>http://ci.lilypadmc.org/plugin/repository/everything</url>
</repository>
</repositories> </repositories>
</project> </project>

View File

@ -50,6 +50,7 @@ import me.lucko.luckperms.common.locale.NoopLocaleManager;
import me.lucko.luckperms.common.locale.SimpleLocaleManager; import me.lucko.luckperms.common.locale.SimpleLocaleManager;
import me.lucko.luckperms.common.managers.TrackManager; import me.lucko.luckperms.common.managers.TrackManager;
import me.lucko.luckperms.common.managers.impl.GenericTrackManager; import me.lucko.luckperms.common.managers.impl.GenericTrackManager;
import me.lucko.luckperms.common.messaging.AbstractMessagingService;
import me.lucko.luckperms.common.messaging.RedisMessaging; import me.lucko.luckperms.common.messaging.RedisMessaging;
import me.lucko.luckperms.common.storage.Storage; import me.lucko.luckperms.common.storage.Storage;
import me.lucko.luckperms.common.storage.StorageFactory; import me.lucko.luckperms.common.storage.StorageFactory;
@ -65,6 +66,7 @@ import me.lucko.luckperms.sponge.commands.SpongeMainCommand;
import me.lucko.luckperms.sponge.contexts.WorldCalculator; import me.lucko.luckperms.sponge.contexts.WorldCalculator;
import me.lucko.luckperms.sponge.managers.SpongeGroupManager; import me.lucko.luckperms.sponge.managers.SpongeGroupManager;
import me.lucko.luckperms.sponge.managers.SpongeUserManager; import me.lucko.luckperms.sponge.managers.SpongeUserManager;
import me.lucko.luckperms.sponge.messaging.BungeeMessagingService;
import me.lucko.luckperms.sponge.service.LuckPermsService; import me.lucko.luckperms.sponge.service.LuckPermsService;
import me.lucko.luckperms.sponge.service.ServiceCacheHousekeepingTask; import me.lucko.luckperms.sponge.service.ServiceCacheHousekeepingTask;
import me.lucko.luckperms.sponge.service.base.LPSubjectCollection; import me.lucko.luckperms.sponge.service.base.LPSubjectCollection;
@ -146,7 +148,7 @@ public class LPSpongePlugin implements LuckPermsPlugin {
private SpongeGroupManager groupManager; private SpongeGroupManager groupManager;
private TrackManager trackManager; private TrackManager trackManager;
private Storage storage; private Storage storage;
private RedisMessaging redisMessaging = null; private AbstractMessagingService messagingService = null;
private UuidCache uuidCache; private UuidCache uuidCache;
private ApiProvider apiProvider; private ApiProvider apiProvider;
private me.lucko.luckperms.api.Logger log; private me.lucko.luckperms.api.Logger log;
@ -185,18 +187,31 @@ public class LPSpongePlugin implements LuckPermsPlugin {
// initialise datastore // initialise datastore
storage = StorageFactory.getInstance(this, StorageType.H2); storage = StorageFactory.getInstance(this, StorageType.H2);
// initialise redis // initialise messaging
if (getConfiguration().get(ConfigKeys.REDIS_ENABLED)) { String messagingType = getConfiguration().get(ConfigKeys.MESSAGING_SERVICE).toLowerCase();
if (messagingType.equals("redis")) {
getLog().info("Loading redis..."); getLog().info("Loading redis...");
redisMessaging = new RedisMessaging(this); if (getConfiguration().get(ConfigKeys.REDIS_ENABLED)) {
try { RedisMessaging redis = new RedisMessaging(this);
redisMessaging.init(getConfiguration().get(ConfigKeys.REDIS_ADDRESS), getConfiguration().get(ConfigKeys.REDIS_PASSWORD)); try {
getLog().info("Loaded redis successfully..."); redis.init(getConfiguration().get(ConfigKeys.REDIS_ADDRESS), getConfiguration().get(ConfigKeys.REDIS_PASSWORD));
} catch (Exception e) { getLog().info("Loaded redis successfully...");
getLog().info("Couldn't load redis...");
e.printStackTrace(); messagingService = redis;
redisMessaging = null; } catch (Exception e) {
getLog().warn("Couldn't load redis...");
e.printStackTrace();
}
} else {
getLog().warn("Messaging Service was set to redis, but redis is not enabled!");
} }
} else if (messagingType.equals("bungee")) {
getLog().info("Loading bungee messaging service...");
BungeeMessagingService bungeeMessaging = new BungeeMessagingService(this);
bungeeMessaging.init();
messagingService = bungeeMessaging;
} else if (!messagingType.equals("none")) {
getLog().warn("Messaging service '" + messagingType + "' not recognised.");
} }
// setup the update task buffer // setup the update task buffer
@ -295,9 +310,9 @@ public class LPSpongePlugin implements LuckPermsPlugin {
getLog().info("Closing datastore..."); getLog().info("Closing datastore...");
storage.shutdown(); storage.shutdown();
if (redisMessaging != null) { if (messagingService != null) {
getLog().info("Closing redis..."); getLog().info("Closing messaging service...");
redisMessaging.shutdown(); messagingService.close();
} }
getLog().info("Unregistering API..."); getLog().info("Unregistering API...");

View File

@ -0,0 +1,83 @@
/*
* 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.sponge.messaging;
import com.google.common.collect.Iterables;
import me.lucko.luckperms.common.messaging.AbstractMessagingService;
import me.lucko.luckperms.sponge.LPSpongePlugin;
import org.spongepowered.api.Platform;
import org.spongepowered.api.entity.living.player.Player;
import org.spongepowered.api.network.ChannelBinding;
import org.spongepowered.api.network.ChannelBuf;
import org.spongepowered.api.network.RawDataListener;
import org.spongepowered.api.network.RemoteConnection;
import java.util.Collection;
import java.util.concurrent.TimeUnit;
/**
* An implementation of {@link me.lucko.luckperms.api.MessagingService} using the plugin messaging channels.
*/
public class BungeeMessagingService extends AbstractMessagingService implements RawDataListener {
private final LPSpongePlugin plugin;
private ChannelBinding.RawDataChannel channel = null;
public BungeeMessagingService(LPSpongePlugin plugin) {
super(plugin, "Bungee");
this.plugin = plugin;
}
public void init() {
channel = plugin.getGame().getChannelRegistrar().createRawChannel(plugin, CHANNEL);
channel.addListener(Platform.Type.SERVER, this);
}
@Override
public void close() {
if (channel != null) {
plugin.getGame().getChannelRegistrar().unbindChannel(channel);
}
}
@Override
protected void sendMessage(String channel, String message) {
plugin.getScheduler().createTaskBuilder().interval(10, TimeUnit.SECONDS).execute(task -> {
Collection<Player> players = plugin.getGame().getServer().getOnlinePlayers();
if (players.isEmpty()) {
return;
}
Player p = Iterables.getFirst(players, null);
this.channel.sendTo(p, buf -> buf.writeUTF(message));
}).submit(plugin);
}
@Override
public void handlePayload(ChannelBuf buf, RemoteConnection connection, Platform.Type type) {
String msg = buf.readUTF();
onMessage(CHANNEL, msg, null);
}
}

View File

@ -162,15 +162,22 @@ data {
sync-minutes=3 sync-minutes=3
} }
# Settings for Redis. # Settings for the messaging service
# #
# If enabled and configured, LuckPerms will use the Redis PubSub system to inform other # If enabled and configured, LuckPerms will use the messaging system to inform other
# connected servers of changes. Use the command "/luckperms networksync" to push changes. # connected servers of changes. Use the command "/luckperms networksync" to push changes.
# Data is NOT stored on redis. It is only used as a messaging platform. # Data is NOT stored using this service. It is only used as a messaging platform.
# #
# If you decide to enable this feature, you should set "sync-minutes" to -1, as there is no need for LuckPerms # If you decide to enable this feature, you should set "sync-minutes" to -1, as there is no need for LuckPerms
# to poll the database for changes. # to poll the database for changes.
# #
# Available options:
# bungee ==> uses the plugin messaging channels. Must be enabled on all connected servers to work.
# redis ==> uses redis pub sub to push changes. Your redis server must be configured below.
# none ==> nothing
messaging-service="none"
# Settings for Redis.
# Port 6379 is used by default; set address to "host:port" if differs # Port 6379 is used by default; set address to "host:port" if differs
redis { redis {
enabled=false enabled=false