Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Support XINFO STREAM FULL #2746

Merged
merged 10 commits into from
Dec 31, 2021
125 changes: 125 additions & 0 deletions src/main/java/redis/clients/jedis/BuilderFactory.java
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,9 @@
import org.json.JSONObject;

import redis.clients.jedis.exceptions.JedisException;
import redis.clients.jedis.params.ConsumerFullInfo;
import redis.clients.jedis.resps.StreamFullInfo;
import redis.clients.jedis.resps.StreamGroupFullInfo;
import redis.clients.jedis.resps.LCSMatchResult.MatchedPosition;
import redis.clients.jedis.resps.LCSMatchResult.Position;
import redis.clients.jedis.resps.*;
Expand Down Expand Up @@ -1109,6 +1112,128 @@ public String toString() {
}
};

private static final Builder<List<ConsumerFullInfo>> STREAM_CONSUMER_FULL_INFO_LIST = new Builder<List<ConsumerFullInfo>>() {

final Map<String, Builder> mappingFunctions = createDecoderMap();

private Map<String, Builder> createDecoderMap() {

Map<String, Builder> tempMappingFunctions = new HashMap<>();
tempMappingFunctions.put(ConsumerFullInfo.NAME, STRING);
tempMappingFunctions.put(ConsumerFullInfo.SEEN_TIME, LONG);
tempMappingFunctions.put(ConsumerFullInfo.PEL_COUNT, LONG);
tempMappingFunctions.put(ConsumerFullInfo.PENDING, LONG_LIST);

return tempMappingFunctions;
}

@Override
@SuppressWarnings("unchecked")
public List<ConsumerFullInfo> build(Object data) {
if (null == data) {
return null;
}

List<ConsumerFullInfo> list = new ArrayList<>();
List<Object> streamsEntries = (List<Object>) data;

for (Object streamsEntry : streamsEntries) {
List<Object> consumerInfoList = (List<Object>) streamsEntry;
Iterator<Object> consumerInfoIterator = consumerInfoList.iterator();
ConsumerFullInfo consumerInfo = new ConsumerFullInfo(createMapFromDecodingFunctions(consumerInfoIterator, mappingFunctions));
list.add(consumerInfo);
}
return list;
}

@Override
public String toString() {
return "List<ConsumerInfo>";
zeekling marked this conversation as resolved.
Show resolved Hide resolved
}
};

private static final Builder<List<StreamGroupFullInfo>> STREAM_GROUP_FULL_INFO_LIST = new Builder<List<StreamGroupFullInfo>>() {

final Map<String, Builder> mappingFunctions = createDecoderMap();

private Map<String, Builder> createDecoderMap() {

Map<String, Builder> tempMappingFunctions = new HashMap<>();
tempMappingFunctions.put(StreamGroupFullInfo.NAME, STRING);
tempMappingFunctions.put(StreamGroupFullInfo.CONSUMERS, STREAM_CONSUMER_FULL_INFO_LIST);
tempMappingFunctions.put(StreamGroupFullInfo.PENDING, STRING_LIST);
tempMappingFunctions.put(StreamGroupFullInfo.LAST_DELIVERED, STREAM_ENTRY_ID);
tempMappingFunctions.put(StreamGroupFullInfo.PEL_COUNT, LONG);

return tempMappingFunctions;
}

@Override
@SuppressWarnings("unchecked")
public List<StreamGroupFullInfo> build(Object data) {
if (null == data) {
return null;
}

List<StreamGroupFullInfo> list = new ArrayList<>();
List<Object> streamsEntries = (List<Object>) data;

for (Object streamsEntry : streamsEntries) {

List<Object> groupInfo = (List<Object>) streamsEntry;

Iterator<Object> groupInfoIterator = groupInfo.iterator();

StreamGroupFullInfo groupFullInfo = new StreamGroupFullInfo(createMapFromDecodingFunctions(
groupInfoIterator, mappingFunctions));
list.add(groupFullInfo);

}
return list;
}

@Override
public String toString() {
return "List<StreamGroupFullInfo>";
}
};

