Support option CREATECONSUMER in XGROUP command

src/main/java/io/lettuce/core/AbstractRedisAsyncCommands.java

@@ -1636,6 +1636,11 @@ public RedisFuture<String> xgroupCreate(XReadArgs.StreamOffset<K> offset, K grou
         return dispatch(commandBuilder.xgroupCreate(offset, group, args));
     }
 
+    @Override
+    public RedisFuture<Boolean> xgroupCreateconsumer(K key, Consumer<K> consumer) {
+        return dispatch(commandBuilder.xgroupCreateconsumer(key, consumer));
+    }
+
     @Override
     public RedisFuture<Long> xgroupDelconsumer(K key, Consumer<K> consumer) {
         return dispatch(commandBuilder.xgroupDelconsumer(key, consumer));

src/main/java/io/lettuce/core/AbstractRedisReactiveCommands.java

@@ -1711,6 +1711,11 @@ public Mono<String> xgroupCreate(XReadArgs.StreamOffset<K> streamOffset, K group
         return createMono(() -> commandBuilder.xgroupCreate(streamOffset, group, args));
     }
 
+    @Override
+    public Mono<Boolean> xgroupCreateconsumer(K key, Consumer<K> consumer) {
+        return createMono(() -> commandBuilder.xgroupCreateconsumer(key, consumer));
+    }
+
     @Override
     public Mono<Long> xgroupDelconsumer(K key, Consumer<K> consumer) {
         return createMono(() -> commandBuilder.xgroupDelconsumer(key, consumer));

src/main/java/io/lettuce/core/RedisCommandBuilder.java

@@ -2314,6 +2314,16 @@ public Command<K, V, String> xgroupCreate(StreamOffset<K> offset, K group, XGrou
         return createCommand(XGROUP, new StatusOutput<>(codec), args);
     }
 
+    public Command<K, V, Boolean> xgroupCreateconsumer(K key, Consumer<K> consumer) {
+        notNullKey(key);
+        LettuceAssert.notNull(consumer, "Consumer " + MUST_NOT_BE_NULL);
+
+        CommandArgs<K, V> args = new CommandArgs<>(codec).add("CREATECONSUMER").addKey(key).addKey(consumer.getGroup())
+                .addKey(consumer.getName());
+
+        return createCommand(XGROUP, new BooleanOutput<>(codec), args);
+    }
+
     public Command<K, V, Long> xgroupDelconsumer(K key, Consumer<K> consumer) {
         notNullKey(key);
         LettuceAssert.notNull(consumer, "Consumer " + MUST_NOT_BE_NULL);

src/main/java/io/lettuce/core/api/async/RedisStreamAsyncCommands.java

@@ -137,6 +137,16 @@ public interface RedisStreamAsyncCommands<K, V> {
      */
     RedisFuture<String> xgroupCreate(StreamOffset<K> streamOffset, K group, XGroupCreateArgs args);
 
+    /**
+     * Create a consumer from a consumer group.
+     *
+     * @param key the stream key.
+     * @param consumer consumer identified by group name and consumer key.
+     * @return simple-reply {@code true} if successful.
+     * @since 6.1
+     */
+    RedisFuture<Boolean> xgroupCreateconsumer(K key, Consumer<K> consumer);
+
     /**
      * Delete a consumer from a consumer group.
      *

src/main/java/io/lettuce/core/api/reactive/RedisStreamReactiveCommands.java

@@ -138,6 +138,16 @@ public interface RedisStreamReactiveCommands<K, V> {
      */
     Mono<String> xgroupCreate(StreamOffset<K> streamOffset, K group, XGroupCreateArgs args);
 
+    /**
+     * Create a consumer from a consumer group.
+     *
+     * @param key the stream key.
+     * @param consumer consumer identified by group name and consumer key.
+     * @return simple-reply {@code true} if successful.
+     * @since 6.1
+     */
+    Mono<Boolean> xgroupCreateconsumer(K key, Consumer<K> consumer);
+
     /**
      * Delete a consumer from a consumer group.
      *

src/main/java/io/lettuce/core/api/sync/RedisStreamCommands.java

@@ -137,6 +137,16 @@ public interface RedisStreamCommands<K, V> {
      */
     String xgroupCreate(StreamOffset<K> streamOffset, K group, XGroupCreateArgs args);
 
+    /**
+     * Create a consumer from a consumer group.
+     *
+     * @param key the stream key.
+     * @param consumer consumer identified by group name and consumer key.
+     * @return simple-reply {@code true} if successful.
+     * @since 6.1
+     */
+    Boolean xgroupCreateconsumer(K key, Consumer<K> consumer);
+
     /**
      * Delete a consumer from a consumer group.
      *

src/main/java/io/lettuce/core/cluster/api/async/NodeSelectionStreamAsyncCommands.java

@@ -137,6 +137,16 @@ public interface NodeSelectionStreamAsyncCommands<K, V> {
      */
     AsyncExecutions<String> xgroupCreate(StreamOffset<K> streamOffset, K group, XGroupCreateArgs args);
 
+    /**
+     * Create a consumer from a consumer group.
+     *
+     * @param key the stream key.
+     * @param consumer consumer identified by group name and consumer key.
+     * @return simple-reply {@code true} if successful.
+     * @since 6.1
+     */
+    AsyncExecutions<Boolean> xgroupCreateconsumer(K key, Consumer<K> consumer);
+
     /**
      * Delete a consumer from a consumer group.
      *

src/main/java/io/lettuce/core/cluster/api/sync/NodeSelectionStreamCommands.java

@@ -137,6 +137,16 @@ public interface NodeSelectionStreamCommands<K, V> {
      */
     Executions<String> xgroupCreate(StreamOffset<K> streamOffset, K group, XGroupCreateArgs args);
 
+    /**
+     * Create a consumer from a consumer group.
+     *
+     * @param key the stream key.
+     * @param consumer consumer identified by group name and consumer key.
+     * @return simple-reply {@code true} if successful.
+     * @since 6.1
+     */
+    Executions<Boolean> xgroupCreateconsumer(K key, Consumer<K> consumer);
+
     /**
      * Delete a consumer from a consumer group.
      *

src/main/kotlin/io/lettuce/core/api/coroutines/RedisStreamCoroutinesCommands.kt

@@ -137,6 +137,16 @@ interface RedisStreamCoroutinesCommands<K : Any, V : Any> {
      */
     suspend fun xgroupCreate(streamOffset: StreamOffset<K>, group: K, args: XGroupCreateArgs): String?
 
+    /**
+     * Create a consumer from a consumer group.
+     *
+     * @param key the stream key.
+     * @param consumer consumer identified by group name and consumer key.
+     * @return simple-reply `true` if successful.
+     * @since 6.1
+     */
+    suspend fun xgroupCreateconsumer(key: K, consumer: Consumer<K>): Boolean?
+
     /**
      * Delete a consumer from a consumer group.
      *

src/main/kotlin/io/lettuce/core/api/coroutines/RedisStreamCoroutinesCommandsImpl.kt

@@ -57,6 +57,8 @@ internal class RedisStreamCoroutinesCommandsImpl<K : Any, V : Any>(internal val
 
     override suspend fun xgroupCreate(streamOffset: StreamOffset<K>, group: K, args: XGroupCreateArgs): String? = ops.xgroupCreate(streamOffset, group, args).awaitFirstOrNull()
 
+    override suspend fun xgroupCreateconsumer(key: K, consumer: Consumer<K>): Boolean? = ops.xgroupCreateconsumer(key, consumer).awaitFirstOrNull()
+
     override suspend fun xgroupDelconsumer(key: K, consumer: Consumer<K>): Long? = ops.xgroupDelconsumer(key, consumer).awaitFirstOrNull()
 
     override suspend fun xgroupDestroy(key: K, group: K): Boolean? = ops.xgroupDestroy(key, group).awaitFirstOrNull()

src/main/templates/io/lettuce/core/api/RedisStreamCommands.java

@@ -143,6 +143,16 @@ public interface RedisStreamCommands<K, V> {
      */
     String xgroupCreate(StreamOffset<K> streamOffset, K group, XGroupCreateArgs args);
 
+    /**
+     * Create a consumer from a consumer group.
+     *
+     * @param key the stream key.
+     * @param consumer consumer identified by group name and consumer key.
+     * @return simple-reply {@code true} if successful.
+     * @since 6.1
+     */
+    Boolean xgroupCreateconsumer(K key, Consumer<K> consumer);
+
     /**
      * Delete a consumer from a consumer group.
      *

src/test/java/io/lettuce/core/commands/StreamCommandIntegrationTests.java

@@ -331,6 +331,17 @@ void xgroupCreate() {
         assertThat(redis.type(key)).isEqualTo("stream");
     }
 
+    @Test
+    @EnabledOnCommand("LMOVE") // Redis 6.2
+    void xgroupCreateconsumer() {
+
+        redis.xgroupCreate(StreamOffset.latest(key), "group", XGroupCreateArgs.Builder.mkstream());
+        redis.xadd(key, Collections.singletonMap("key", "value"));
+
+        assertThat(redis.xgroupCreateconsumer(key, Consumer.from("group", "consumer1"))).isTrue();
+        assertThat(redis.xgroupCreateconsumer(key, Consumer.from("group", "consumer1"))).isFalse();
+    }
+
     @Test
     void xgroupread() {