-
Notifications
You must be signed in to change notification settings - Fork 58
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Java: Add the
XREADGROUP command (#1613)
* Java: Add the `XREADGROUP` command (#376) * Add XGROUP CreateConsumer, DelConsumer Signed-off-by: Andrew Carbonetto <[email protected]> * Add XREADGROUP command Signed-off-by: Andrew Carbonetto <[email protected]> * Udpate IT tests Signed-off-by: Andrew Carbonetto <[email protected]> * Fix IT tests Signed-off-by: Andrew Carbonetto <[email protected]> * SPOTLESS & merge conflict fix Signed-off-by: Andrew Carbonetto <[email protected]> * Update for review comments Signed-off-by: Andrew Carbonetto <[email protected]> --------- Signed-off-by: Andrew Carbonetto <[email protected]> * Remove old test from IT suite Signed-off-by: Andrew Carbonetto <[email protected]> * Update xreadgroup docs Signed-off-by: Andrew Carbonetto <[email protected]> --------- Signed-off-by: Andrew Carbonetto <[email protected]>
- Loading branch information
1 parent
499e2cf
commit 75d784e
Showing
11 changed files
with
584 additions
and
16 deletions.
There are no files selected for viewing
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
72 changes: 72 additions & 0 deletions
72
java/client/src/main/java/glide/api/models/commands/stream/StreamReadGroupOptions.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,72 @@ | ||
| /** Copyright GLIDE-for-Redis Project Contributors - SPDX Identifier: Apache-2.0 */ | ||
| package glide.api.models.commands.stream; | ||
|
|
||
| import glide.api.commands.StreamBaseCommands; | ||
| import java.util.ArrayList; | ||
| import java.util.List; | ||
| import java.util.Map; | ||
| import java.util.Set; | ||
| import java.util.stream.Collectors; | ||
| import lombok.experimental.SuperBuilder; | ||
|
|
||
| /** | ||
| * Optional arguments for {@link StreamBaseCommands#xreadgroup(Map, String, String, | ||
| * StreamReadGroupOptions)} | ||
| * | ||
| * @see <a href="https://valkey.io/commands/xreadgroup/">redis.io</a> | ||
| */ | ||
| @SuperBuilder | ||
| public final class StreamReadGroupOptions extends StreamReadOptions { | ||
|
|
||
| public static final String READ_GROUP_REDIS_API = "GROUP"; | ||
| public static final String READ_NOACK_REDIS_API = "NOACK"; | ||
|
|
||
| /** | ||
| * If set, messages are not added to the Pending Entries List (PEL). This is equivalent to | ||
| * acknowledging the message when it is read. | ||
| */ | ||
| private boolean noack; | ||
|
|
||
| public abstract static class StreamReadGroupOptionsBuilder< | ||
| C extends StreamReadGroupOptions, B extends StreamReadGroupOptionsBuilder<C, B>> | ||
| extends StreamReadOptions.StreamReadOptionsBuilder<C, B> { | ||
| public B noack() { | ||
| this.noack = true; | ||
| return self(); | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Converts options and the key-to-id input for {@link StreamBaseCommands#xreadgroup(Map, String, | ||
| * String, StreamReadGroupOptions)} into a String[]. | ||
| * | ||
| * @return String[] | ||
| */ | ||
| public String[] toArgs(String group, String consumer, Map<String, String> streams) { | ||
| List<String> optionArgs = new ArrayList<>(); | ||
| optionArgs.add(READ_GROUP_REDIS_API); | ||
| optionArgs.add(group); | ||
| optionArgs.add(consumer); | ||
|
|
||
| if (this.count != null) { | ||
| optionArgs.add(READ_COUNT_REDIS_API); | ||
| optionArgs.add(count.toString()); | ||
| } | ||
|
|
||
| if (this.block != null) { | ||
| optionArgs.add(READ_BLOCK_REDIS_API); | ||
| optionArgs.add(block.toString()); | ||
| } | ||
|
|
||
| if (this.noack) { | ||
| optionArgs.add(READ_NOACK_REDIS_API); | ||
| } | ||
|
|
||
| optionArgs.add(READ_STREAMS_REDIS_API); | ||
| Set<Map.Entry<String, String>> entrySet = streams.entrySet(); | ||
| optionArgs.addAll(entrySet.stream().map(Map.Entry::getKey).collect(Collectors.toList())); | ||
| optionArgs.addAll(entrySet.stream().map(Map.Entry::getValue).collect(Collectors.toList())); | ||
|
|
||
| return optionArgs.toArray(new String[0]); | ||
| } | ||
| } |
Oops, something went wrong.