redis/node-redis
1
0
Fork 0
mirror of https://github.com/redis/node-redis.git synced 2025-08-07 13:22:56 +03:00
Code Activity

wip

Browse Source
This commit is contained in:
Leibale
2023-04-26 18:04:27 -04:00
parent c78f3354e0
commit 1254ff7ebd
5 changed files with 1343 additions and 1343 deletions
Download Patch File Download Diff File Expand all files Collapse all files

346
packages/client/lib/RESP/decoder.spec.ts
View File

@@ -1,195 +1,195 @@
import { strict as assert } from 'assert'; // import { strict as assert } from 'assert';
import { SinonSpy, spy } from 'sinon'; // import { SinonSpy, spy } from 'sinon';
import RESP2Decoder from './decoder'; // import RESP2Decoder from './decoder';
import { ErrorReply } from '../../errors'; // import { ErrorReply } from '../../errors';
interface DecoderAndSpies { // interface DecoderAndSpies {
decoder: RESP2Decoder; // decoder: RESP2Decoder;
returnStringsAsBuffersSpy: SinonSpy; // returnStringsAsBuffersSpy: SinonSpy;
onReplySpy: SinonSpy; // onReplySpy: SinonSpy;
} // }
function createDecoderAndSpies(returnStringsAsBuffers: boolean): DecoderAndSpies { // function createDecoderAndSpies(returnStringsAsBuffers: boolean): DecoderAndSpies {
const returnStringsAsBuffersSpy = spy(() => returnStringsAsBuffers), // const returnStringsAsBuffersSpy = spy(() => returnStringsAsBuffers),
onReplySpy = spy(); // onReplySpy = spy();
return { // return {
decoder: new RESP2Decoder({ // decoder: new RESP2Decoder({
returnStringsAsBuffers: returnStringsAsBuffersSpy, // returnStringsAsBuffers: returnStringsAsBuffersSpy,
onReply: onReplySpy // onReply: onReplySpy
}), // }),
returnStringsAsBuffersSpy, // returnStringsAsBuffersSpy,
onReplySpy // onReplySpy
}; // };
} // }
function writeChunks(stream: RESP2Decoder, buffer: Buffer) { // function writeChunks(stream: RESP2Decoder, buffer: Buffer) {
let i = 0; // let i = 0;
while (i < buffer.length) { // while (i < buffer.length) {
stream.write(buffer.slice(i, ++i)); // stream.write(buffer.slice(i, ++i));
} // }
} // }
type Replies = Array<Array<unknown>>; // type Replies = Array<Array<unknown>>;
interface TestsOptions { // interface TestsOptions {
toWrite: Buffer; // toWrite: Buffer;
returnStringsAsBuffers: boolean; // returnStringsAsBuffers: boolean;
replies: Replies; // replies: Replies;
} // }
function generateTests({ // function generateTests({
toWrite, // toWrite,
returnStringsAsBuffers, // returnStringsAsBuffers,
replies // replies
}: TestsOptions): void { // }: TestsOptions): void {
it('single chunk', () => { // it('single chunk', () => {
const { decoder, returnStringsAsBuffersSpy, onReplySpy } = // const { decoder, returnStringsAsBuffersSpy, onReplySpy } =
createDecoderAndSpies(returnStringsAsBuffers); // createDecoderAndSpies(returnStringsAsBuffers);
decoder.write(toWrite); // decoder.write(toWrite);
assert.equal(returnStringsAsBuffersSpy.callCount, replies.length); // assert.equal(returnStringsAsBuffersSpy.callCount, replies.length);
testReplies(onReplySpy, replies); // testReplies(onReplySpy, replies);
}); // });
it('multiple chunks', () => { // it('multiple chunks', () => {
const { decoder, returnStringsAsBuffersSpy, onReplySpy } = // const { decoder, returnStringsAsBuffersSpy, onReplySpy } =
createDecoderAndSpies(returnStringsAsBuffers); // createDecoderAndSpies(returnStringsAsBuffers);
writeChunks(decoder, toWrite); // writeChunks(decoder, toWrite);
assert.equal(returnStringsAsBuffersSpy.callCount, replies.length); // assert.equal(returnStringsAsBuffersSpy.callCount, replies.length);
testReplies(onReplySpy, replies); // testReplies(onReplySpy, replies);
}); // });
} // }
function testReplies(spy: SinonSpy, replies: Replies): void { // function testReplies(spy: SinonSpy, replies: Replies): void {
if (!replies) { // if (!replies) {
assert.equal(spy.callCount, 0); // assert.equal(spy.callCount, 0);
return; // return;
} // }
assert.equal(spy.callCount, replies.length); // assert.equal(spy.callCount, replies.length);
for (const [i, reply] of replies.entries()) { // for (const [i, reply] of replies.entries()) {
assert.deepEqual( // assert.deepEqual(
spy.getCall(i).args, // spy.getCall(i).args,
reply // reply
); // );
} // }
} // }
describe('RESP2Parser', () => { // describe('RESP2Parser', () => {
describe('Simple String', () => { // describe('Simple String', () => {
describe('as strings', () => { // describe('as strings', () => {
generateTests({ // generateTests({
toWrite: Buffer.from('+OK\r\n'), // toWrite: Buffer.from('+OK\r\n'),
returnStringsAsBuffers: false, // returnStringsAsBuffers: false,
replies: [['OK']] // replies: [['OK']]
}); // });
}); // });
describe('as buffers', () => { // describe('as buffers', () => {
generateTests({ // generateTests({
toWrite: Buffer.from('+OK\r\n'), // toWrite: Buffer.from('+OK\r\n'),
returnStringsAsBuffers: true, // returnStringsAsBuffers: true,
replies: [[Buffer.from('OK')]] // replies: [[Buffer.from('OK')]]
}); // });
}); // });
}); // });
describe('Error', () => { // describe('Error', () => {
generateTests({ // generateTests({
toWrite: Buffer.from('-ERR\r\n'), // toWrite: Buffer.from('-ERR\r\n'),
returnStringsAsBuffers: false, // returnStringsAsBuffers: false,
replies: [[new ErrorReply('ERR')]] // replies: [[new ErrorReply('ERR')]]
}); // });
}); // });
describe('Integer', () => { // describe('Integer', () => {
describe('-1', () => { // describe('-1', () => {
generateTests({ // generateTests({
toWrite: Buffer.from(':-1\r\n'), // toWrite: Buffer.from(':-1\r\n'),
returnStringsAsBuffers: false, // returnStringsAsBuffers: false,
replies: [[-1]] // replies: [[-1]]
}); // });
}); // });
describe('0', () => { // describe('0', () => {
generateTests({ // generateTests({
toWrite: Buffer.from(':0\r\n'), // toWrite: Buffer.from(':0\r\n'),
returnStringsAsBuffers: false, // returnStringsAsBuffers: false,
replies: [[0]] // replies: [[0]]
}); // });
}); // });
}); // });
describe('Bulk String', () => { // describe('Bulk String', () => {
describe('null', () => { // describe('null', () => {
generateTests({ // generateTests({
toWrite: Buffer.from('$-1\r\n'), // toWrite: Buffer.from('$-1\r\n'),
returnStringsAsBuffers: false, // returnStringsAsBuffers: false,
replies: [[null]] // replies: [[null]]
}); // });
}); // });
describe('as strings', () => { // describe('as strings', () => {
generateTests({ // generateTests({
toWrite: Buffer.from('$2\r\naa\r\n'), // toWrite: Buffer.from('$2\r\naa\r\n'),
returnStringsAsBuffers: false, // returnStringsAsBuffers: false,
replies: [['aa']] // replies: [['aa']]
}); // });
}); // });
describe('as buffers', () => { // describe('as buffers', () => {
generateTests({ // generateTests({
toWrite: Buffer.from('$2\r\naa\r\n'), // toWrite: Buffer.from('$2\r\naa\r\n'),
returnStringsAsBuffers: true, // returnStringsAsBuffers: true,
replies: [[Buffer.from('aa')]] // replies: [[Buffer.from('aa')]]
}); // });
}); // });
}); // });
describe('Array', () => { // describe('Array', () => {
describe('null', () => { // describe('null', () => {
generateTests({ // generateTests({
toWrite: Buffer.from('*-1\r\n'), // toWrite: Buffer.from('*-1\r\n'),
returnStringsAsBuffers: false, // returnStringsAsBuffers: false,
replies: [[null]] // replies: [[null]]
}); // });
}); // });
const arrayBuffer = Buffer.from( // const arrayBuffer = Buffer.from(
'*5\r\n' + // '*5\r\n' +
'+OK\r\n' + // '+OK\r\n' +
'-ERR\r\n' + // '-ERR\r\n' +
':0\r\n' + // ':0\r\n' +
'$1\r\na\r\n' + // '$1\r\na\r\n' +
'*0\r\n' // '*0\r\n'
); // );
describe('as strings', () => { // describe('as strings', () => {
generateTests({ // generateTests({
toWrite: arrayBuffer, // toWrite: arrayBuffer,
returnStringsAsBuffers: false, // returnStringsAsBuffers: false,
replies: [[[ // replies: [[[
'OK', // 'OK',
new ErrorReply('ERR'), // new ErrorReply('ERR'),
0, // 0,
'a', // 'a',
[] // []
]]] // ]]]
}); // });
}); // });
describe('as buffers', () => { // describe('as buffers', () => {
generateTests({ // generateTests({
toWrite: arrayBuffer, // toWrite: arrayBuffer,
returnStringsAsBuffers: true, // returnStringsAsBuffers: true,
replies: [[[ // replies: [[[
Buffer.from('OK'), // Buffer.from('OK'),
new ErrorReply('ERR'), // new ErrorReply('ERR'),
0, // 0,
Buffer.from('a'), // Buffer.from('a'),
[] // []
]]] // ]]]
}); // });
}); // });
}); // });
}); // });

