From 1a8fde001f1da18bf07ba80974f07f4e9c08f30a Mon Sep 17 00:00:00 2001 From: shacharPash Date: Wed, 8 Feb 2023 18:52:07 +0200 Subject: [PATCH] implement h/s/zScanIterator Co-authored-by: Leibale Eidelman --- packages/client/lib/client/index.spec.ts | 181 +++++++++++++++++------ packages/client/lib/client/index.ts | 100 ++++++++++++- packages/test-utils/lib/dockers.ts | 2 +- 3 files changed, 225 insertions(+), 58 deletions(-) diff --git a/packages/client/lib/client/index.spec.ts b/packages/client/lib/client/index.spec.ts index fbe5558fa8..50a325908b 100644 --- a/packages/client/lib/client/index.spec.ts +++ b/packages/client/lib/client/index.spec.ts @@ -2,13 +2,16 @@ import { strict as assert } from 'assert'; import testUtils, { GLOBAL, waitTillBeenCalled } from '../test-utils'; import RedisClient, { RedisClientType } from '.'; import { RedisClientMultiCommandType } from './multi-command'; -import { RedisCommandArguments, RedisCommandRawReply, RedisModules, RedisFunctions, RedisScripts } from '../commands'; +import { RedisCommandArguments, RedisCommandRawReply, RedisModules, RedisFunctions, RedisScripts, ConvertArgumentType } from '../commands'; import { AbortError, ClientClosedError, ClientOfflineError, ConnectionTimeoutError, DisconnectsClientError, SocketClosedUnexpectedlyError, WatchError } from '../errors'; import { defineScript } from '../lua-script'; import { spy } from 'sinon'; import { once } from 'events'; import { ClientKillFilters } from '../commands/CLIENT_KILL'; import { promisify } from 'util'; +import { commandOptions } from '../../dist/lib/command-options'; +import { ZMember } from '../commands/generic-transformers'; + export const SQUARE_SCRIPT = defineScript({ SCRIPT: 'return ARGV[1] * ARGV[1];', @@ -639,10 +642,10 @@ describe('Client', () => { await client.mSet(args); const results = new Set(), - iteartor = client.scanIterator( + iterator = client.scanIterator( client.commandOptions({ returnBuffers: true }) ); - for await (const key of iteartor) { + for await (const key of iterator) { results.add(key); } @@ -650,65 +653,145 @@ describe('Client', () => { }, GLOBAL.SERVERS.OPEN); }); - testUtils.testWithClient('hScanIterator', async client => { - const hash: Record = {}; - for (let i = 0; i < 100; i++) { - hash[i.toString()] = i.toString(); - } + describe('hScanIterator', () => { + testUtils.testWithClient('strings', async client => { + const hash: Record = {}; + for (let i = 0; i < 100; i++) { + hash[i.toString()] = i.toString(); + } - await client.hSet('key', hash); + await client.hSet('key', hash); - const results: Record = {}; - for await (const { field, value } of client.hScanIterator('key')) { - results[field] = value; - } + const results: Record = {}; + for await (const { field, value } of client.hScanIterator('key')) { + results[field] = value; + } - assert.deepEqual(hash, results); - }, GLOBAL.SERVERS.OPEN); + assert.deepEqual(hash, results); + }, GLOBAL.SERVERS.OPEN); - testUtils.testWithClient('sScanIterator', async client => { - const members = new Set(); - for (let i = 0; i < 100; i++) { - members.add(i.toString()); - } + testUtils.testWithClient('buffers', async client => { + const hash = new Map(); + for (let i = 0; i < 100; i++) { + const buffer = Buffer.from([i]); + hash.set(buffer, buffer); + } - await client.sAdd('key', Array.from(members)); + await client.hSet('key', hash); - const results = new Set(); - for await (const key of client.sScanIterator('key')) { - results.add(key); - } + const results = new Map(), + iterator = client.hScanIterator( + client.commandOptions({ returnBuffers: true }), + 'key' + ); + for await (const { field, value } of iterator) { + results.set(field, value); + } - assert.deepEqual(members, results); - }, GLOBAL.SERVERS.OPEN); + assert.deepEqual(hash, results); + }, GLOBAL.SERVERS.OPEN); + }); - testUtils.testWithClient('zScanIterator', async client => { - const members = []; - for (let i = 0; i < 100; i++) { - members.push({ - score: 1, - value: i.toString() - }); - } + describe('sScanIterator', () => { + testUtils.testWithClient('strings', async client => { + const members = new Set(); + for (let i = 0; i < 100; i++) { + members.add(i.toString()); + } - await client.zAdd('key', members); + await client.sAdd('key', Array.from(members)); - const map = new Map(); - for await (const member of client.zScanIterator('key')) { - map.set(member.value, member.score); - } + const results = new Set(); + for await (const key of client.sScanIterator('key')) { + results.add(key); + } - type MemberTuple = [string, number]; + assert.deepEqual(members, results); + }, GLOBAL.SERVERS.OPEN); - function sort(a: MemberTuple, b: MemberTuple) { - return Number(b[0]) - Number(a[0]); - } + testUtils.testWithClient('buffers', async client => { + const members = new Set(); + for (let i = 0; i < 100; i++) { + members.add(Buffer.from([i])); + } - assert.deepEqual( - [...map.entries()].sort(sort), - members.map(member => [member.value, member.score]).sort(sort) - ); - }, GLOBAL.SERVERS.OPEN); + await client.sAdd('key', Array.from(members)); + + const results = new Set(), + iterator = client.sScanIterator( + client.commandOptions({ returnBuffers: true }), + 'key' + ); + for await (const key of iterator) { + results.add(key); + } + + assert.deepEqual(members, results); + }, GLOBAL.SERVERS.OPEN); + }); + + describe('zScanIterator', () => { + testUtils.testWithClient('strings', async client => { + const members: Array> = []; + for (let i = 0; i < 100; i++) { + members.push({ + score: i, + value: i.toString() + }); + } + + await client.zAdd('key', members); + + const map = new Map(); + for await (const member of client.zScanIterator('key')) { + map.set(member.value, member.score); + } + + type MemberTuple = [string, number]; + + function sort(a: MemberTuple, b: MemberTuple) { + return Number(b[0]) - Number(a[0]); + } + + assert.deepEqual( + [...map.entries()].sort(sort), + members.map(member => [member.value, member.score]).sort(sort) + ); + }, GLOBAL.SERVERS.OPEN); + + testUtils.testWithClient('buffers', async client => { + const members: Array> = []; + for (let i = 0; i < 100; i++) { + members.push({ + score: i, + value: Buffer.from([i]) + }); + } + + await client.zAdd('key', members); + + const map = new Map(), + iterator = client.zScanIterator( + client.commandOptions({ returnBuffers: true }), + 'key' + ); + + for await (const member of iterator) { + map.set(member.value, member.score); + } + + type MemberTuple = [Buffer, number]; + + function sort(a: MemberTuple, b: MemberTuple) { + return b[0][0] - a[0][0]; + } + + assert.deepEqual( + [...map.entries()].sort(sort), + members.map(member => [member.value, member.score]).sort(sort) + ); + }, GLOBAL.SERVERS.OPEN); + }); describe('PubSub', () => { testUtils.testWithClient('should be able to publish and subscribe to messages', async publisher => { diff --git a/packages/client/lib/client/index.ts b/packages/client/lib/client/index.ts index 85c6219102..e0c43e4ae9 100644 --- a/packages/client/lib/client/index.ts +++ b/packages/client/lib/client/index.ts @@ -1,5 +1,5 @@ import COMMANDS from './commands'; -import { RedisCommand, RedisCommandArguments, RedisCommandRawReply, RedisCommandReply, RedisFunctions, RedisModules, RedisExtensions, RedisScript, RedisScripts, RedisCommandSignature, ConvertArgumentType, RedisFunction, ExcludeMappedString, RedisCommands } from '../commands'; +import { RedisCommand, RedisCommandArguments, RedisCommandRawReply, RedisCommandReply, RedisFunctions, RedisModules, RedisExtensions, RedisScript, RedisScripts, RedisCommandSignature, ConvertArgumentType, RedisFunction, ExcludeMappedString, RedisCommands, RedisCommandArgument } from '../commands'; import RedisSocket, { RedisSocketOptions, RedisTlsSocketOptions } from './socket'; import RedisCommandsQueue, { PubSubListener, PubSubSubscribeCommands, PubSubUnsubscribeCommands, QueueCommandOptions } from './commands-queue'; import RedisClientMultiCommand, { RedisClientMultiCommandType } from './multi-command'; @@ -656,6 +656,20 @@ export default class RedisClient< return results; } + // #scanIterator>( + // commandOptions: T, + // options?: ScanCommandOptions + // ): AsyncIterable; + // #scanIterator( + // options?: ScanCommandOptions + // ): AsyncIterable; + // async* #scanIterator>( + // commandOptions?: T | ScanCommandOptions, + // options?: ScanCommandOptions + // ): AsyncIterable { + + // } + scanIterator>( commandOptions: T, options?: ScanCommandOptions @@ -674,7 +688,7 @@ export default class RedisClient< const scan = commandOptions ? (...args: Array) => (this as any).scan(commandOptions, ...args) : - (...args: Array) => (this as any).scan(...args); + (this as any).scan.bind(this); let cursor = 0; do { @@ -686,10 +700,33 @@ export default class RedisClient< } while (cursor !== 0); } - async* hScanIterator(key: string, options?: ScanOptions): AsyncIterable> { + hScanIterator>( + commandOptions: T, + key: RedisCommandArgument, + options?: ScanOptions + ): AsyncIterable>; + hScanIterator( + key: RedisCommandArgument, + options?: ScanOptions + ): AsyncIterable>; + async* hScanIterator>( + commandOptions?: T | RedisCommandArgument, + key?: RedisCommandArgument | ScanOptions, + options?: ScanOptions + ): AsyncIterable> { + if (!isCommandOptions(commandOptions)) { + options = key as ScanOptions | undefined; + key = commandOptions; + commandOptions = undefined; + } + + const hScan = commandOptions ? + (...args: Array) => (this as any).hScan(commandOptions, ...args) : + (this as any).hScan.bind(this); + let cursor = 0; do { - const reply = await (this as any).hScan(key, cursor, options); + const reply = await hScan(key, cursor, options); cursor = reply.cursor; for (const tuple of reply.tuples) { yield tuple; @@ -697,10 +734,34 @@ export default class RedisClient< } while (cursor !== 0); } - async* sScanIterator(key: string, options?: ScanOptions): AsyncIterable { + + sScanIterator>( + commandOptions: T, + key: RedisCommandArgument, + options?: ScanOptions + ): AsyncIterable; + sScanIterator( + key: RedisCommandArgument, + options?: ScanOptions + ): AsyncIterable; + async* sScanIterator>( + commandOptions?: T | RedisCommandArgument, + key?: RedisCommandArgument | ScanOptions, + options?: ScanOptions + ): AsyncIterable { + if (!isCommandOptions(commandOptions)) { + options = key as ScanOptions | undefined; + key = commandOptions; + commandOptions = undefined; + } + + const sScan = commandOptions ? + (...args: Array) => (this as any).sScan(commandOptions, ...args) : + (this as any).sScan.bind(this); + let cursor = 0; do { - const reply = await (this as any).sScan(key, cursor, options); + const reply = await sScan(key, cursor, options); cursor = reply.cursor; for (const member of reply.members) { yield member; @@ -708,10 +769,33 @@ export default class RedisClient< } while (cursor !== 0); } - async* zScanIterator(key: string, options?: ScanOptions): AsyncIterable> { + zScanIterator>( + commandOptions: T, + key: RedisCommandArgument, + options?: ScanOptions + ): AsyncIterable>; + zScanIterator( + key: RedisCommandArgument, + options?: ScanOptions + ): AsyncIterable>; + async* zScanIterator>( + commandOptions?: T | RedisCommandArgument, + key?: RedisCommandArgument | ScanOptions, + options?: ScanOptions + ): AsyncIterable> { + if (!isCommandOptions(commandOptions)) { + options = key as ScanOptions | undefined; + key = commandOptions; + commandOptions = undefined; + } + + const zScan = commandOptions ? + (...args: Array) => (this as any).zScan(commandOptions, ...args) : + (this as any).zScan.bind(this); + let cursor = 0; do { - const reply = await (this as any).zScan(key, cursor, options); + const reply = await zScan(key, cursor, options); cursor = reply.cursor; for (const member of reply.members) { yield member; diff --git a/packages/test-utils/lib/dockers.ts b/packages/test-utils/lib/dockers.ts index 8f0be95b09..c067913b6f 100644 --- a/packages/test-utils/lib/dockers.ts +++ b/packages/test-utils/lib/dockers.ts @@ -52,7 +52,7 @@ const DOCKER_FODLER_PATH = path.join(__dirname, '../docker'); async function spawnRedisServerDocker({ image, version }: RedisServerDockerConfig, serverArguments: Array): Promise { const port = (await portIterator.next()).value, { stdout, stderr } = await execAsync( - 'docker run -d --network host $(' + + `docker run -d -p ${port}:${port} host $(` + `docker build ${DOCKER_FODLER_PATH} -q ` + `--build-arg IMAGE=${image}:${version} ` + `--build-arg REDIS_ARGUMENTS="--save '' --port ${port.toString()} ${serverArguments.join(' ')}"` +