diff --git a/CHANGELOG.md b/CHANGELOG.md index cb5d755dc6..0120bd9d22 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -24,6 +24,7 @@ * Node: Added ZMPOP command ([#1994](https://github.com/valkey-io/valkey-glide/pull/1994)) #### Fixes +* Java: Add overloads for XADD to allow duplicate entry keys ([#1970](https://github.com/valkey-io/valkey-glide/pull/1970)) * Node: Fix ZADD bug where command could not be called with only the `changed` optional parameter ([#1995](https://github.com/valkey-io/valkey-glide/pull/1995)) ## 1.0.0 (2024-07-09) diff --git a/java/client/src/main/java/glide/api/BaseClient.java b/java/client/src/main/java/glide/api/BaseClient.java index fceaf82634..d136a7be77 100644 --- a/java/client/src/main/java/glide/api/BaseClient.java +++ b/java/client/src/main/java/glide/api/BaseClient.java @@ -187,6 +187,8 @@ import static glide.utils.ArrayTransformUtils.convertMapToKeyValueStringArray; import static glide.utils.ArrayTransformUtils.convertMapToValueKeyStringArray; import static glide.utils.ArrayTransformUtils.convertMapToValueKeyStringArrayBinary; +import static glide.utils.ArrayTransformUtils.convertNestedArrayToKeyValueGlideStringArray; +import static glide.utils.ArrayTransformUtils.convertNestedArrayToKeyValueStringArray; import static glide.utils.ArrayTransformUtils.mapGeoDataToArray; import static glide.utils.ArrayTransformUtils.mapGeoDataToGlideStringArray; @@ -2588,12 +2590,23 @@ public CompletableFuture xadd(@NonNull String key, @NonNull Map xadd(@NonNull String key, @NonNull String[][] values) { + return xadd(key, values, StreamAddOptions.builder().build()); + } + @Override public CompletableFuture xadd( @NonNull GlideString key, @NonNull Map values) { return xadd(key, values, StreamAddOptionsBinary.builder().build()); } + @Override + public CompletableFuture xadd( + @NonNull GlideString key, @NonNull GlideString[][] values) { + return xadd(key, values, StreamAddOptionsBinary.builder().build()); + } + @Override public CompletableFuture xadd( @NonNull String key, @NonNull Map values, @NonNull StreamAddOptions options) { @@ -2603,6 +2616,16 @@ public CompletableFuture xadd( return commandManager.submitNewCommand(XAdd, arguments, this::handleStringOrNullResponse); } + @Override + public CompletableFuture xadd( + @NonNull String key, @NonNull String[][] values, @NonNull StreamAddOptions options) { + String[] arguments = + ArrayUtils.addAll( + ArrayUtils.addFirst(options.toArgs(), key), + convertNestedArrayToKeyValueStringArray(values)); + return commandManager.submitNewCommand(XAdd, arguments, this::handleStringOrNullResponse); + } + @Override public CompletableFuture xadd( @NonNull GlideString key, @@ -2618,6 +2641,21 @@ public CompletableFuture xadd( return commandManager.submitNewCommand(XAdd, arguments, this::handleGlideStringOrNullResponse); } + @Override + public CompletableFuture xadd( + @NonNull GlideString key, + @NonNull GlideString[][] values, + @NonNull StreamAddOptionsBinary options) { + GlideString[] arguments = + new ArgsBuilder() + .add(key) + .add(options.toArgs()) + .add(convertNestedArrayToKeyValueGlideStringArray(values)) + .toArray(); + + return commandManager.submitNewCommand(XAdd, arguments, this::handleGlideStringOrNullResponse); + } + @Override public CompletableFuture>> xread( @NonNull Map keysAndIds) { diff --git a/java/client/src/main/java/glide/api/commands/StreamBaseCommands.java b/java/client/src/main/java/glide/api/commands/StreamBaseCommands.java index c88bda1201..a62000c3c8 100644 --- a/java/client/src/main/java/glide/api/commands/StreamBaseCommands.java +++ b/java/client/src/main/java/glide/api/commands/StreamBaseCommands.java @@ -29,7 +29,8 @@ public interface StreamBaseCommands { /** * Adds an entry to the specified stream stored at key.
- * If the key doesn't exist, the stream is created. + * If the key doesn't exist, the stream is created. To add entries with duplicate + * keys, use {@link #xadd(String, String[][])}. * * @see valkey.io for details. * @param key The key of the stream. @@ -45,7 +46,25 @@ public interface StreamBaseCommands { /** * Adds an entry to the specified stream stored at key.
- * If the key doesn't exist, the stream is created. + * If the key doesn't exist, the stream is created. This method overload allows + * entries with duplicate keys to be added. + * + * @see valkey.io for details. + * @param key The key of the stream. + * @param values Field-value pairs to be added to the entry. + * @return The id of the added entry. + * @example + * 
{@code
+     * String streamId = client.xadd("key", new String[][] {{"name", "Sara"}, {"surname", "OConnor"}}).get();
+     * System.out.println("Stream: " + streamId);
+     * }
+ */ + CompletableFuture xadd(String key, String[][] values); + + /** + * Adds an entry to the specified stream stored at key.
+ * If the key doesn't exist, the stream is created. To add entries with duplicate + * keys, use {@link #xadd(GlideString, GlideString[][])}. * * @see valkey.io for details. * @param key The key of the stream. @@ -61,7 +80,25 @@ public interface StreamBaseCommands { /** * Adds an entry to the specified stream stored at key.
- * If the key doesn't exist, the stream is created. + * If the key doesn't exist, the stream is created. This method overload allows + * entries with duplicate keys to be added. + * + * @see valkey.io for details. + * @param key The key of the stream. + * @param values Field-value pairs to be added to the entry. + * @return The id of the added entry. + * @example + * 
{@code
+     * String streamId = client.xadd(gs("key"), new String[][] {{gs("name"), gs("Sara")}, {gs("surname"), gs("OConnor")}}).get();
+     * System.out.println("Stream: " + streamId);
+     * }
+ */ + CompletableFuture xadd(GlideString key, GlideString[][] values); + + /** + * Adds an entry to the specified stream stored at key.
+ * If the key doesn't exist, the stream is created. To add entries with duplicate + * keys, use {@link #xadd(String, String[][], StreamAddOptions)}. * * @see valkey.io for details. * @param key The key of the stream. @@ -73,10 +110,10 @@ public interface StreamBaseCommands { * @example * 
{@code
      * // Option to use the existing stream, or return null if the stream doesn't already exist at "key"
-     * StreamAddOptions options = StreamAddOptions.builder().id("sid").makeStream(Boolean.FALSE).build();
+     * StreamAddOptions options = StreamAddOptions.builder().id("1-0").makeStream(Boolean.FALSE).build();
      * String streamId = client.xadd("key", Map.of("name", "Sara", "surname", "OConnor"), options).get();
      * if (streamId != null) {
-     *     assert streamId.equals("sid");
+     *     assert streamId.equals("1-0");
      * }
      * }
*/ @@ -84,7 +121,32 @@ public interface StreamBaseCommands { /** * Adds an entry to the specified stream stored at key.
- * If the key doesn't exist, the stream is created. + * If the key doesn't exist, the stream is created. This method overload allows + * entries with duplicate keys to be added. + * + * @see valkey.io for details. + * @param key The key of the stream. + * @param values Field-value pairs to be added to the entry. + * @param options Stream add options {@link StreamAddOptions}. + * @return The id of the added entry, or null if {@link + * StreamAddOptionsBuilder#makeStream(Boolean)} is set to false and no stream + * with the matching key exists. + * @example + * 
{@code
+     * // Option to use the existing stream, or return null if the stream doesn't already exist at "key"
+     * StreamAddOptions options = StreamAddOptions.builder().id("1-0").makeStream(Boolean.FALSE).build();
+     * String streamId = client.xadd("key", new String[][] {{"name", "Sara"}, {"surname", "OConnor"}}, options).get();
+     * if (streamId != null) {
+     *     assert streamId.equals("1-0");
+     * }
+     * }
+ */ + CompletableFuture xadd(String key, String[][] values, StreamAddOptions options); + + /** + * Adds an entry to the specified stream stored at key.
+ * If the key doesn't exist, the stream is created. To add entries with duplicate + * keys, use {@link #xadd(GlideString, GlideString[][], StreamAddOptionsBinary)}. * * @see valkey.io for details. * @param key The key of the stream. @@ -96,16 +158,41 @@ public interface StreamBaseCommands { * @example * 
{@code
      * // Option to use the existing stream, or return null if the stream doesn't already exist at "key"
-     * StreamAddOptionsBinary options = StreamAddOptions.builder().id(gs("sid")).makeStream(Boolean.FALSE).build();
+     * StreamAddOptionsBinary options = StreamAddOptions.builder().id(gs("1-0")).makeStream(Boolean.FALSE).build();
      * String streamId = client.xadd(gs("key"), Map.of(gs("name"), gs("Sara"), gs("surname"), gs("OConnor")), options).get();
      * if (streamId != null) {
-     *     assert streamId.equals("sid");
+     *     assert streamId.equals("1-0");
      * }
      * }
*/ CompletableFuture xadd( GlideString key, Map values, StreamAddOptionsBinary options); + /** + * Adds an entry to the specified stream stored at key.
+ * If the key doesn't exist, the stream is created. This method overload allows + * entries with duplicate keys to be added. + * + * @see valkey.io for details. + * @param key The key of the stream. + * @param values Field-value pairs to be added to the entry. + * @param options Stream add options {@link StreamAddOptions}. + * @return The id of the added entry, or null if {@link + * StreamAddOptionsBinaryBuilder#makeStream(Boolean)} is set to false and no + * stream with the matching key exists. + * @example + * 
{@code
+     * // Option to use the existing stream, or return null if the stream doesn't already exist at "key"
+     * StreamAddOptionsBinary options = StreamAddOptions.builder().id(gs("1-0")).makeStream(Boolean.FALSE).build();
+     * String streamId = client.xadd(gs("key"), new GlideString[][] {{gs("name"), gs("Sara")}, {gs("surname"), gs("OConnor")}}, options).get();
+     * if (streamId != null) {
+     *     assert streamId.equals("1-0");
+     * }
+     * }
+ */ + CompletableFuture xadd( + GlideString key, GlideString[][] values, StreamAddOptionsBinary options); + /** * Reads entries from the given streams. * diff --git a/java/client/src/main/java/glide/api/models/BaseTransaction.java b/java/client/src/main/java/glide/api/models/BaseTransaction.java index ec2ab02b0c..6766617426 100644 --- a/java/client/src/main/java/glide/api/models/BaseTransaction.java +++ b/java/client/src/main/java/glide/api/models/BaseTransaction.java @@ -215,6 +215,7 @@ import static glide.utils.ArrayTransformUtils.flattenAllKeysFollowedByAllValues; import static glide.utils.ArrayTransformUtils.flattenMapToGlideStringArray; import static glide.utils.ArrayTransformUtils.flattenMapToGlideStringArrayValueFirst; +import static glide.utils.ArrayTransformUtils.flattenNestedArrayToGlideStringArray; import static glide.utils.ArrayTransformUtils.mapGeoDataToGlideStringArray; import command_request.CommandRequestOuterClass.Command; @@ -3343,7 +3344,8 @@ public T zinterWithScores( /** * Adds an entry to the specified stream stored at key.
- * If the key doesn't exist, the stream is created. + * If the key doesn't exist, the stream is created. To add entries with duplicate + * keys, use {@link #xadd(ArgType, ArgType[][])}. * * @implNote {@link ArgType} is limited to {@link String} or {@link GlideString}, any other type * will throw {@link IllegalArgumentException}. @@ -3358,7 +3360,24 @@ public T xadd(@NonNull ArgType key, @NonNull Map val /** * Adds an entry to the specified stream stored at key.
- * If the key doesn't exist, the stream is created. + * If the key doesn't exist, the stream is created. This method overload allows + * entries with duplicate keys to be added. + * + * @implNote {@link ArgType} is limited to {@link String} or {@link GlideString}, any other type + * will throw {@link IllegalArgumentException}. + * @see valkey.io for details. + * @param key The key of the stream. + * @param values Field-value pairs to be added to the entry. + * @return Command Response - The id of the added entry. + */ + public T xadd(@NonNull ArgType key, @NonNull ArgType[][] values) { + return xadd(key, values, StreamAddOptions.builder().build()); + } + + /** + * Adds an entry to the specified stream stored at key.
+ * If the key doesn't exist, the stream is created. To add entries with duplicate + * keys, use {@link #xadd(ArgType, ArgType[][], StreamAddOptions)}. * * @implNote {@link ArgType} is limited to {@link String} or {@link GlideString}, any other type * will throw {@link IllegalArgumentException}. @@ -3385,6 +3404,34 @@ public T xadd( return getThis(); } + /** + * Adds an entry to the specified stream stored at key.
+ * If the key doesn't exist, the stream is created. This method overload allows + * entries with duplicate keys to be added. + * + * @implNote {@link ArgType} is limited to {@link String} or {@link GlideString}, any other type + * will throw {@link IllegalArgumentException}. + * @see valkey.io for details. + * @param key The key of the stream. + * @param values Field-value pairs to be added to the entry. + * @param options Stream add options {@link StreamAddOptions}. + * @return Command Response - The id of the added entry, or null if {@link + * StreamAddOptionsBuilder#makeStream(Boolean)} is set to false and no stream + * with the matching key exists. + */ + public T xadd( + @NonNull ArgType key, @NonNull ArgType[][] values, @NonNull StreamAddOptions options) { + checkTypeOrThrow(key); + protobufTransaction.addCommands( + buildCommand( + XAdd, + newArgsBuilder() + .add(key) + .add(options.toArgs()) + .add(flattenNestedArrayToGlideStringArray(values)))); + return getThis(); + } + /** * Reads entries from the given streams. * diff --git a/java/client/src/main/java/glide/utils/ArrayTransformUtils.java b/java/client/src/main/java/glide/utils/ArrayTransformUtils.java index fccdac14f3..c89545faef 100644 --- a/java/client/src/main/java/glide/utils/ArrayTransformUtils.java +++ b/java/client/src/main/java/glide/utils/ArrayTransformUtils.java @@ -44,6 +44,44 @@ public static GlideString[] convertMapToKeyValueGlideStringArray(Map Stream.of(entry[0], entry[1])) + .toArray(String[]::new); + } + + /** + * Converts a nested array of GlideString keys and values of any type in to an array of + * GlideStrings with alternating keys and values. + * + * @param args Nested array of GlideString keys to values of any type to convert. + * @return Array of strings [key1, gs(value1.toString()), key2, gs(value2.toString()), ...]. + */ + public static GlideString[] convertNestedArrayToKeyValueGlideStringArray(GlideString[][] args) { + for (GlideString[] entry : args) { + if (entry.length != 2) { + throw new IllegalArgumentException( + "Array entry had the wrong length. Expected length 2 but got length " + entry.length); + } + } + return Arrays.stream(args) + .flatMap(entry -> Stream.of(entry[0], GlideString.gs(entry[1].toString()))) + .toArray(GlideString[]::new); + } + /** * Converts a map of string keys and values of any type into an array of strings with alternating * values and keys. @@ -250,6 +288,25 @@ public static GlideString[] flattenMapToGlideStringArray(Map args) { .toArray(GlideString[]::new); } + /** + * Converts a nested array of any type of keys and values in to an array of GlideString with + * alternating keys and values. + * + * @param args Nested array of keys to values of any type to convert. + * @return Array of GlideString [key1, value1, key2, value2, ...]. + */ + public static GlideString[] flattenNestedArrayToGlideStringArray(T[][] args) { + for (T[] entry : args) { + if (entry.length != 2) { + throw new IllegalArgumentException( + "Array entry had the wrong length. Expected length 2 but got length " + entry.length); + } + } + return Arrays.stream(args) + .flatMap(entry -> Stream.of(GlideString.of(entry[0]), GlideString.of(entry[1]))) + .toArray(GlideString[]::new); + } + /** * Converts a map of any type of keys and values in to an array of GlideString with alternating * values and keys. diff --git a/java/client/src/test/java/glide/api/GlideClientTest.java b/java/client/src/test/java/glide/api/GlideClientTest.java index 2727db7346..cb29738bbb 100644 --- a/java/client/src/test/java/glide/api/GlideClientTest.java +++ b/java/client/src/test/java/glide/api/GlideClientTest.java @@ -271,6 +271,8 @@ import static glide.utils.ArrayTransformUtils.convertMapToKeyValueStringArray; import static glide.utils.ArrayTransformUtils.convertMapToValueKeyStringArray; import static glide.utils.ArrayTransformUtils.convertMapToValueKeyStringArrayBinary; +import static glide.utils.ArrayTransformUtils.convertNestedArrayToKeyValueGlideStringArray; +import static glide.utils.ArrayTransformUtils.convertNestedArrayToKeyValueStringArray; import static org.junit.jupiter.api.Assertions.assertArrayEquals; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertInstanceOf; @@ -7174,6 +7176,59 @@ public void xadd_returns_success() { assertEquals(returnId, response.get()); } + @SneakyThrows + @Test + public void xadd_nested_array_returns_success() { + // setup + String key = "testKey"; + String[][] fieldValues = {{"testField1", "testValue1"}, {"testField2", "testValue2"}}; + String[] fieldValuesArgs = convertNestedArrayToKeyValueStringArray(fieldValues); + String[] arguments = new String[] {key, "*"}; + arguments = ArrayUtils.addAll(arguments, fieldValuesArgs); + String returnId = "testId"; + + CompletableFuture testResponse = new CompletableFuture<>(); + testResponse.complete(returnId); + + // match on protobuf request + when(commandManager.submitNewCommand(eq(XAdd), eq(arguments), any())) + .thenReturn(testResponse); + + // exercise + CompletableFuture response = service.xadd(key, fieldValues); + + // verify + assertEquals(testResponse, response); + assertEquals(returnId, response.get()); + } + + @SneakyThrows + @Test + public void xadd_nested_array_with_options_returns_success() { + // setup + String key = "testKey"; + String[][] fieldValues = {{"testField1", "testValue1"}, {"testField2", "testValue2"}}; + String[] fieldValuesArgs = convertNestedArrayToKeyValueStringArray(fieldValues); + String[] arguments = new String[] {key, "*"}; + arguments = ArrayUtils.addAll(arguments, fieldValuesArgs); + String returnId = "testId"; + StreamAddOptions options = StreamAddOptions.builder().build(); + + CompletableFuture testResponse = new CompletableFuture<>(); + testResponse.complete(returnId); + + // match on protobuf request + when(commandManager.submitNewCommand(eq(XAdd), eq(arguments), any())) + .thenReturn(testResponse); + + // exercise + CompletableFuture response = service.xadd(key, fieldValues, options); + + // verify + assertEquals(testResponse, response); + assertEquals(returnId, response.get()); + } + @SneakyThrows @Test public void xadd_binary_returns_success() { @@ -7202,6 +7257,63 @@ public void xadd_binary_returns_success() { assertEquals(returnId, response.get()); } + @SneakyThrows + @Test + public void xadd_nested_array_binary_returns_success() { + // setup + GlideString key = gs("testKey"); + GlideString[][] fieldValues = { + {gs("testField1"), gs("testValue1")}, {gs("testField2"), gs("testValue2")} + }; + GlideString[] fieldValuesArgs = convertNestedArrayToKeyValueGlideStringArray(fieldValues); + GlideString[] arguments = new GlideString[] {key, gs("*")}; + arguments = ArrayUtils.addAll(arguments, fieldValuesArgs); + GlideString returnId = gs("testId"); + + CompletableFuture testResponse = new CompletableFuture<>(); + testResponse.complete(returnId); + + // match on protobuf request + when(commandManager.submitNewCommand(eq(XAdd), eq(arguments), any())) + .thenReturn(testResponse); + + // exercise + CompletableFuture response = service.xadd(key, fieldValues); + + // verify + assertEquals(testResponse, response); + assertEquals(returnId, response.get()); + } + + @SneakyThrows + @Test + public void xadd_nested_array_with_options_binary_returns_success() { + // setup + GlideString key = gs("testKey"); + GlideString[][] fieldValues = { + {gs("testField1"), gs("testValue1")}, {gs("testField2"), gs("testValue2")} + }; + GlideString[] fieldValuesArgs = convertNestedArrayToKeyValueGlideStringArray(fieldValues); + GlideString[] arguments = new GlideString[] {key, gs("*")}; + arguments = ArrayUtils.addAll(arguments, fieldValuesArgs); + GlideString returnId = gs("testId"); + StreamAddOptionsBinary options = StreamAddOptionsBinary.builder().build(); + + CompletableFuture testResponse = new CompletableFuture<>(); + testResponse.complete(returnId); + + // match on protobuf request + when(commandManager.submitNewCommand(eq(XAdd), eq(arguments), any())) + .thenReturn(testResponse); + + // exercise + CompletableFuture response = service.xadd(key, fieldValues, options); + + // verify + assertEquals(testResponse, response); + assertEquals(returnId, response.get()); + } + private static List getStreamAddOptions() { return List.of( Arguments.of( diff --git a/java/client/src/test/java/glide/api/models/TransactionTests.java b/java/client/src/test/java/glide/api/models/TransactionTests.java index 00aed8661f..aa49a6849b 100644 --- a/java/client/src/test/java/glide/api/models/TransactionTests.java +++ b/java/client/src/test/java/glide/api/models/TransactionTests.java @@ -769,6 +769,15 @@ InfScoreBound.NEGATIVE_INFINITY, new ScoreBoundary(3, false), new Limit(1, 2)), transaction.xadd("key", Map.of("field1", "foo1"), StreamAddOptions.builder().id("id").build()); results.add(Pair.of(XAdd, buildArgs("key", "id", "field1", "foo1"))); + transaction.xadd("key", new String[][] {new String[] {"field1", "foo1"}}); + results.add(Pair.of(XAdd, buildArgs("key", "*", "field1", "foo1"))); + + transaction.xadd( + "key", + new String[][] {new String[] {"field1", "foo1"}}, + StreamAddOptions.builder().id("id").build()); + results.add(Pair.of(XAdd, buildArgs("key", "id", "field1", "foo1"))); + transaction.xtrim("key", new MinId(true, "id")); results.add( Pair.of(XTrim, buildArgs("key", TRIM_MINID_VALKEY_API, TRIM_EXACT_VALKEY_API, "id"))); diff --git a/java/integTest/src/test/java/glide/SharedCommandTests.java b/java/integTest/src/test/java/glide/SharedCommandTests.java index 3a93cb7939..a015d30e67 100644 --- a/java/integTest/src/test/java/glide/SharedCommandTests.java +++ b/java/integTest/src/test/java/glide/SharedCommandTests.java @@ -5769,6 +5769,112 @@ public void bzmpop_binary_timeout_check(BaseClient client) { } } + @SneakyThrows + @ParameterizedTest(autoCloseArguments = false) + @MethodSource("getClients") + public void xadd_duplicate_entry_keys(BaseClient client) { + String key = UUID.randomUUID().toString(); + String field = UUID.randomUUID().toString(); + String foo1 = "foo1"; + String bar1 = "bar1"; + + String[][] entry = new String[][] {{field, foo1}, {field, bar1}}; + String streamId = client.xadd(key, entry).get(); + // get everything from the stream + Map result = client.xrange(key, InfRangeBound.MIN, InfRangeBound.MAX).get(); + assertEquals(1, result.size()); + String[][] actualEntry = result.get(streamId); + assertDeepEquals(entry, actualEntry); + } + + @SneakyThrows + @ParameterizedTest(autoCloseArguments = false) + @MethodSource("getClients") + public void xadd_duplicate_entry_keys_with_options(BaseClient client) { + String key = UUID.randomUUID().toString(); + String field = UUID.randomUUID().toString(); + String foo1 = "foo1"; + String bar1 = "bar1"; + + String[][] entry = new String[][] {{field, foo1}, {field, bar1}}; + String streamId = client.xadd(key, entry, StreamAddOptions.builder().build()).get(); + // get everything from the stream + Map result = client.xrange(key, InfRangeBound.MIN, InfRangeBound.MAX).get(); + assertEquals(1, result.size()); + String[][] actualEntry = result.get(streamId); + assertDeepEquals(entry, actualEntry); + } + + @SneakyThrows + @ParameterizedTest(autoCloseArguments = false) + @MethodSource("getClients") + public void xadd_duplicate_entry_keys_binary(BaseClient client) { + GlideString key = gs(UUID.randomUUID().toString()); + GlideString field = gs(UUID.randomUUID().toString()); + GlideString foo1 = gs("foo1"); + GlideString bar1 = gs("bar1"); + + GlideString[][] entry = new GlideString[][] {{field, foo1}, {field, bar1}}; + GlideString streamId = client.xadd(key, entry).get(); + // get everything from the stream + Map result = + client.xrange(key, InfRangeBound.MIN, InfRangeBound.MAX).get(); + assertEquals(1, result.size()); + GlideString[][] actualEntry = result.get(streamId); + assertDeepEquals(entry, actualEntry); + } + + @SneakyThrows + @ParameterizedTest(autoCloseArguments = false) + @MethodSource("getClients") + public void xadd_duplicate_entry_keys_with_options_binary(BaseClient client) { + GlideString key = gs(UUID.randomUUID().toString()); + GlideString field = gs(UUID.randomUUID().toString()); + GlideString foo1 = gs("foo1"); + GlideString bar1 = gs("bar1"); + + GlideString[][] entry = new GlideString[][] {{field, foo1}, {field, bar1}}; + GlideString streamId = client.xadd(key, entry, StreamAddOptionsBinary.builder().build()).get(); + // get everything from the stream + Map result = + client.xrange(key, InfRangeBound.MIN, InfRangeBound.MAX).get(); + assertEquals(1, result.size()); + GlideString[][] actualEntry = result.get(streamId); + assertDeepEquals(entry, actualEntry); + } + + @SneakyThrows + @ParameterizedTest(autoCloseArguments = false) + @MethodSource("getClients") + public void xadd_wrong_length_entries(BaseClient client) { + String key = UUID.randomUUID().toString(); + String timestamp = "0-1"; + + // Entry too long + assertThrows( + IllegalArgumentException.class, + () -> + client + .xadd( + key, + new String[][] { + new String[] {"field1", "foo1"}, new String[] {"field2", "bar2", "oh no"} + }, + StreamAddOptions.builder().id(timestamp).build()) + .get()); + + // Entry too short + assertThrows( + IllegalArgumentException.class, + () -> + client + .xadd( + key, + new String[][] {new String[] {"field1", "foo1"}, new String[] {"oh no"}}, + StreamAddOptions.builder().id(timestamp).build()) + .get()); + } + @SneakyThrows @ParameterizedTest(autoCloseArguments = false) @MethodSource("getClients") diff --git a/java/integTest/src/test/java/glide/TransactionTestUtilities.java b/java/integTest/src/test/java/glide/TransactionTestUtilities.java index 110e71e29b..96b0c9f894 100644 --- a/java/integTest/src/test/java/glide/TransactionTestUtilities.java +++ b/java/integTest/src/test/java/glide/TransactionTestUtilities.java @@ -815,6 +815,7 @@ private static Object[] streamCommands(BaseTransaction transaction) { final String streamKey1 = "{streamKey}-1-" + UUID.randomUUID(); final String streamKey2 = "{streamKey}-2-" + UUID.randomUUID(); final String streamKey3 = "{streamKey}-3-" + UUID.randomUUID(); + final String streamKey4 = "{streamKey}-4-" + UUID.randomUUID(); final String groupName1 = "{groupName}-1-" + UUID.randomUUID(); final String groupName2 = "{groupName}-2-" + UUID.randomUUID(); final String groupName3 = "{groupName}-2-" + UUID.randomUUID(); @@ -824,6 +825,10 @@ private static Object[] streamCommands(BaseTransaction transaction) { .xadd(streamKey1, Map.of("field1", "value1"), StreamAddOptions.builder().id("0-1").build()) .xadd(streamKey1, Map.of("field2", "value2"), StreamAddOptions.builder().id("0-2").build()) .xadd(streamKey1, Map.of("field3", "value3"), StreamAddOptions.builder().id("0-3").build()) + .xadd( + streamKey4, + new String[][] {{"field4", "value4"}, {"field4", "value5"}}, + StreamAddOptions.builder().id("0-4").build()) .xlen(streamKey1) .xread(Map.of(streamKey1, "0-2")) .xread(Map.of(streamKey1, "0-2"), StreamReadOptions.builder().count(1L).build()) @@ -896,6 +901,8 @@ private static Object[] streamCommands(BaseTransaction transaction) { "0-1", // xadd(streamKey1, Map.of("field1", "value1"), ... .id("0-1").build()); "0-2", // xadd(streamKey1, Map.of("field2", "value2"), ... .id("0-2").build()); "0-3", // xadd(streamKey1, Map.of("field3", "value3"), ... .id("0-3").build()); + "0-4", // xadd(streamKey4, new String[][] {{"field4", "value4"}, {"field4", "value5"}}), + // ... .id("0-4").build()); 3L, // xlen(streamKey1) Map.of( streamKey1,