Merge branch 'master' into pipeline-and-transaction-can-handle-responses-on-their-own-status
* it's broken with later features, resetState() * fixed resetState() to make it work with this PR Conflicts: src/main/java/redis/clients/jedis/BinaryJedis.java src/main/java/redis/clients/jedis/Connection.java src/main/java/redis/clients/jedis/Pipeline.java src/main/java/redis/clients/jedis/Transaction.java src/main/java/redis/clients/jedis/TransactionBlock.java
This commit is contained in:
Jungtaek Lim
@@ -16,12 +16,12 @@ import redis.clients.util.SafeEncoder;
|
||||
|
||||
public abstract class JedisPubSub {
|
||||
private int subscribedChannels = 0;
|
||||
private Client client;
|
||||
private volatile Client client;
|
||||
|
||||
public abstract void onMessage(String channel, String message);
|
||||
|
||||
public abstract void onPMessage(String pattern, String channel,
|
||||
String message);
|
||||
String message);
|
||||
|
||||
public abstract void onSubscribe(String channel, int subscribedChannels);
|
||||
|
||||
@@ -32,115 +32,139 @@ public abstract class JedisPubSub {
|
||||
public abstract void onPSubscribe(String pattern, int subscribedChannels);
|
||||
|
||||
public void unsubscribe() {
|
||||
if (client == null) {
|
||||
throw new JedisConnectionException(
|
||||
"JedisPubSub was not subscribed to a Jedis instance.");
|
||||
}
|
||||
client.unsubscribe();
|
||||
client.flush();
|
||||
if (client == null) {
|
||||
throw new JedisConnectionException(
|
||||
"JedisPubSub was not subscribed to a Jedis instance.");
|
||||
}
|
||||
client.unsubscribe();
|
||||
client.flush();
|
||||
}
|
||||
|
||||
public void unsubscribe(String... channels) {
|
||||
client.unsubscribe(channels);
|
||||
client.flush();
|
||||
if (client == null) {
|
||||
throw new JedisConnectionException(
|
||||
"JedisPubSub is not subscribed to a Jedis instance.");
|
||||
}
|
||||
client.unsubscribe(channels);
|
||||
client.flush();
|
||||
}
|
||||
|
||||
public void subscribe(String... channels) {
|
||||
client.subscribe(channels);
|
||||
client.flush();
|
||||
if (client == null) {
|
||||
throw new JedisConnectionException(
|
||||
"JedisPubSub is not subscribed to a Jedis instance.");
|
||||
}
|
||||
client.subscribe(channels);
|
||||
client.flush();
|
||||
}
|
||||
|
||||
public void psubscribe(String... patterns) {
|
||||
client.psubscribe(patterns);
|
||||
client.flush();
|
||||
if (client == null) {
|
||||
throw new JedisConnectionException(
|
||||
"JedisPubSub is not subscribed to a Jedis instance.");
|
||||
}
|
||||
client.psubscribe(patterns);
|
||||
client.flush();
|
||||
}
|
||||
|
||||
public void punsubscribe() {
|
||||
client.punsubscribe();
|
||||
client.flush();
|
||||
if (client == null) {
|
||||
throw new JedisConnectionException(
|
||||
"JedisPubSub is not subscribed to a Jedis instance.");
|
||||
}
|
||||
client.punsubscribe();
|
||||
client.flush();
|
||||
}
|
||||
|
||||
public void punsubscribe(String... patterns) {
|
||||
client.punsubscribe(patterns);
|
||||
client.flush();
|
||||
if (client == null) {
|
||||
throw new JedisConnectionException(
|
||||
"JedisPubSub is not subscribed to a Jedis instance.");
|
||||
}
|
||||
client.punsubscribe(patterns);
|
||||
client.flush();
|
||||
}
|
||||
|
||||
public boolean isSubscribed() {
|
||||
return subscribedChannels > 0;
|
||||
return subscribedChannels > 0;
|
||||
}
|
||||
|
||||
public void proceedWithPatterns(Client client, String... patterns) {
|
||||
this.client = client;
|
||||
client.psubscribe(patterns);
|
||||
client.flush();
|
||||
process(client);
|
||||
this.client = client;
|
||||
client.psubscribe(patterns);
|
||||
client.flush();
|
||||
process(client);
|
||||
}
|
||||
|
||||
public void proceed(Client client, String... channels) {
|
||||
this.client = client;
|
||||
client.subscribe(channels);
|
||||
client.flush();
|
||||
process(client);
|
||||
this.client = client;
|
||||
client.subscribe(channels);
|
||||
client.flush();
|
||||
process(client);
|
||||
}
|
||||
|
||||
private void process(Client client) {
|
||||
do {
|
||||
List<Object> reply = client.getObjectMultiBulkReply();
|
||||
final Object firstObj = reply.get(0);
|
||||
if (!(firstObj instanceof byte[])) {
|
||||
throw new JedisException("Unknown message type: " + firstObj);
|
||||
}
|
||||
final byte[] resp = (byte[]) firstObj;
|
||||
if (Arrays.equals(SUBSCRIBE.raw, resp)) {
|
||||
subscribedChannels = ((Long) reply.get(2)).intValue();
|
||||
final byte[] bchannel = (byte[]) reply.get(1);
|
||||
final String strchannel = (bchannel == null) ? null
|
||||
: SafeEncoder.encode(bchannel);
|
||||
onSubscribe(strchannel, subscribedChannels);
|
||||
} else if (Arrays.equals(UNSUBSCRIBE.raw, resp)) {
|
||||
subscribedChannels = ((Long) reply.get(2)).intValue();
|
||||
final byte[] bchannel = (byte[]) reply.get(1);
|
||||
final String strchannel = (bchannel == null) ? null
|
||||
: SafeEncoder.encode(bchannel);
|
||||
onUnsubscribe(strchannel, subscribedChannels);
|
||||
} else if (Arrays.equals(MESSAGE.raw, resp)) {
|
||||
final byte[] bchannel = (byte[]) reply.get(1);
|
||||
final byte[] bmesg = (byte[]) reply.get(2);
|
||||
final String strchannel = (bchannel == null) ? null
|
||||
: SafeEncoder.encode(bchannel);
|
||||
final String strmesg = (bmesg == null) ? null : SafeEncoder
|
||||
.encode(bmesg);
|
||||
onMessage(strchannel, strmesg);
|
||||
} else if (Arrays.equals(PMESSAGE.raw, resp)) {
|
||||
final byte[] bpattern = (byte[]) reply.get(1);
|
||||
final byte[] bchannel = (byte[]) reply.get(2);
|
||||
final byte[] bmesg = (byte[]) reply.get(3);
|
||||
final String strpattern = (bpattern == null) ? null
|
||||
: SafeEncoder.encode(bpattern);
|
||||
final String strchannel = (bchannel == null) ? null
|
||||
: SafeEncoder.encode(bchannel);
|
||||
final String strmesg = (bmesg == null) ? null : SafeEncoder
|
||||
.encode(bmesg);
|
||||
onPMessage(strpattern, strchannel, strmesg);
|
||||
} else if (Arrays.equals(PSUBSCRIBE.raw, resp)) {
|
||||
subscribedChannels = ((Long) reply.get(2)).intValue();
|
||||
final byte[] bpattern = (byte[]) reply.get(1);
|
||||
final String strpattern = (bpattern == null) ? null
|
||||
: SafeEncoder.encode(bpattern);
|
||||
onPSubscribe(strpattern, subscribedChannels);
|
||||
} else if (Arrays.equals(PUNSUBSCRIBE.raw, resp)) {
|
||||
subscribedChannels = ((Long) reply.get(2)).intValue();
|
||||
final byte[] bpattern = (byte[]) reply.get(1);
|
||||
final String strpattern = (bpattern == null) ? null
|
||||
: SafeEncoder.encode(bpattern);
|
||||
onPUnsubscribe(strpattern, subscribedChannels);
|
||||
} else {
|
||||
throw new JedisException("Unknown message type: " + firstObj);
|
||||
}
|
||||
} while (isSubscribed());
|
||||
|
||||
do {
|
||||
List<Object> reply = client.getRawObjectMultiBulkReply();
|
||||
final Object firstObj = reply.get(0);
|
||||
if (!(firstObj instanceof byte[])) {
|
||||
throw new JedisException("Unknown message type: " + firstObj);
|
||||
}
|
||||
final byte[] resp = (byte[]) firstObj;
|
||||
if (Arrays.equals(SUBSCRIBE.raw, resp)) {
|
||||
subscribedChannels = ((Long) reply.get(2)).intValue();
|
||||
final byte[] bchannel = (byte[]) reply.get(1);
|
||||
final String strchannel = (bchannel == null) ? null
|
||||
: SafeEncoder.encode(bchannel);
|
||||
onSubscribe(strchannel, subscribedChannels);
|
||||
} else if (Arrays.equals(UNSUBSCRIBE.raw, resp)) {
|
||||
subscribedChannels = ((Long) reply.get(2)).intValue();
|
||||
final byte[] bchannel = (byte[]) reply.get(1);
|
||||
final String strchannel = (bchannel == null) ? null
|
||||
: SafeEncoder.encode(bchannel);
|
||||
onUnsubscribe(strchannel, subscribedChannels);
|
||||
} else if (Arrays.equals(MESSAGE.raw, resp)) {
|
||||
final byte[] bchannel = (byte[]) reply.get(1);
|
||||
final byte[] bmesg = (byte[]) reply.get(2);
|
||||
final String strchannel = (bchannel == null) ? null
|
||||
: SafeEncoder.encode(bchannel);
|
||||
final String strmesg = (bmesg == null) ? null : SafeEncoder
|
||||
.encode(bmesg);
|
||||
onMessage(strchannel, strmesg);
|
||||
} else if (Arrays.equals(PMESSAGE.raw, resp)) {
|
||||
final byte[] bpattern = (byte[]) reply.get(1);
|
||||
final byte[] bchannel = (byte[]) reply.get(2);
|
||||
final byte[] bmesg = (byte[]) reply.get(3);
|
||||
final String strpattern = (bpattern == null) ? null
|
||||
: SafeEncoder.encode(bpattern);
|
||||
final String strchannel = (bchannel == null) ? null
|
||||
: SafeEncoder.encode(bchannel);
|
||||
final String strmesg = (bmesg == null) ? null : SafeEncoder
|
||||
.encode(bmesg);
|
||||
onPMessage(strpattern, strchannel, strmesg);
|
||||
} else if (Arrays.equals(PSUBSCRIBE.raw, resp)) {
|
||||
subscribedChannels = ((Long) reply.get(2)).intValue();
|
||||
final byte[] bpattern = (byte[]) reply.get(1);
|
||||
final String strpattern = (bpattern == null) ? null
|
||||
: SafeEncoder.encode(bpattern);
|
||||
onPSubscribe(strpattern, subscribedChannels);
|
||||
} else if (Arrays.equals(PUNSUBSCRIBE.raw, resp)) {
|
||||
subscribedChannels = ((Long) reply.get(2)).intValue();
|
||||
final byte[] bpattern = (byte[]) reply.get(1);
|
||||
final String strpattern = (bpattern == null) ? null
|
||||
: SafeEncoder.encode(bpattern);
|
||||
onPUnsubscribe(strpattern, subscribedChannels);
|
||||
} else {
|
||||
throw new JedisException("Unknown message type: " + firstObj);
|
||||
}
|
||||
} while (isSubscribed());
|
||||
|
||||
/* Invalidate instance since this thread is no longer listening */
|
||||
this.client = null;
|
||||
}
|
||||
|
||||
public int getSubscribedChannels() {
|
||||
return subscribedChannels;
|
||||
return subscribedChannels;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user