Skip to content

Commit

Permalink
Merge pull request #523 from xetorthio/pubsub
Browse files Browse the repository at this point in the history 
Add pubsub commands
  • Loading branch information
@xetorthio
xetorthio committed Feb 3, 2014
2 parents d5f984a + 4ab54d9 commit bfecfcb
Show file tree
Hide file tree
Showing 6 changed files with 171 additions and 3 deletions.
5 changes: 4 additions & 1 deletion src/main/java/redis/clients/jedis/BinaryClient.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -560,7 +560,10 @@ public void punsubscribe() {
public void punsubscribe(final byte[]... patterns) {
sendCommand(PUNSUBSCRIBE, patterns);
}


public void pubsub(final byte[]... args) {
sendCommand(PUBSUB, args);
}
public void zcount(final byte[] key, final double min, final double max) {

byte byteArrayMin[] = (min == Double.NEGATIVE_INFINITY) ? "-inf"
Expand Down
1 change: 1 addition & 0 deletions src/main/java/redis/clients/jedis/BuilderFactory.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -103,6 +103,7 @@ public String toString() {
}

};

public static final Builder<Set<String>> STRING_SET = new Builder<Set<String>>() {
@SuppressWarnings("unchecked")
public Set<String> build(Object data) {
Expand Down
21 changes: 21 additions & 0 deletions src/main/java/redis/clients/jedis/Client.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -672,6 +672,18 @@ public void subscribe(final String... channels) {
}
subscribe(cs);
}

public void pubsubChannels(String pattern) {
pubsub(Protocol.PUBSUB_CHANNELS, pattern);
}

public void pubsubNumPat() {
pubsub(Protocol.PUBSUB_NUM_PAT);
}

public void pubsubNumSub(String... channels) {
pubsub(Protocol.PUBSUB_NUMSUB, channels);
}

public void configSet(String parameter, String value) {
configSet(SafeEncoder.encode(parameter), SafeEncoder.encode(value));
Expand Down Expand Up @@ -841,6 +853,15 @@ public void cluster(final String subcommand, final int... args) {
arg[0] = SafeEncoder.encode(subcommand);
cluster(arg);
}

public void pubsub(final String subcommand, final String... args) {
final byte[][] arg = new byte[args.length+1][];
for (int i = 1; i < arg.length; i++) {
arg[i] = SafeEncoder.encode(args[i-1]);
}
arg[0] = SafeEncoder.encode(subcommand);
pubsub(arg);
}

public void cluster(final String subcommand, final String... args) {
final byte[][] arg = new byte[args.length + 1][];
Expand Down
19 changes: 19 additions & 0 deletions src/main/java/redis/clients/jedis/Jedis.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -3232,4 +3232,23 @@ public String asking() {
client.asking();
return client.getStatusCodeReply();
}

public List<String> pubsubChannels(String pattern) {
checkIsInMulti();
client.pubsubChannels(pattern);
return client.getMultiBulkReply();
}

public Long pubsubNumPat() {
checkIsInMulti();
client.pubsubNumPat();
return client.getIntegerReply();
}

public Map<String, String> pubsubNumSub(String... channels) {
checkIsInMulti();
client.pubsubNumSub(channels);
return BuilderFactory.STRING_MAP
.build(client.getBinaryMultiBulkReply());
}
}
7 changes: 5 additions & 2 deletions src/main/java/redis/clients/jedis/Protocol.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,9 @@ public final class Protocol {
public static final String CLUSTER_SETSLOT_NODE = "node";
public static final String CLUSTER_SETSLOT_MIGRATING = "migrating";
public static final String CLUSTER_SETSLOT_IMPORTING = "importing";
public static final String PUBSUB_CHANNELS= "channels";
public static final String PUBSUB_NUMSUB = "numsub";
public static final String PUBSUB_NUM_PAT = "numpat";

private Protocol() {
// this prevent the class from instantiation
Expand Down Expand Up @@ -181,7 +184,7 @@ public static Object read(final RedisInputStream is) {
}

public static final byte[] toByteArray(final boolean value) {
return toByteArray(value ? 1 : 0);
return toByteArray(value ? 1 : 0);
}

public static final byte[] toByteArray(final int value) {
Expand All @@ -197,7 +200,7 @@ public static final byte[] toByteArray(final double value) {
}

public static enum Command {
PING, SET, GET, QUIT, EXISTS, DEL, TYPE, FLUSHDB, KEYS, RANDOMKEY, RENAME, RENAMENX, RENAMEX, DBSIZE, EXPIRE, EXPIREAT, TTL, SELECT, MOVE, FLUSHALL, GETSET, MGET, SETNX, SETEX, MSET, MSETNX, DECRBY, DECR, INCRBY, INCR, APPEND, SUBSTR, HSET, HGET, HSETNX, HMSET, HMGET, HINCRBY, HEXISTS, HDEL, HLEN, HKEYS, HVALS, HGETALL, RPUSH, LPUSH, LLEN, LRANGE, LTRIM, LINDEX, LSET, LREM, LPOP, RPOP, RPOPLPUSH, SADD, SMEMBERS, SREM, SPOP, SMOVE, SCARD, SISMEMBER, SINTER, SINTERSTORE, SUNION, SUNIONSTORE, SDIFF, SDIFFSTORE, SRANDMEMBER, ZADD, ZRANGE, ZREM, ZINCRBY, ZRANK, ZREVRANK, ZREVRANGE, ZCARD, ZSCORE, MULTI, DISCARD, EXEC, WATCH, UNWATCH, SORT, BLPOP, BRPOP, AUTH, SUBSCRIBE, PUBLISH, UNSUBSCRIBE, PSUBSCRIBE, PUNSUBSCRIBE, ZCOUNT, ZRANGEBYSCORE, ZREVRANGEBYSCORE, ZREMRANGEBYRANK, ZREMRANGEBYSCORE, ZUNIONSTORE, ZINTERSTORE, SAVE, BGSAVE, BGREWRITEAOF, LASTSAVE, SHUTDOWN, INFO, MONITOR, SLAVEOF, CONFIG, STRLEN, SYNC, LPUSHX, PERSIST, RPUSHX, ECHO, LINSERT, DEBUG, BRPOPLPUSH, SETBIT, GETBIT, SETRANGE, GETRANGE, EVAL, EVALSHA, SCRIPT, SLOWLOG, OBJECT, BITCOUNT, BITOP, SENTINEL, DUMP, RESTORE, PEXPIRE, PEXPIREAT, PTTL, INCRBYFLOAT, PSETEX, CLIENT, TIME, MIGRATE, HINCRBYFLOAT, SCAN, HSCAN, SSCAN, ZSCAN, WAIT, CLUSTER, ASKING;
PING, SET, GET, QUIT, EXISTS, DEL, TYPE, FLUSHDB, KEYS, RANDOMKEY, RENAME, RENAMENX, RENAMEX, DBSIZE, EXPIRE, EXPIREAT, TTL, SELECT, MOVE, FLUSHALL, GETSET, MGET, SETNX, SETEX, MSET, MSETNX, DECRBY, DECR, INCRBY, INCR, APPEND, SUBSTR, HSET, HGET, HSETNX, HMSET, HMGET, HINCRBY, HEXISTS, HDEL, HLEN, HKEYS, HVALS, HGETALL, RPUSH, LPUSH, LLEN, LRANGE, LTRIM, LINDEX, LSET, LREM, LPOP, RPOP, RPOPLPUSH, SADD, SMEMBERS, SREM, SPOP, SMOVE, SCARD, SISMEMBER, SINTER, SINTERSTORE, SUNION, SUNIONSTORE, SDIFF, SDIFFSTORE, SRANDMEMBER, ZADD, ZRANGE, ZREM, ZINCRBY, ZRANK, ZREVRANK, ZREVRANGE, ZCARD, ZSCORE, MULTI, DISCARD, EXEC, WATCH, UNWATCH, SORT, BLPOP, BRPOP, AUTH, SUBSCRIBE, PUBLISH, UNSUBSCRIBE, PSUBSCRIBE, PUNSUBSCRIBE, PUBSUB, ZCOUNT, ZRANGEBYSCORE, ZREVRANGEBYSCORE, ZREMRANGEBYRANK, ZREMRANGEBYSCORE, ZUNIONSTORE, ZINTERSTORE, SAVE, BGSAVE, BGREWRITEAOF, LASTSAVE, SHUTDOWN, INFO, MONITOR, SLAVEOF, CONFIG, STRLEN, SYNC, LPUSHX, PERSIST, RPUSHX, ECHO, LINSERT, DEBUG, BRPOPLPUSH, SETBIT, GETBIT, SETRANGE, GETRANGE, EVAL, EVALSHA, SCRIPT, SLOWLOG, OBJECT, BITCOUNT, BITOP, SENTINEL, DUMP, RESTORE, PEXPIRE, PEXPIREAT, PTTL, INCRBYFLOAT, PSETEX, CLIENT, TIME, MIGRATE, HINCRBYFLOAT, SCAN, HSCAN, SSCAN, ZSCAN, WAIT, CLUSTER, ASKING;

public final byte[] raw;

Expand Down
121 changes: 121 additions & 0 deletions src/test/java/redis/clients/jedis/tests/commands/PublishSubscribeCommandsTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,9 @@
import java.io.IOException;
import java.net.UnknownHostException;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;

import org.junit.Test;
Expand Down Expand Up @@ -61,6 +64,124 @@ public void onPMessage(String pattern, String channel,
}
}, "foo");
}


@Test
public void pubSubChannels(){
final List<String> expectedActiveChannels = Arrays.asList("testchan1", "testchan2", "testchan3");
jedis.subscribe(new JedisPubSub() {
private int count = 0;

@Override
public void onUnsubscribe(String channel, int subscribedChannels) {
}

@Override
public void onSubscribe(String channel, int subscribedChannels) {
count++;
//All channels are subscribed
if (count == 3) {
Jedis otherJedis = createJedis();
List<String> activeChannels = otherJedis.pubsubChannels("test*");
assertTrue(expectedActiveChannels.containsAll(activeChannels));
unsubscribe();
}
}

@Override
public void onPUnsubscribe(String pattern, int subscribedChannels) {
}

@Override
public void onPSubscribe(String pattern, int subscribedChannels) {
}

@Override
public void onPMessage(String pattern, String channel, String message) {
}

@Override
public void onMessage(String channel, String message) {
}
}, "testchan1", "testchan2", "testchan3");
}

@Test
public void pubSubNumPat(){
jedis.psubscribe(new JedisPubSub() {
private int count=0;
@Override
public void onUnsubscribe(String channel, int subscribedChannels) {
}

@Override
public void onSubscribe(String channel, int subscribedChannels) {
}

@Override
public void onPUnsubscribe(String pattern, int subscribedChannels) {
}

@Override
public void onPSubscribe(String pattern, int subscribedChannels) {
count++;
if (count == 3) {
Jedis otherJedis = createJedis();
Long numPatterns = otherJedis.pubsubNumPat();
assertEquals(new Long(2l), numPatterns);
punsubscribe();
}
}

@Override
public void onPMessage(String pattern, String channel, String message) {
}

@Override
public void onMessage(String channel, String message) {
}
}, "test*", "test*", "chan*");
}

@Test
public void pubSubNumSub(){
final Map<String, String> expectedNumSub = new HashMap<String, String>();
expectedNumSub.put("testchannel2", "1");
expectedNumSub.put("testchannel1", "1");
jedis.subscribe(new JedisPubSub() {
private int count=0;
@Override
public void onUnsubscribe(String channel, int subscribedChannels) {
}

@Override
public void onSubscribe(String channel, int subscribedChannels) {
count++;
if (count == 2) {
Jedis otherJedis = createJedis();
Map<String, String> numSub = otherJedis.pubsubNumSub("testchannel1", "testchannel2");
assertEquals(expectedNumSub, numSub);
unsubscribe();
}
}

@Override
public void onPUnsubscribe(String pattern, int subscribedChannels) {
}

@Override
public void onPSubscribe(String pattern, int subscribedChannels) {
}

@Override
public void onPMessage(String pattern, String channel, String message) {
}

@Override
public void onMessage(String channel, String message) {
}
}, "testchannel1", "testchannel2");
}

@Test
public void subscribeMany() throws UnknownHostException, IOException,
Expand Down

0 comments on commit bfecfcb

Please sign in to comment.