Implement the FT.CURSOR command

Add Integration tests for the FT.CURSOR

Add RESP2 tests for all aggragation integration tests

Author: tishun

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

@@ -1568,6 +1568,21 @@ public RedisFuture<SearchReply<K, V>> ftAggregate(K index, V query) {
         return dispatch(searchCommandBuilder.ftAggregate(index, query, null));
     }
 
+    @Override
+    public RedisFuture<SearchReply<K, V>> ftCursorread(K index, long cursorId, int count) {
+        return dispatch(searchCommandBuilder.ftCursorread(index, cursorId, count));
+    }
+
+    @Override
+    public RedisFuture<SearchReply<K, V>> ftCursorread(K index, long cursorId) {
+        return dispatch(searchCommandBuilder.ftCursorread(index, cursorId));
+    }
+
+    @Override
+    public RedisFuture<String> ftCursordel(K index, long cursorId) {
+        return dispatch(searchCommandBuilder.ftCursordel(index, cursorId));
+    }
+
     @Override
     public RedisFuture<List<Long>> jsonArrappend(K key, JsonPath jsonPath, JsonValue... values) {
         return dispatch(jsonCommandBuilder.jsonArrappend(key, jsonPath, values));

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

@@ -1603,6 +1603,11 @@ public Mono<String> ftCreate(K index, List<FieldArgs<K>> fieldArgs) {
         return createMono(() -> searchCommandBuilder.ftCreate(index, null, fieldArgs));
     }
 
+    @Override
+    public Mono<String> ftCursordel(K index, long cursorId) {
+        return createMono(() -> searchCommandBuilder.ftCursordel(index, cursorId));
+    }
+
     @Override
     public Mono<String> ftDropindex(K index, boolean deleteDocumentKeys) {
         return createMono(() -> searchCommandBuilder.ftDropindex(index, deleteDocumentKeys));
@@ -1633,6 +1638,16 @@ public Mono<SearchReply<K, V>> ftAggregate(K index, V query) {
         return createMono(() -> searchCommandBuilder.ftAggregate(index, query, null));
     }
 
+    @Override
+    public Mono<SearchReply<K, V>> ftCursorread(K index, long cursorId, int count) {
+        return createMono(() -> searchCommandBuilder.ftCursorread(index, cursorId, count));
+    }
+
+    @Override
+    public Mono<SearchReply<K, V>> ftCursorread(K index, long cursorId) {
+        return createMono(() -> searchCommandBuilder.ftCursorread(index, cursorId));
+    }
+
     @Override
     public Flux<Long> jsonArrappend(K key, JsonPath jsonPath, JsonValue... values) {
         return createDissolvingFlux(() -> jsonCommandBuilder.jsonArrappend(key, jsonPath, values));

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

@@ -109,6 +109,57 @@ public Command<K, V, SearchReply<K, V>> ftAggregate(K index, V query, AggregateA
         return createCommand(FT_AGGREGATE, new EncodedComplexOutput<>(codec, new SearchReplyParser<>(codec, null)), args);
     }
 
+    /**
+     * Read next results from an existing cursor.
+     *
+     * @param index the index name
+     * @param cursorId the cursor id
+     * @param count the number of results to read
+     * @return the result of the cursor read command
+     */
+    public Command<K, V, SearchReply<K, V>> ftCursorread(K index, long cursorId, int count) {
+        notNullKey(index);
+
+        CommandArgs<K, V> args = new CommandArgs<>(codec).add(CommandKeyword.READ).addKey(index);
+        args.add(cursorId);
+        args.add(CommandKeyword.COUNT);
+        args.add(count);
+
+        return createCommand(FT_CURSOR, new EncodedComplexOutput<>(codec, new SearchReplyParser<>(codec, null)), args);
+    }
+
+    /**
+     * Read next results from an existing cursor.
+     *
+     * @param index the index name
+     * @param cursorId the cursor id
+     * @return the result of the cursor read command
+     */
+    public Command<K, V, SearchReply<K, V>> ftCursorread(K index, long cursorId) {
+        notNullKey(index);
+
+        CommandArgs<K, V> args = new CommandArgs<>(codec).add(CommandKeyword.READ).addKey(index);
+        args.add(cursorId);
+
+        return createCommand(FT_CURSOR, new EncodedComplexOutput<>(codec, new SearchReplyParser<>(codec, null)), args);
+    }
+
+    /**
+     * Delete a cursor.
+     *
+     * @param index the index name
+     * @param cursorId the cursor id
+     * @return the result of the cursor delete command
+     */
+    public Command<K, V, String> ftCursordel(K index, long cursorId) {
+        notNullKey(index);
+
+        CommandArgs<K, V> args = new CommandArgs<>(codec).add(CommandKeyword.DEL).addKey(index);
+        args.add(cursorId);
+
+        return createCommand(FT_CURSOR, new StatusOutput<>(codec), args);
+    }
+
     /**
      * Drop the index with the given name.
      *

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

@@ -46,6 +46,7 @@
 import static io.lettuce.core.protocol.CommandKeyword.*;
 import static io.lettuce.core.protocol.CommandType.*;
 import static io.lettuce.core.protocol.CommandType.COPY;
+import static io.lettuce.core.protocol.CommandType.DEL;
 import static io.lettuce.core.protocol.CommandType.SAVE;
 
 /**