close #2166 - add support for cursor api (#2217)

Author: leibale

packages/search/lib/commands/AGGREGATE.ts

@@ -126,6 +126,10 @@ export interface AggregateOptions {
     DIALECT?: number;
 }
 
+export const FIRST_KEY_INDEX = 1;
+
+export const IS_READ_ONLY = true;
+
 export function transformArguments(
     index: string,
     query: string,

packages/search/lib/commands/AGGREGATE_WITHCURSOR.spec.ts

@@ -0,0 +1,37 @@
+import { strict as assert } from 'assert';
+import testUtils, { GLOBAL } from '../test-utils';
+import { transformArguments } from './AGGREGATE_WITHCURSOR';
+import { SchemaFieldTypes } from '.';
+
+describe('AGGREGATE WITHCURSOR', () => {
+    describe('transformArguments', () => {
+        it('without options', () => {
+            assert.deepEqual(
+                transformArguments('index', '*'),
+                ['FT.AGGREGATE', 'index', '*', 'WITHCURSOR']
+            );
+        });
+
+        it('with COUNT', () => {
+            assert.deepEqual(
+                transformArguments('index', '*', { COUNT: 1 }),
+                ['FT.AGGREGATE', 'index', '*', 'WITHCURSOR', 'COUNT', '1']
+            );
+        });
+    });
+
+    testUtils.testWithClient('client.ft.aggregateWithCursor', async client => {
+        await client.ft.create('index', {
+            field: SchemaFieldTypes.NUMERIC
+        });
+
+        assert.deepEqual(
+            await client.ft.aggregateWithCursor('index', '*'),
+            {
+                total: 0,
+                results: [],
+                cursor: 0
+            }
+        );
+    }, GLOBAL.SERVERS.OPEN);
+});

packages/search/lib/commands/AGGREGATE_WITHCURSOR.ts

@@ -0,0 +1,44 @@
+import {
+    AggregateOptions,
+    AggregateRawReply,
+    AggregateReply,
+    transformArguments as transformAggregateArguments,
+    transformReply as transformAggregateReply
+} from './AGGREGATE';
+
+export { FIRST_KEY_INDEX, IS_READ_ONLY } from './AGGREGATE';
+
+interface AggregateWithCursorOptions extends AggregateOptions {
+    COUNT?: number;
+}
+
+export function transformArguments(
+    index: string,
+    query: string,
+    options?: AggregateWithCursorOptions
+) {
+    const args = transformAggregateArguments(index, query, options);
+
+    args.push('WITHCURSOR');
+    if (options?.COUNT) {
+        args.push('COUNT', options.COUNT.toString());
+    }
+
+    return args;
+}
+
+type AggregateWithCursorRawReply = [
+    result: AggregateRawReply,
+    cursor: number
+];
+
+interface AggregateWithCursorReply extends AggregateReply {
+    cursor: number;
+}
+
+export function transformReply(reply: AggregateWithCursorRawReply): AggregateWithCursorReply {
+    return {
+        ...transformAggregateReply(reply[0]),
+        cursor: reply[1]
+    };
+}

packages/search/lib/commands/CURSOR_DEL.spec.ts

@@ -0,0 +1,33 @@
+import { strict as assert } from 'assert';
+import { SchemaFieldTypes } from '.';
+import testUtils, { GLOBAL } from '../test-utils';
+import { transformArguments } from './CURSOR_DEL';
+
+describe('CURSOR DEL', () => {
+    it('transformArguments', () => {
+        assert.deepEqual(
+            transformArguments('index', 0),
+            ['FT.CURSOR', 'DEL', 'index', '0']
+        );
+    });
+
+    testUtils.testWithClient('client.ft.cursorDel', async client => {
+        const [ ,, { cursor } ] = await Promise.all([
+            client.ft.create('idx', {
+                field: {
+                    type: SchemaFieldTypes.TEXT
+                }
+            }),
+            client.hSet('key', 'field', 'value'),
+            client.ft.aggregateWithCursor('idx', '*', {
+                COUNT: 1
+            })
+        ]);
+
+
+        assert.equal(
+            await client.ft.cursorDel('idx', cursor),
+            'OK'
+        );
+    }, GLOBAL.SERVERS.OPEN);
+});

packages/search/lib/commands/CURSOR_DEL.ts

@@ -0,0 +1,14 @@
+import { RedisCommandArgument } from '@redis/client/dist/lib/commands';
+
+export const FIRST_KEY_INDEX = 1;
+
+export function transformArguments(index: RedisCommandArgument, cursorId: number) {
+    return [
+        'FT.CURSOR',
+        'DEL',
+        index,
+        cursorId.toString()
+    ];
+}
+
+export declare function transformReply(): 'OK';

packages/search/lib/commands/CURSOR_READ.spec.ts

@@ -0,0 +1,36 @@
+import { strict as assert } from 'assert';
+import { SchemaFieldTypes } from '.';
+import testUtils, { GLOBAL } from '../test-utils';
+import { transformArguments } from './CURSOR_READ';
+
+describe('CURSOR READ', () => {
+    it('transformArguments', () => {
+        assert.deepEqual(
+            transformArguments('index', 0),
+            ['FT.CURSOR', 'READ', 'index', '0']
+        );
+    });
+
+    testUtils.testWithClient('client.ft.cursorRead', async client => {
+        const [ ,, { cursor } ] = await Promise.all([
+            client.ft.create('idx', {
+                field: {
+                    type: SchemaFieldTypes.TEXT
+                }
+            }),
+            client.hSet('key', 'field', 'value'),
+            client.ft.aggregateWithCursor('idx', '*', {
+                COUNT: 1
+            })
+        ]);
+
+        assert.deepEqual(
+            await client.ft.cursorRead('idx', cursor),
+            {
+                total: 0,
+                results: [],
+                cursor: 0
+            }
+        );
+    }, GLOBAL.SERVERS.OPEN);
+});

packages/search/lib/commands/CURSOR_READ.ts

@@ -0,0 +1,19 @@
+import { RedisCommandArgument, RedisCommandArguments } from '@redis/client/dist/lib/commands';
+
+export const FIRST_KEY_INDEX = 1;
+
+export const IS_READ_ONLY = true;
+
+export function transformArguments(
+    index: RedisCommandArgument,
+    cursor: number
+): RedisCommandArguments {
+    return [
+        'FT.CURSOR',
+        'READ',
+        index,
+        cursor.toString()
+    ];
+}
+
+export { transformReply } from './AGGREGATE_WITHCURSOR';

packages/search/lib/commands/index.ts

@@ -1,12 +1,15 @@
 import * as _LIST from './_LIST';
 import * as ALTER from './ALTER';
+import * as AGGREGATE_WITHCURSOR from './AGGREGATE_WITHCURSOR';
 import * as AGGREGATE from './AGGREGATE';
 import * as ALIASADD from './ALIASADD';
 import * as ALIASDEL from './ALIASDEL';
 import * as ALIASUPDATE from './ALIASUPDATE';
 import * as CONFIG_GET from './CONFIG_GET';
 import * as CONFIG_SET from './CONFIG_SET';
 import * as CREATE from './CREATE';
+import * as CURSOR_DEL from './CURSOR_DEL';
+import * as CURSOR_READ from './CURSOR_READ';
 import * as DICTADD from './DICTADD';
 import * as DICTDEL from './DICTDEL';
 import * as DICTDUMP from './DICTDUMP';
@@ -37,6 +40,8 @@ export default {
     _list: _LIST,
     ALTER,
     alter: ALTER,
+    AGGREGATE_WITHCURSOR,
+    aggregateWithCursor: AGGREGATE_WITHCURSOR,
     AGGREGATE,
     aggregate: AGGREGATE,
     ALIASADD,
@@ -51,6 +56,10 @@ export default {
     configSet: CONFIG_SET,
     CREATE,
     create: CREATE,
+    CURSOR_DEL,
+    cursorDel: CURSOR_DEL,
+    CURSOR_READ,
+    cursorRead: CURSOR_READ,
     DICTADD,
     dictAdd: DICTADD,
     DICTDEL,