public static final Builder<StreamFullInfo> STREAM_INFO_FULL = new Builder<StreamFullInfo>() {

final Map<String, Builder> mappingFunctions = createDecoderMap();

private Map<String, Builder> createDecoderMap() {

Map<String, Builder> tempMappingFunctions = new HashMap<>();
tempMappingFunctions.put(StreamFullInfo.LAST_GENERATED_ID, STREAM_ENTRY_ID);
tempMappingFunctions.put(StreamFullInfo.LENGTH, LONG);
tempMappingFunctions.put(StreamFullInfo.RADIX_TREE_KEYS, LONG);
tempMappingFunctions.put(StreamFullInfo.RADIX_TREE_NODES, LONG);
tempMappingFunctions.put(StreamFullInfo.GROUPS, STREAM_GROUP_FULL_INFO_LIST);
tempMappingFunctions.put(StreamFullInfo.ENTRIES, STREAM_ENTRY_LIST);

return tempMappingFunctions;
}

@Override
@SuppressWarnings("unchecked")
public StreamFullInfo build(Object data) {
if (null == data) {
return null;
}

List<Object> streamsEntries = (List<Object>) data;
Iterator<Object> iterator = streamsEntries.iterator();

return new StreamFullInfo(createMapFromDecodingFunctions(iterator, mappingFunctions));
}

@Override
public String toString() {
return "StreamFullInfo";
}
};

