Add new PubSub handling along with associated API methods.

src/main/java/com/imaginarycode/minecraft/redisbungee/RedisBungee.java

@@ -15,6 +15,7 @@ import com.google.common.io.ByteArrayDataInput;
 import com.google.common.io.ByteArrayDataOutput;
 import com.google.common.io.ByteStreams;
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import com.imaginarycode.minecraft.redisbungee.events.PubSubMessageEvent;
 import lombok.NonNull;
 import net.md_5.bungee.api.ServerPing;
 import net.md_5.bungee.api.config.ServerInfo;
@@ -51,9 +52,9 @@ import static com.google.common.base.Preconditions.checkArgument;
  */
 public final class RedisBungee extends Plugin implements Listener {
     private static Configuration configuration;
-    private JedisPool pool;
+    private static JedisPool pool;
     private static RedisBungeeAPI api;
-    private PubSubListener psl = null;
+    private static PubSubListener psl = null;
     private static List<String> serverIds;
     private int globalCount;
 
@@ -74,6 +75,10 @@ public final class RedisBungee extends Plugin implements Listener {
         return serverIds;
     }
 
+    static PubSubListener getPubSubListener() {
+        return psl;
+    }
+
     final Multimap<String, String> serversToPlayers() {
         ImmutableMultimap.Builder<String, String> multimapBuilder = ImmutableMultimap.builder();
         for (String p : getPlayers()) {
@@ -594,13 +599,24 @@ public final class RedisBungee extends Plugin implements Listener {
         }
     }
 
+    @EventHandler
+    public void onPubSubMessage(PubSubMessageEvent event) {
+        if (event.getChannel().equals("redisbungee-allservers") || event.getChannel().equals("redisbungee-" + configuration.getString("server-id"))) {
+            String message = event.getMessage();
+            if (message.startsWith("/"))
+                message = message.substring(1);
+            getLogger().info("Invoking command via PubSub: /" + message);
+            getProxy().getPluginManager().dispatchCommand(RedisBungeeCommandSender.instance, message);
+        }
+    }
+
     private void cleanUpPlayer(String player, Jedis rsc) {
         rsc.srem("server:" + configuration.getString("server-id") + ":usersOnline", player);
         rsc.hdel("player:" + player, "server");
         rsc.hdel("player:" + player, "ip");
     }
 
-    private class PubSubListener implements Runnable {
+    class PubSubListener implements Runnable {
         private Jedis rsc;
         private JedisPubSubHandler jpsh;
 
@@ -621,25 +637,26 @@ public final class RedisBungee extends Plugin implements Listener {
             jpsh.unsubscribe();
             pool.returnResource(rsc);
         }
+
+        public void addChannel(String... channel) {
+            jpsh.subscribe(channel);
+        }
+
+        public void removeChannel(String... channel) {
+            jpsh.unsubscribe(channel);
+        }
     }
 
-    private class JedisPubSubHandler extends JedisPubSub {
+    class JedisPubSubHandler extends JedisPubSub {
-        private ExecutorService executor = Executors.newSingleThreadExecutor(new ThreadFactoryBuilder().setDaemon(true).setNameFormat("RedisBungee PubSub Command Executor").build());
+        private ExecutorService executor = Executors.newCachedThreadPool(new ThreadFactoryBuilder().setDaemon(true).setNameFormat("RedisBungee PubSub Handler - #%d").build());
 
         @Override
-        public void onMessage(String s, String s2) {
+        public void onMessage(final String s, final String s2) {
-            final String cmd;
-            if (s2.startsWith("/")) {
-                cmd = s2.substring(1);
-            } else {
-                cmd = s2;
-            }
             if (s2.trim().length() == 0) return;
-            getLogger().info("Invoking command from PubSub: /" + s2);
             executor.submit(new Runnable() {
                 @Override
                 public void run() {
-                    getProxy().getPluginManager().dispatchCommand(RedisBungeeCommandSender.instance, cmd);
+                    getProxy().getPluginManager().callEvent(new PubSubMessageEvent(s, s2));
                 }
             });
         }

src/main/java/com/imaginarycode/minecraft/redisbungee/RedisBungeeAPI.java

@@ -149,4 +149,24 @@ public class RedisBungeeAPI {
     public final List<String> getAllServers() {
         return RedisBungee.getServerIds();
     }
+
+    /**
+     * Register (a) PubSub channel(s), so that you may capture {@link com.imaginarycode.minecraft.redisbungee.events.PubSubMessageEvent} for it.
+     *
+     * @param channels the channels to register
+     * @since 0.2.6
+     */
+    public final void registerPubSubChannels(String... channels) {
+        RedisBungee.getPubSubListener().addChannel(channels);
+    }
+
+    /**
+     * Unregister (a) PubSub channel(s).
+     *
+     * @param channels the channels to unregister
+     * @since 0.2.6
+     */
+    public final void unregisterPubSubChannels(String... channels) {
+        RedisBungee.getPubSubListener().removeChannel(channels);
+    }
 }

src/main/java/com/imaginarycode/minecraft/redisbungee/events/PubSubMessageEvent.java

@@ -0,0 +1,31 @@
+/**
+ * Copyright © 2013 tuxed <write@imaginarycode.com>
+ * This work is free. You can redistribute it and/or modify it under the
+ * terms of the Do What The Fuck You Want To Public License, Version 2,
+ * as published by Sam Hocevar. See http://www.wtfpl.net/ for more details.
+ */
+package com.imaginarycode.minecraft.redisbungee.events;
+
+import lombok.RequiredArgsConstructor;
+import net.md_5.bungee.api.plugin.Event;
+
+/**
+ * This event is posted when a PubSub message is received.
+ *
+ * <strong>Warning</strong>: This event is fired in a separate thread!
+ *
+ * @since 0.2.6
+ */
+@RequiredArgsConstructor
+public class PubSubMessageEvent extends Event {
+    private final String channel;
+    private final String message;
+
+    public String getChannel() {
+        return channel;
+    }
+
+    public String getMessage() {
+        return message;
+    }
+}