1962
packages/client/lib/client/index.spec.ts
View File

File diff suppressed because it is too large Load Diff

604
packages/client/lib/cluster/index.spec.ts
View File

@@ -1,362 +1,362 @@
import { strict as assert } from 'assert'; // import { strict as assert } from 'assert';
import testUtils, { GLOBAL, waitTillBeenCalled } from '../test-utils'; // import testUtils, { GLOBAL, waitTillBeenCalled } from '../test-utils';
import RedisCluster from '.'; // import RedisCluster from '.';
import { ClusterSlotStates } from '../commands/CLUSTER_SETSLOT'; // import { ClusterSlotStates } from '../commands/CLUSTER_SETSLOT';
import { commandOptions } from '../command-options'; // import { commandOptions } from '../command-options';
import { SQUARE_SCRIPT } from '../client/index.spec'; // import { SQUARE_SCRIPT } from '../client/index.spec';
import { RootNodesUnavailableError } from '../errors'; // import { RootNodesUnavailableError } from '../errors';
import { spy } from 'sinon'; // import { spy } from 'sinon';
import { promiseTimeout } from '../utils'; // import { promiseTimeout } from '../utils';
import RedisClient from '../client'; // import RedisClient from '../client';
describe('Cluster', () => { // describe('Cluster', () => {
testUtils.testWithCluster('sendCommand', async cluster => { // testUtils.testWithCluster('sendCommand', async cluster => {
assert.equal( // assert.equal(
await cluster.sendCommand(undefined, true, ['PING']), // await cluster.sendCommand(undefined, true, ['PING']),
'PONG' // 'PONG'
); // );
}, GLOBAL.CLUSTERS.OPEN); // }, GLOBAL.CLUSTERS.OPEN);
testUtils.testWithCluster('isOpen', async cluster => { // testUtils.testWithCluster('isOpen', async cluster => {
assert.equal(cluster.isOpen, true); // assert.equal(cluster.isOpen, true);
await cluster.disconnect(); // await cluster.disconnect();
assert.equal(cluster.isOpen, false); // assert.equal(cluster.isOpen, false);
}, GLOBAL.CLUSTERS.OPEN); // }, GLOBAL.CLUSTERS.OPEN);
testUtils.testWithCluster('connect should throw if already connected', async cluster => { // testUtils.testWithCluster('connect should throw if already connected', async cluster => {
await assert.rejects(cluster.connect()); // await assert.rejects(cluster.connect());
}, GLOBAL.CLUSTERS.OPEN); // }, GLOBAL.CLUSTERS.OPEN);
testUtils.testWithCluster('multi', async cluster => { // testUtils.testWithCluster('multi', async cluster => {
const key = 'key'; // const key = 'key';
assert.deepEqual( // assert.deepEqual(
await cluster.multi() // await cluster.multi()
.set(key, 'value') // .set(key, 'value')
.get(key) // .get(key)
.exec(), // .exec(),
['OK', 'value'] // ['OK', 'value']
); // );
}, GLOBAL.CLUSTERS.OPEN); // }, GLOBAL.CLUSTERS.OPEN);
testUtils.testWithCluster('scripts', async cluster => { // testUtils.testWithCluster('scripts', async cluster => {
assert.equal( // assert.equal(
await cluster.square(2), // await cluster.square(2),
4 // 4
); // );
}, { // }, {
...GLOBAL.CLUSTERS.OPEN, // ...GLOBAL.CLUSTERS.OPEN,
clusterConfiguration: { // clusterConfiguration: {
scripts: { // scripts: {
square: SQUARE_SCRIPT // square: SQUARE_SCRIPT
} // }
} // }
}); // });
it('should throw RootNodesUnavailableError', async () => { // it('should throw RootNodesUnavailableError', async () => {
const cluster = RedisCluster.create({ // const cluster = RedisCluster.create({
rootNodes: [] // rootNodes: []
}); // });
try { // try {
await assert.rejects( // await assert.rejects(
cluster.connect(), // cluster.connect(),
RootNodesUnavailableError // RootNodesUnavailableError
); // );
} catch (err) { // } catch (err) {
await cluster.disconnect(); // await cluster.disconnect();
throw err; // throw err;
} // }
}); // });
testUtils.testWithCluster('should handle live resharding', async cluster => { // testUtils.testWithCluster('should handle live resharding', async cluster => {
const slot = 12539, // const slot = 12539,
key = 'key', // key = 'key',
value = 'value'; // value = 'value';
await cluster.set(key, value); // await cluster.set(key, value);
const importing = cluster.slots[0].master, // const importing = cluster.slots[0].master,
migrating = cluster.slots[slot].master, // migrating = cluster.slots[slot].master,
[ importingClient, migratingClient ] = await Promise.all([ // [ importingClient, migratingClient ] = await Promise.all([
cluster.nodeClient(importing), // cluster.nodeClient(importing),
cluster.nodeClient(migrating) // cluster.nodeClient(migrating)
]); // ]);
await Promise.all([ // await Promise.all([
importingClient.clusterSetSlot(slot, ClusterSlotStates.IMPORTING, migrating.id), // importingClient.clusterSetSlot(slot, ClusterSlotStates.IMPORTING, migrating.id),
migratingClient.clusterSetSlot(slot, ClusterSlotStates.MIGRATING, importing.id) // migratingClient.clusterSetSlot(slot, ClusterSlotStates.MIGRATING, importing.id)
]); // ]);
// should be able to get the key from the migrating node // // should be able to get the key from the migrating node
assert.equal( // assert.equal(
await cluster.get(key), // await cluster.get(key),
value // value
); // );
await migratingClient.migrate( // await migratingClient.migrate(
importing.host, // importing.host,
importing.port, // importing.port,
key, // key,
0, // 0,
10 // 10
); // );
// should be able to get the key from the importing node using `ASKING` // // should be able to get the key from the importing node using `ASKING`
assert.equal( // assert.equal(
await cluster.get(key), // await cluster.get(key),
value // value
); // );
await Promise.all([ // await Promise.all([
importingClient.clusterSetSlot(slot, ClusterSlotStates.NODE, importing.id), // importingClient.clusterSetSlot(slot, ClusterSlotStates.NODE, importing.id),
migratingClient.clusterSetSlot(slot, ClusterSlotStates.NODE, importing.id), // migratingClient.clusterSetSlot(slot, ClusterSlotStates.NODE, importing.id),
]); // ]);
// should handle `MOVED` errors // // should handle `MOVED` errors
assert.equal( // assert.equal(
await cluster.get(key), // await cluster.get(key),
value // value
); // );
}, { // }, {
serverArguments: [], // serverArguments: [],
numberOfMasters: 2 // numberOfMasters: 2
}); // });
testUtils.testWithCluster('getRandomNode should spread the the load evenly', async cluster => { // testUtils.testWithCluster('getRandomNode should spread the the load evenly', async cluster => {
const totalNodes = cluster.masters.length + cluster.replicas.length, // const totalNodes = cluster.masters.length + cluster.replicas.length,
ids = new Set<string>(); // ids = new Set<string>();
for (let i = 0; i < totalNodes; i++) { // for (let i = 0; i < totalNodes; i++) {
ids.add(cluster.getRandomNode().id); // ids.add(cluster.getRandomNode().id);
} // }
assert.equal(ids.size, totalNodes); // assert.equal(ids.size, totalNodes);
}, GLOBAL.CLUSTERS.WITH_REPLICAS); // }, GLOBAL.CLUSTERS.WITH_REPLICAS);
testUtils.testWithCluster('getSlotRandomNode should spread the the load evenly', async cluster => { // testUtils.testWithCluster('getSlotRandomNode should spread the the load evenly', async cluster => {
const totalNodes = 1 + cluster.slots[0].replicas!.length, // const totalNodes = 1 + cluster.slots[0].replicas!.length,
ids = new Set<string>(); // ids = new Set<string>();
for (let i = 0; i < totalNodes; i++) { // for (let i = 0; i < totalNodes; i++) {
ids.add(cluster.getSlotRandomNode(0).id); // ids.add(cluster.getSlotRandomNode(0).id);
} // }
assert.equal(ids.size, totalNodes); // assert.equal(ids.size, totalNodes);
}, GLOBAL.CLUSTERS.WITH_REPLICAS); // }, GLOBAL.CLUSTERS.WITH_REPLICAS);
testUtils.testWithCluster('cluster topology', async cluster => { // testUtils.testWithCluster('cluster topology', async cluster => {
assert.equal(cluster.slots.length, 16384); // assert.equal(cluster.slots.length, 16384);
const { numberOfMasters, numberOfReplicas } = GLOBAL.CLUSTERS.WITH_REPLICAS; // const { numberOfMasters, numberOfReplicas } = GLOBAL.CLUSTERS.WITH_REPLICAS;
assert.equal(cluster.shards.length, numberOfMasters); // assert.equal(cluster.shards.length, numberOfMasters);
assert.equal(cluster.masters.length, numberOfMasters); // assert.equal(cluster.masters.length, numberOfMasters);
assert.equal(cluster.replicas.length, numberOfReplicas * numberOfMasters); // assert.equal(cluster.replicas.length, numberOfReplicas * numberOfMasters);
assert.equal(cluster.nodeByAddress.size, numberOfMasters + numberOfMasters * numberOfReplicas); // assert.equal(cluster.nodeByAddress.size, numberOfMasters + numberOfMasters * numberOfReplicas);
}, GLOBAL.CLUSTERS.WITH_REPLICAS); // }, GLOBAL.CLUSTERS.WITH_REPLICAS);
testUtils.testWithCluster('getMasters should be backwards competiable (without `minimizeConnections`)', async cluster => { // testUtils.testWithCluster('getMasters should be backwards competiable (without `minimizeConnections`)', async cluster => {
const masters = cluster.getMasters(); // const masters = cluster.getMasters();
assert.ok(Array.isArray(masters)); // assert.ok(Array.isArray(masters));
for (const master of masters) { // for (const master of masters) {
assert.equal(typeof master.id, 'string'); // assert.equal(typeof master.id, 'string');
assert.ok(master.client instanceof RedisClient); // assert.ok(master.client instanceof RedisClient);
} // }
}, { // }, {
...GLOBAL.CLUSTERS.OPEN, // ...GLOBAL.CLUSTERS.OPEN,
clusterConfiguration: { // clusterConfiguration: {
minimizeConnections: undefined // reset to default // minimizeConnections: undefined // reset to default
} // }
}); // });
testUtils.testWithCluster('getSlotMaster should be backwards competiable (without `minimizeConnections`)', async cluster => { // testUtils.testWithCluster('getSlotMaster should be backwards competiable (without `minimizeConnections`)', async cluster => {
const master = cluster.getSlotMaster(0); // const master = cluster.getSlotMaster(0);
assert.equal(typeof master.id, 'string'); // assert.equal(typeof master.id, 'string');
assert.ok(master.client instanceof RedisClient); // assert.ok(master.client instanceof RedisClient);
}, { // }, {
...GLOBAL.CLUSTERS.OPEN, // ...GLOBAL.CLUSTERS.OPEN,
clusterConfiguration: { // clusterConfiguration: {
minimizeConnections: undefined // reset to default // minimizeConnections: undefined // reset to default
} // }
}); // });
testUtils.testWithCluster('should throw CROSSSLOT error', async cluster => { // testUtils.testWithCluster('should throw CROSSSLOT error', async cluster => {
await assert.rejects(cluster.mGet(['a', 'b'])); // await assert.rejects(cluster.mGet(['a', 'b']));
}, GLOBAL.CLUSTERS.OPEN); // }, GLOBAL.CLUSTERS.OPEN);
testUtils.testWithCluster('should send commands with commandOptions to correct cluster slot (without redirections)', async cluster => { // testUtils.testWithCluster('should send commands with commandOptions to correct cluster slot (without redirections)', async cluster => {
// 'a' and 'b' hash to different cluster slots (see previous unit test) // // 'a' and 'b' hash to different cluster slots (see previous unit test)
// -> maxCommandRedirections 0: rejects on MOVED/ASK reply // // -> maxCommandRedirections 0: rejects on MOVED/ASK reply
await cluster.set(commandOptions({ isolated: true }), 'a', '1'), // await cluster.set(commandOptions({ isolated: true }), 'a', '1'),
await cluster.set(commandOptions({ isolated: true }), 'b', '2'), // await cluster.set(commandOptions({ isolated: true }), 'b', '2'),
assert.equal(await cluster.get('a'), '1'); // assert.equal(await cluster.get('a'), '1');
assert.equal(await cluster.get('b'), '2'); // assert.equal(await cluster.get('b'), '2');
}, { // }, {
...GLOBAL.CLUSTERS.OPEN, // ...GLOBAL.CLUSTERS.OPEN,
clusterConfiguration: { // clusterConfiguration: {
maxCommandRedirections: 0 // maxCommandRedirections: 0
} // }
}); // });
describe('minimizeConnections', () => { // describe('minimizeConnections', () => {
testUtils.testWithCluster('false', async cluster => { // testUtils.testWithCluster('false', async cluster => {
for (const master of cluster.masters) { // for (const master of cluster.masters) {
assert.ok(master.client instanceof RedisClient); // assert.ok(master.client instanceof RedisClient);
} // }
}, { // }, {
...GLOBAL.CLUSTERS.OPEN, // ...GLOBAL.CLUSTERS.OPEN,
clusterConfiguration: { // clusterConfiguration: {
minimizeConnections: false // minimizeConnections: false
} // }
}); // });
testUtils.testWithCluster('true', async cluster => { // testUtils.testWithCluster('true', async cluster => {
for (const master of cluster.masters) { // for (const master of cluster.masters) {
assert.equal(master.client, undefined); // assert.equal(master.client, undefined);
} // }
}, { // }, {
...GLOBAL.CLUSTERS.OPEN, // ...GLOBAL.CLUSTERS.OPEN,
clusterConfiguration: { // clusterConfiguration: {
minimizeConnections: true // minimizeConnections: true
} // }
}); // });
}); // });
describe('PubSub', () => { // describe('PubSub', () => {
testUtils.testWithCluster('subscribe & unsubscribe', async cluster => { // testUtils.testWithCluster('subscribe & unsubscribe', async cluster => {
const listener = spy(); // const listener = spy();
await cluster.subscribe('channel', listener); // await cluster.subscribe('channel', listener);
await Promise.all([ // await Promise.all([
waitTillBeenCalled(listener), // waitTillBeenCalled(listener),
cluster.publish('channel', 'message') // cluster.publish('channel', 'message')
]); // ]);
assert.ok(listener.calledOnceWithExactly('message', 'channel')); // assert.ok(listener.calledOnceWithExactly('message', 'channel'));
await cluster.unsubscribe('channel', listener); // await cluster.unsubscribe('channel', listener);
assert.equal(cluster.pubSubNode, undefined); // assert.equal(cluster.pubSubNode, undefined);
}, GLOBAL.CLUSTERS.OPEN); // }, GLOBAL.CLUSTERS.OPEN);
testUtils.testWithCluster('psubscribe & punsubscribe', async cluster => { // testUtils.testWithCluster('psubscribe & punsubscribe', async cluster => {
const listener = spy(); // const listener = spy();
await cluster.pSubscribe('channe*', listener); // await cluster.pSubscribe('channe*', listener);
await Promise.all([ // await Promise.all([
waitTillBeenCalled(listener), // waitTillBeenCalled(listener),
cluster.publish('channel', 'message') // cluster.publish('channel', 'message')
]); // ]);
assert.ok(listener.calledOnceWithExactly('message', 'channel')); // assert.ok(listener.calledOnceWithExactly('message', 'channel'));
await cluster.pUnsubscribe('channe*', listener); // await cluster.pUnsubscribe('channe*', listener);
assert.equal(cluster.pubSubNode, undefined); // assert.equal(cluster.pubSubNode, undefined);
}, GLOBAL.CLUSTERS.OPEN); // }, GLOBAL.CLUSTERS.OPEN);
testUtils.testWithCluster('should move listeners when PubSub node disconnects from the cluster', async cluster => { // testUtils.testWithCluster('should move listeners when PubSub node disconnects from the cluster', async cluster => {
const listener = spy(); // const listener = spy();
await cluster.subscribe('channel', listener); // await cluster.subscribe('channel', listener);
assert.ok(cluster.pubSubNode); // assert.ok(cluster.pubSubNode);
const [ migrating, importing ] = cluster.masters[0].address === cluster.pubSubNode.address ? // const [ migrating, importing ] = cluster.masters[0].address === cluster.pubSubNode.address ?
cluster.masters : // cluster.masters :
[cluster.masters[1], cluster.masters[0]], // [cluster.masters[1], cluster.masters[0]],
[ migratingClient, importingClient ] = await Promise.all([ // [ migratingClient, importingClient ] = await Promise.all([
cluster.nodeClient(migrating), // cluster.nodeClient(migrating),
cluster.nodeClient(importing) // cluster.nodeClient(importing)
]); // ]);
const range = cluster.slots[0].master === migrating ? { // const range = cluster.slots[0].master === migrating ? {
key: 'bar', // 5061 // key: 'bar', // 5061
start: 0, // start: 0,
end: 8191 // end: 8191
} : { // } : {
key: 'foo', // 12182 // key: 'foo', // 12182
start: 8192, // start: 8192,
end: 16383 // end: 16383
}; // };
await Promise.all([ // await Promise.all([
migratingClient.clusterDelSlotsRange(range), // migratingClient.clusterDelSlotsRange(range),
importingClient.clusterDelSlotsRange(range), // importingClient.clusterDelSlotsRange(range),
importingClient.clusterAddSlotsRange(range) // importingClient.clusterAddSlotsRange(range)
]); // ]);
// wait for migrating node to be notified about the new topology // // wait for migrating node to be notified about the new topology
while ((await migratingClient.clusterInfo()).state !== 'ok') { // while ((await migratingClient.clusterInfo()).state !== 'ok') {
await promiseTimeout(50); // await promiseTimeout(50);
} // }
// make sure to cause `MOVED` error // // make sure to cause `MOVED` error
await cluster.get(range.key); // await cluster.get(range.key);
await Promise.all([ // await Promise.all([
cluster.publish('channel', 'message'), // cluster.publish('channel', 'message'),
waitTillBeenCalled(listener) // waitTillBeenCalled(listener)
]); // ]);
assert.ok(listener.calledOnceWithExactly('message', 'channel')); // assert.ok(listener.calledOnceWithExactly('message', 'channel'));
}, { // }, {
serverArguments: [], // serverArguments: [],
numberOfMasters: 2, // numberOfMasters: 2,
minimumDockerVersion: [7] // minimumDockerVersion: [7]
}); // });
testUtils.testWithCluster('ssubscribe & sunsubscribe', async cluster => { // testUtils.testWithCluster('ssubscribe & sunsubscribe', async cluster => {
const listener = spy(); // const listener = spy();
await cluster.sSubscribe('channel', listener); // await cluster.sSubscribe('channel', listener);
await Promise.all([ // await Promise.all([
waitTillBeenCalled(listener), // waitTillBeenCalled(listener),
cluster.sPublish('channel', 'message') // cluster.sPublish('channel', 'message')
]); // ]);
assert.ok(listener.calledOnceWithExactly('message', 'channel')); // assert.ok(listener.calledOnceWithExactly('message', 'channel'));
await cluster.sUnsubscribe('channel', listener); // await cluster.sUnsubscribe('channel', listener);
// 10328 is the slot of `channel` // // 10328 is the slot of `channel`
assert.equal(cluster.slots[10328].master.pubSubClient, undefined); // assert.equal(cluster.slots[10328].master.pubSubClient, undefined);
}, { // }, {
...GLOBAL.CLUSTERS.OPEN, // ...GLOBAL.CLUSTERS.OPEN,
minimumDockerVersion: [7] // minimumDockerVersion: [7]
}); // });
testUtils.testWithCluster('should handle sharded-channel-moved events', async cluster => { // testUtils.testWithCluster('should handle sharded-channel-moved events', async cluster => {
const SLOT = 10328, // const SLOT = 10328,
migrating = cluster.slots[SLOT].master, // migrating = cluster.slots[SLOT].master,
importing = cluster.masters.find(master => master !== migrating)!, // importing = cluster.masters.find(master => master !== migrating)!,
[ migratingClient, importingClient ] = await Promise.all([ // [ migratingClient, importingClient ] = await Promise.all([
cluster.nodeClient(migrating), // cluster.nodeClient(migrating),
cluster.nodeClient(importing) // cluster.nodeClient(importing)
]); // ]);
await Promise.all([ // await Promise.all([
migratingClient.clusterDelSlots(SLOT), // migratingClient.clusterDelSlots(SLOT),
importingClient.clusterDelSlots(SLOT), // importingClient.clusterDelSlots(SLOT),
importingClient.clusterAddSlots(SLOT) // importingClient.clusterAddSlots(SLOT)
]); // ]);
// wait for migrating node to be notified about the new topology // // wait for migrating node to be notified about the new topology
while ((await migratingClient.clusterInfo()).state !== 'ok') { // while ((await migratingClient.clusterInfo()).state !== 'ok') {
await promiseTimeout(50); // await promiseTimeout(50);
} // }
const listener = spy(); // const listener = spy();
// will trigger `MOVED` error // // will trigger `MOVED` error
await cluster.sSubscribe('channel', listener); // await cluster.sSubscribe('channel', listener);
await Promise.all([ // await Promise.all([
waitTillBeenCalled(listener), // waitTillBeenCalled(listener),
cluster.sPublish('channel', 'message') // cluster.sPublish('channel', 'message')
]); // ]);
assert.ok(listener.calledOnceWithExactly('message', 'channel')); // assert.ok(listener.calledOnceWithExactly('message', 'channel'));
}, { // }, {
serverArguments: [], // serverArguments: [],
minimumDockerVersion: [7] // minimumDockerVersion: [7]
}); // });
}); // });
}); // });

2
packages/client/lib/commands/APPEND.spec.ts
View File

@@ -2,7 +2,7 @@ import { strict as assert } from 'assert';
import testUtils, { GLOBAL } from '../test-utils'; import testUtils, { GLOBAL } from '../test-utils';
import APPEND from './APPEND'; import APPEND from './APPEND';
describe.only('APPEND', () => { describe('APPEND', () => {
it('transformArguments', () => { it('transformArguments', () => {
assert.deepEqual( assert.deepEqual(
APPEND.transformArguments('key', 'value'), APPEND.transformArguments('key', 'value'),

2
packages/client/lib/commands/PING.spec.ts
View File

@@ -2,7 +2,7 @@ import { strict as assert } from 'assert';
import testUtils, { GLOBAL } from '../test-utils'; import testUtils, { GLOBAL } from '../test-utils';
import PING from './PING'; import PING from './PING';
describe.only('PING', () => { describe('PING', () => {
describe('transformArguments', () => { describe('transformArguments', () => {
it('default', () => { it('default', () => {
assert.deepEqual( assert.deepEqual(