public static final Builder<StreamPendingSummary> STREAM_PENDING_SUMMARY = new Builder<StreamPendingSummary>() {
@Override
@SuppressWarnings("unchecked")
Expand Down
8 changes: 8 additions & 0 deletions src/main/java/redis/clients/jedis/CommandObjects.java
Original file line number Diff line number Diff line change
Expand Up @@ -2155,6 +2155,14 @@ public final CommandObject<Object> xinfoStream(byte[] key) {
return new CommandObject<>(commandArguments(XINFO).add(STREAM).key(key), BuilderFactory.RAW_OBJECT);
}

public final CommandObject<StreamFullInfo> xinfoStreamFull(byte[] key, int count) {
return new CommandObject<>(commandArguments(XINFO).add(STREAM).key(key).add(FULL).add(COUNT).add(count), BuilderFactory.STREAM_INFO_FULL);
}

public final CommandObject<StreamFullInfo> xinfoStreamFull(String key, int count) {
return new CommandObject<>(commandArguments(XINFO).add(STREAM).key(key).add(FULL).add(COUNT).add(count), BuilderFactory.STREAM_INFO_FULL);
}

public final CommandObject<List<Object>> xinfoGroup(byte[] key) {
return new CommandObject<>(commandArguments(XINFO).add(GROUPS).key(key), BuilderFactory.RAW_OBJECT_LIST);
}
Expand Down
25 changes: 25 additions & 0 deletions src/main/java/redis/clients/jedis/Jedis.java
Original file line number Diff line number Diff line change
Expand Up @@ -4458,6 +4458,31 @@ public Object xinfoStream(byte[] key) {
return connection.executeCommand(commandObjects.xinfoStream(key));
}

@Override
public StreamFullInfo xinfoStreamFull(byte[] key) {
checkIsInMultiOrPipeline();
return connection.executeCommand(commandObjects.xinfoStreamFull(key, 10));
zeekling marked this conversation as resolved.
Show resolved Hide resolved

}

@Override
public StreamFullInfo xinfoStreamFull(byte[] key, int count) {
checkIsInMultiOrPipeline();
return connection.executeCommand(commandObjects.xinfoStreamFull(key, count));
}

@Override
public StreamFullInfo xinfoStreamFull(String key) {
checkIsInMultiOrPipeline();
return connection.executeCommand(commandObjects.xinfoStreamFull(key, 10));
zeekling marked this conversation as resolved.
Show resolved Hide resolved
zeekling marked this conversation as resolved.
Show resolved Hide resolved
}

@Override
public StreamFullInfo xinfoStreamFull(String key, int count) {
checkIsInMultiOrPipeline();
return connection.executeCommand(commandObjects.xinfoStreamFull(key, count));
}

@Override
public List<Object> xinfoGroup(byte[] key) {
checkIsInMultiOrPipeline();
Expand Down
21 changes: 21 additions & 0 deletions src/main/java/redis/clients/jedis/MultiNodePipelineBase.java
Original file line number Diff line number Diff line change
Expand Up @@ -1341,6 +1341,27 @@ public Response<StreamInfo> xinfoStream(String key) {
return appendCommand(commandObjects.xinfoStream(key));
}

@Override
public Response<StreamFullInfo> xinfoStreamFull(byte[] key) {
return appendCommand(commandObjects.xinfoStreamFull(key, 10));

zeekling marked this conversation as resolved.
Show resolved Hide resolved
}

@Override
public Response<StreamFullInfo> xinfoStreamFull(byte[] key, int count) {
return appendCommand(commandObjects.xinfoStreamFull(key, count));
}

@Override
public Response<StreamFullInfo> xinfoStreamFull(String key) {
return appendCommand(commandObjects.xinfoStreamFull(key, 10));
}

@Override
public Response<StreamFullInfo> xinfoStreamFull(String key, int count) {
return appendCommand(commandObjects.xinfoStreamFull(key, count));
}

@Override
public Response<List<StreamGroupInfo>> xinfoGroup(String key) {
return appendCommand(commandObjects.xinfoGroup(key));
Expand Down
21 changes: 21 additions & 0 deletions src/main/java/redis/clients/jedis/Pipeline.java
Original file line number Diff line number Diff line change
Expand Up @@ -2577,6 +2577,27 @@ public Response<Object> xinfoStream(byte[] key) {
return appendCommand(commandObjects.xinfoStream(key));
}

@Override
public Response<StreamFullInfo> xinfoStreamFull(byte[] key) {
return appendCommand(commandObjects.xinfoStreamFull(key, 10));

zeekling marked this conversation as resolved.
Show resolved Hide resolved
}

@Override
public Response<StreamFullInfo> xinfoStreamFull(byte[] key, int count) {
return appendCommand(commandObjects.xinfoStreamFull(key, count));
}

@Override
public Response<StreamFullInfo> xinfoStreamFull(String key) {
return appendCommand(commandObjects.xinfoStreamFull(key, 10));
}

@Override
public Response<StreamFullInfo> xinfoStreamFull(String key, int count) {
return appendCommand(commandObjects.xinfoStreamFull(key, count));
}

@Override
public Response<List<Object>> xinfoGroup(byte[] key) {
return appendCommand(commandObjects.xinfoGroup(key));
Expand Down
2 changes: 1 addition & 1 deletion src/main/java/redis/clients/jedis/Protocol.java
Original file line number Diff line number Diff line change
Expand Up @@ -258,7 +258,7 @@ public static enum Keyword implements Rawable {
RETRYCOUNT, USAGE, SAMPLES, STREAM, GROUPS, CONSUMERS, SETUSER, GETUSER, DELUSER, WHOAMI, USERS,
CAT, GENPASS, LOG, INCR, SAVE, JUSTID, WITHVALUES, UNBLOCK, NOMKSTREAM, MINID, DB, ABSTTL, TO,
TIMEOUT, ABORT, NX, XX, EX, PX, EXAT, PXAT, CH, WITHCOORD, WITHDIST, WITHHASH, STOREDIST, COPY,
KEEPTTL, AUTH, AUTH2, INFO, CHANNELS, NUMPAT, NUMSUB, LCS, KEYS, STRINGS;
KEEPTTL, AUTH, AUTH2, INFO, CHANNELS, NUMPAT, NUMSUB, LCS, KEYS, STRINGS, FULL;

private final byte[] raw;

Expand Down
21 changes: 21 additions & 0 deletions src/main/java/redis/clients/jedis/TransactionBase.java
Original file line number Diff line number Diff line change
Expand Up @@ -1395,6 +1395,27 @@ public Response<StreamInfo> xinfoStream(String key) {
return appendCommand(commandObjects.xinfoStream(key));
}

@Override
public Response<StreamFullInfo> xinfoStreamFull(byte[] key) {
return appendCommand(commandObjects.xinfoStreamFull(key, 10));

zeekling marked this conversation as resolved.
Show resolved Hide resolved
}

@Override
public Response<StreamFullInfo> xinfoStreamFull(byte[] key, int count) {
return appendCommand(commandObjects.xinfoStreamFull(key, count));
}

@Override
public Response<StreamFullInfo> xinfoStreamFull(String key) {
return appendCommand(commandObjects.xinfoStreamFull(key, 10));
}

@Override
public Response<StreamFullInfo> xinfoStreamFull(String key, int count) {
return appendCommand(commandObjects.xinfoStreamFull(key, count));
}

@Override
public Response<List<StreamGroupInfo>> xinfoGroup(String key) {
return appendCommand(commandObjects.xinfoGroup(key));
Expand Down
20 changes: 20 additions & 0 deletions src/main/java/redis/clients/jedis/UnifiedJedis.java
Original file line number Diff line number Diff line change
Expand Up @@ -2458,6 +2458,26 @@ public StreamInfo xinfoStream(String key) {
return executeCommand(commandObjects.xinfoStream(key));
}

@Override
public StreamFullInfo xinfoStreamFull(byte[] key) {
return executeCommand(commandObjects.xinfoStreamFull(key, 10));
zeekling marked this conversation as resolved.
Show resolved Hide resolved
zeekling marked this conversation as resolved.
Show resolved Hide resolved
}

@Override
public StreamFullInfo xinfoStreamFull(byte[] key, int count) {
return executeCommand(commandObjects.xinfoStreamFull(key, count));
}

@Override
public StreamFullInfo xinfoStreamFull(String key) {
return executeCommand(commandObjects.xinfoStreamFull(key, 10));
zeekling marked this conversation as resolved.
Show resolved Hide resolved
sazzad16 marked this conversation as resolved.
Show resolved Hide resolved
}

@Override
public StreamFullInfo xinfoStreamFull(String key, int count) {
return executeCommand(commandObjects.xinfoStreamFull(key, count));
}

@Override
public List<StreamGroupInfo> xinfoGroup(String key) {
return executeCommand(commandObjects.xinfoGroup(key));
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@
import java.util.Map;

import redis.clients.jedis.params.*;
import redis.clients.jedis.resps.StreamFullInfo;

public interface StreamBinaryCommands {

Expand Down Expand Up @@ -57,6 +58,14 @@ List<Object> xautoclaimJustId(byte[] key, byte[] groupName, byte[] consumerName,

Object xinfoStream(byte[] key);

StreamFullInfo xinfoStreamFull(byte[] key);

StreamFullInfo xinfoStreamFull(byte[] key, int count);

StreamFullInfo xinfoStreamFull(String key);

StreamFullInfo xinfoStreamFull(String key, int count);
zeekling marked this conversation as resolved.
Show resolved Hide resolved

List<Object> xinfoGroup(byte[] key);

List<Object> xinfoConsumers(byte[] key, byte[] group);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@

import redis.clients.jedis.Response;
import redis.clients.jedis.params.*;
import redis.clients.jedis.resps.StreamFullInfo;

public interface StreamPipelineBinaryCommands {
//
Expand Down Expand Up @@ -60,6 +61,14 @@ Response<List<Object>> xautoclaimJustId(byte[] key, byte[] groupName, byte[] con

Response<Object> xinfoStream(byte[] key);

Response<StreamFullInfo> xinfoStreamFull(byte[] key);

Response<StreamFullInfo> xinfoStreamFull(byte[] key, int count);

Response<StreamFullInfo> xinfoStreamFull(String key);

Response<StreamFullInfo> xinfoStreamFull(String key, int count);
zeekling marked this conversation as resolved.
Show resolved Hide resolved

Response<List<Object>> xinfoGroup(byte[] key);

Response<List<Object>> xinfoConsumers(byte[] key, byte[] group);
Expand Down