From 2a1ca391e053bf32f99bc87b115e0370119ece09 Mon Sep 17 00:00:00 2001 From: Jonathan Leibiusky Date: Sat, 7 Aug 2010 17:54:14 -0300 Subject: [PATCH] Added pubsub support --- src/main/java/redis/clients/jedis/Client.java | 28 +++ src/main/java/redis/clients/jedis/Jedis.java | 12 ++ .../java/redis/clients/jedis/JedisPubSub.java | 84 ++++++++ .../tests/commands/JedisCommandTestBase.java | 11 ++ .../PublishSubscribeCommandsTest.java | 180 ++++++++++++++++++ 5 files changed, 315 insertions(+) create mode 100644 src/main/java/redis/clients/jedis/JedisPubSub.java create mode 100644 src/test/java/redis/clients/jedis/tests/commands/PublishSubscribeCommandsTest.java diff --git a/src/main/java/redis/clients/jedis/Client.java b/src/main/java/redis/clients/jedis/Client.java index 5a5b233..0779265 100644 --- a/src/main/java/redis/clients/jedis/Client.java +++ b/src/main/java/redis/clients/jedis/Client.java @@ -407,4 +407,32 @@ public class Client extends Connection { public void auth(String password) { sendCommand("AUTH", password); } + + public void subscribe(String... channels) { + sendCommand("SUBSCRIBE", channels); + } + + public void publish(String channel, String message) { + sendCommand("PUBLISH", channel, message); + } + + public void unsubscribe() { + sendCommand("UNSUBSCRIBE"); + } + + public void unsubscribe(String... channels) { + sendCommand("UNSUBSCRIBE", channels); + } + + public void psubscribe(String[] patterns) { + sendCommand("PSUBSCRIBE", patterns); + } + + public void punsubscribe() { + sendCommand("PUNSUBSCRIBE"); + } + + public void punsubscribe(String... patterns) { + sendCommand("PUNSUBSCRIBE", patterns); + } } diff --git a/src/main/java/redis/clients/jedis/Jedis.java b/src/main/java/redis/clients/jedis/Jedis.java index 8ede713..3186d0c 100644 --- a/src/main/java/redis/clients/jedis/Jedis.java +++ b/src/main/java/redis/clients/jedis/Jedis.java @@ -533,4 +533,16 @@ public class Jedis { jedisPipeline.execute(); return client.getAll(); } + + public void subscribe(JedisPubSub jedisPubSub, String... channels) { + jedisPubSub.proceed(client, channels); + } + + public void publish(String channel, String message) { + client.publish(channel, message); + } + + public void psubscribe(JedisPubSub jedisPubSub, String... patterns) { + jedisPubSub.proceedWithPatterns(client, patterns); + } } \ No newline at end of file diff --git a/src/main/java/redis/clients/jedis/JedisPubSub.java b/src/main/java/redis/clients/jedis/JedisPubSub.java new file mode 100644 index 0000000..7bc9e1d --- /dev/null +++ b/src/main/java/redis/clients/jedis/JedisPubSub.java @@ -0,0 +1,84 @@ +package redis.clients.jedis; + +import java.util.List; + +public abstract class JedisPubSub { + private int subscribedChannels = 0; + private Client client; + + public abstract void onMessage(String channel, String message); + + public abstract void onPMessage(String pattern, String channel, + String message); + + public abstract void onSubscribe(String channel, int subscribedChannels); + + public abstract void onUnsubscribe(String channel, int subscribedChannels); + + public abstract void onPUnsubscribe(String pattern, int subscribedChannels); + + public abstract void onPSubscribe(String pattern, int subscribedChannels); + + protected void unsubscribe() { + client.unsubscribe(); + } + + protected void unsubscribe(String... channels) { + client.unsubscribe(channels); + } + + protected void subscribe(String... channels) { + client.subscribe(channels); + } + + public boolean isSubscribed() { + return subscribedChannels > 0; + } + + public void proceedWithPatterns(Client client, String... patterns) { + this.client = client; + client.psubscribe(patterns); + process(client); + } + + public void proceed(Client client, String... channels) { + this.client = client; + client.subscribe(channels); + process(client); + } + + private void process(Client client) { + do { + List reply = client.getObjectMultiBulkReply(); + if (reply.get(0).equals("subscribe")) { + subscribedChannels = (Integer) reply.get(2); + onSubscribe((String) reply.get(1), subscribedChannels); + } else if (reply.get(0).equals("unsubscribe")) { + subscribedChannels = (Integer) reply.get(2); + onUnsubscribe((String) reply.get(1), subscribedChannels); + } else if (reply.get(0).equals("message")) { + onMessage((String) reply.get(1), (String) reply.get(2)); + } else if (reply.get(0).equals("pmessage")) { + onPMessage((String) reply.get(1), (String) reply.get(2), + (String) reply.get(3)); + } else if (reply.get(0).equals("psubscribe")) { + subscribedChannels = (Integer) reply.get(2); + onPSubscribe((String) reply.get(1), subscribedChannels); + } else if (reply.get(0).equals("punsubscribe")) { + subscribedChannels = (Integer) reply.get(2); + onPUnsubscribe((String) reply.get(1), subscribedChannels); + } else { + throw new JedisException("Unknown message type: " + + reply.get(0)); + } + } while (isSubscribed()); + } + + protected void punsubscribe() { + client.punsubscribe(); + } + + protected void punsubscribe(String... patterns) { + client.punsubscribe(patterns); + } +} diff --git a/src/test/java/redis/clients/jedis/tests/commands/JedisCommandTestBase.java b/src/test/java/redis/clients/jedis/tests/commands/JedisCommandTestBase.java index 3a1e5b8..ec4dd04 100644 --- a/src/test/java/redis/clients/jedis/tests/commands/JedisCommandTestBase.java +++ b/src/test/java/redis/clients/jedis/tests/commands/JedisCommandTestBase.java @@ -1,5 +1,8 @@ package redis.clients.jedis.tests.commands; +import java.io.IOException; +import java.net.UnknownHostException; + import junit.framework.Assert; import org.junit.After; @@ -27,4 +30,12 @@ public abstract class JedisCommandTestBase extends Assert { public void tearDown() throws Exception { jedis.disconnect(); } + + protected Jedis createJedis() throws UnknownHostException, IOException { + Jedis j = new Jedis("localhost"); + j.connect(); + j.auth("foobared"); + j.flushAll(); + return j; + } } \ No newline at end of file diff --git a/src/test/java/redis/clients/jedis/tests/commands/PublishSubscribeCommandsTest.java b/src/test/java/redis/clients/jedis/tests/commands/PublishSubscribeCommandsTest.java new file mode 100644 index 0000000..38b3c8c --- /dev/null +++ b/src/test/java/redis/clients/jedis/tests/commands/PublishSubscribeCommandsTest.java @@ -0,0 +1,180 @@ +package redis.clients.jedis.tests.commands; + +import java.io.IOException; +import java.net.UnknownHostException; + +import org.junit.Test; + +import redis.clients.jedis.Jedis; +import redis.clients.jedis.JedisPubSub; + +public class PublishSubscribeCommandsTest extends JedisCommandTestBase { + @Test + public void subscribe() throws UnknownHostException, IOException, + InterruptedException { + new Thread(new Runnable() { + public void run() { + try { + Jedis j = createJedis(); + Thread.sleep(1000); + j.publish("foo", "exit"); + j.disconnect(); + } catch (Exception ex) { + fail(ex.getMessage()); + } + } + }).start(); + jedis.subscribe(new JedisPubSub() { + public void onMessage(String channel, String message) { + assertEquals("foo", channel); + assertEquals("exit", message); + unsubscribe(); + } + + public void onSubscribe(String channel, int subscribedChannels) { + assertEquals("foo", channel); + assertEquals(1, subscribedChannels); + } + + public void onUnsubscribe(String channel, int subscribedChannels) { + assertEquals("foo", channel); + assertEquals(0, subscribedChannels); + } + + public void onPSubscribe(String pattern, int subscribedChannels) { + } + + public void onPUnsubscribe(String pattern, int subscribedChannels) { + } + + public void onPMessage(String pattern, String channel, + String message) { + } + }, "foo"); + } + + @Test + public void subscribeMany() throws UnknownHostException, IOException, + InterruptedException { + new Thread(new Runnable() { + public void run() { + try { + Jedis j = createJedis(); + Thread.sleep(1000); + j.publish("foo", "exit"); + Thread.sleep(1000); + j.publish("bar", "exit"); + j.disconnect(); + } catch (Exception ex) { + fail(ex.getMessage()); + } + } + }).start(); + jedis.subscribe(new JedisPubSub() { + public void onMessage(String channel, String message) { + unsubscribe(channel); + } + + public void onSubscribe(String channel, int subscribedChannels) { + } + + public void onUnsubscribe(String channel, int subscribedChannels) { + } + + public void onPSubscribe(String pattern, int subscribedChannels) { + } + + public void onPUnsubscribe(String pattern, int subscribedChannels) { + } + + public void onPMessage(String pattern, String channel, + String message) { + } + }, "foo", "bar"); + } + + @Test + public void psubscribe() throws UnknownHostException, IOException, + InterruptedException { + new Thread(new Runnable() { + public void run() { + try { + Jedis j = createJedis(); + Thread.sleep(1000); + j.publish("foo.bar", "exit"); + j.disconnect(); + } catch (Exception ex) { + fail(ex.getMessage()); + } + } + }).start(); + jedis.psubscribe(new JedisPubSub() { + public void onMessage(String channel, String message) { + } + + public void onSubscribe(String channel, int subscribedChannels) { + } + + public void onUnsubscribe(String channel, int subscribedChannels) { + } + + public void onPSubscribe(String pattern, int subscribedChannels) { + assertEquals("foo.*", pattern); + assertEquals(1, subscribedChannels); + } + + public void onPUnsubscribe(String pattern, int subscribedChannels) { + assertEquals("foo.*", pattern); + assertEquals(0, subscribedChannels); + } + + public void onPMessage(String pattern, String channel, + String message) { + assertEquals("foo.*", pattern); + assertEquals("foo.bar", channel); + assertEquals("exit", message); + punsubscribe(); + } + }, "foo.*"); + } + + @Test + public void psubscribeMany() throws UnknownHostException, IOException, + InterruptedException { + new Thread(new Runnable() { + public void run() { + try { + Jedis j = createJedis(); + Thread.sleep(1000); + j.publish("foo.123", "exit"); + Thread.sleep(1000); + j.publish("bar.123", "exit"); + j.disconnect(); + } catch (Exception ex) { + fail(ex.getMessage()); + } + } + }).start(); + jedis.psubscribe(new JedisPubSub() { + public void onMessage(String channel, String message) { + } + + public void onSubscribe(String channel, int subscribedChannels) { + } + + public void onUnsubscribe(String channel, int subscribedChannels) { + } + + public void onPSubscribe(String pattern, int subscribedChannels) { + } + + public void onPUnsubscribe(String pattern, int subscribedChannels) { + } + + public void onPMessage(String pattern, String channel, + String message) { + punsubscribe(pattern); + } + }, "foo.*", "bar.*"); + } +} \ No newline at end of file