-
Notifications
You must be signed in to change notification settings - Fork 3.9k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
10 changed files
with
368 additions
and
30 deletions.
There are no files selected for viewing
9 changes: 9 additions & 0 deletions
9
src/main/java/redis/clients/jedis/BinaryJedisShardedPubSub.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,9 @@ | ||
package redis.clients.jedis; | ||
|
||
public abstract class BinaryJedisShardedPubSub extends JedisShardedPubSubBase<byte[]> { | ||
|
||
@Override | ||
protected final byte[] encode(byte[] raw) { | ||
return raw; | ||
} | ||
} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,11 @@ | ||
package redis.clients.jedis; | ||
|
||
import redis.clients.jedis.util.SafeEncoder; | ||
|
||
public abstract class JedisShardedPubSub extends JedisShardedPubSubBase<String> { | ||
|
||
@Override | ||
protected final String encode(byte[] raw) { | ||
return SafeEncoder.encode(raw); | ||
} | ||
} |
108 changes: 108 additions & 0 deletions
108
src/main/java/redis/clients/jedis/JedisShardedPubSubBase.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,108 @@ | ||
package redis.clients.jedis; | ||
|
||
import static redis.clients.jedis.Protocol.ResponseKeyword.*; | ||
|
||
import java.util.Arrays; | ||
import java.util.List; | ||
|
||
import redis.clients.jedis.Protocol.Command; | ||
import redis.clients.jedis.exceptions.JedisException; | ||
|
||
public abstract class JedisShardedPubSubBase<T> { | ||
|
||
private int subscribedChannels = 0; | ||
private volatile Connection client; | ||
|
||
public void onSMessage(T channel, T message) { | ||
} | ||
|
||
public void onSSubscribe(T channel, int subscribedChannels) { | ||
} | ||
|
||
public void onSUnsubscribe(T channel, int subscribedChannels) { | ||
} | ||
|
||
private void sendAndFlushCommand(Command command, T... args) { | ||
if (client == null) { | ||
throw new JedisException(getClass() + " is not connected to a Connection."); | ||
} | ||
CommandArguments cargs = new CommandArguments(command).addObjects(args); | ||
client.sendCommand(cargs); | ||
client.flush(); | ||
} | ||
|
||
public final void sunsubscribe() { | ||
sendAndFlushCommand(Command.SUNSUBSCRIBE); | ||
} | ||
|
||
public final void sunsubscribe(T... channels) { | ||
sendAndFlushCommand(Command.SUNSUBSCRIBE, channels); | ||
} | ||
|
||
public final void ssubscribe(T... channels) { | ||
sendAndFlushCommand(Command.SSUBSCRIBE, channels); | ||
} | ||
|
||
public final boolean isSubscribed() { | ||
return subscribedChannels > 0; | ||
} | ||
|
||
public final int getSubscribedChannels() { | ||
return subscribedChannels; | ||
} | ||
|
||
public final void proceed(Connection client, T... channels) { | ||
this.client = client; | ||
this.client.setTimeoutInfinite(); | ||
try { | ||
ssubscribe(channels); | ||
process(); | ||
} finally { | ||
this.client.rollbackTimeout(); | ||
} | ||
} | ||
|
||
protected abstract T encode(byte[] raw); | ||
|
||
// private void process(Client client) { | ||
private void process() { | ||
|
||
do { | ||
Object reply = client.getUnflushedObject(); | ||
|
||
if (reply instanceof List) { | ||
List<Object> listReply = (List<Object>) reply; | ||
final Object firstObj = listReply.get(0); | ||
if (!(firstObj instanceof byte[])) { | ||
throw new JedisException("Unknown message type: " + firstObj); | ||
} | ||
final byte[] resp = (byte[]) firstObj; | ||
if (Arrays.equals(SSUBSCRIBE.getRaw(), resp)) { | ||
subscribedChannels = ((Long) listReply.get(2)).intValue(); | ||
final byte[] bchannel = (byte[]) listReply.get(1); | ||
final T enchannel = (bchannel == null) ? null : encode(bchannel); | ||
onSSubscribe(enchannel, subscribedChannels); | ||
} else if (Arrays.equals(SUNSUBSCRIBE.getRaw(), resp)) { | ||
subscribedChannels = ((Long) listReply.get(2)).intValue(); | ||
final byte[] bchannel = (byte[]) listReply.get(1); | ||
final T enchannel = (bchannel == null) ? null : encode(bchannel); | ||
onSUnsubscribe(enchannel, subscribedChannels); | ||
} else if (Arrays.equals(SMESSAGE.getRaw(), resp)) { | ||
final byte[] bchannel = (byte[]) listReply.get(1); | ||
final byte[] bmesg = (byte[]) listReply.get(2); | ||
final T enchannel = (bchannel == null) ? null : encode(bchannel); | ||
final T enmesg = (bmesg == null) ? null : encode(bmesg); | ||
onSMessage(enchannel, enmesg); | ||
} else { | ||
System.out.println(redis.clients.jedis.util.SafeEncoder.encodeObject(resp)); | ||
throw new JedisException("Unknown message type: " + firstObj); | ||
} | ||
} else { | ||
throw new JedisException("Unknown message type: " + reply); | ||
} | ||
} while (isSubscribed()); | ||
|
||
// /* Invalidate instance since this thread is no longer listening */ | ||
// this.client = null; | ||
} | ||
} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.