Added pubsub support

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);
+    }
 }

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);
+    }
 }

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<Object> 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);
+    }
+}

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;
+    }
 }

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.*");
+    }
+}