1195241456

Build # 1195241456

Build Type

push

github

Committed by leibale

Commit Message

fix client.quit, add error events on cluster, fix some "deepsource.io" warnings

Run Details

988 of 1098 branches covered (89.98%)

Branch coverage included in aggregate %.

21 of 25 new or added lines in 6 files covered. (84.0%)

1 existing line in 1 file now uncovered.

2706 of 2784 relevant lines covered (97.2%)

2935.35 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

71.43

/lib/cluster-slots.ts

import calculateSlot from 'cluster-key-slot';
import RedisClient, { RedisClientType } from './client';
import { RedisSocketOptions } from './socket';
import { RedisClusterMasterNode, RedisClusterReplicaNode } from './commands/CLUSTER_NODES';
import { RedisClusterOptions } from './cluster';
import { RedisModules } from './commands';
import { RedisLuaScripts } from './lua-script';

export interface ClusterNode<M extends RedisModules, S extends RedisLuaScripts> {
    id: string;
    client: RedisClientType<M, S>;
}

interface SlotNodes<M extends RedisModules, S extends RedisLuaScripts> {
    master: ClusterNode<M, S>;
    replicas: Array<ClusterNode<M, S>>;
    clientIterator: IterableIterator<RedisClientType<M, S>> | undefined;
}

type OnError = (err: unknown) => void;

export default class RedisClusterSlots<M extends RedisModules, S extends RedisLuaScripts> {
    readonly #options: RedisClusterOptions;
    readonly #onError: OnError;
    readonly #nodeByUrl = new Map<string, ClusterNode<M, S>>();
    readonly #slots: Array<SlotNodes<M, S>> = [];

    constructor(options: RedisClusterOptions, onError: OnError) {
        this.#options = options;
        this.#onError = onError;
    }

    async connect(): Promise<void> {
        for (const rootNode of this.#options.rootNodes) {
            if (await this.#discoverNodes(rootNode)) return;
        }

        throw new Error('None of the root nodes is available');
    }

    async discover(startWith: RedisClientType<M, S>): Promise<void> {
        if (await this.#discoverNodes(startWith.options?.socket)) return;

        for (const { client } of this.#nodeByUrl.values()) {
            if (client === startWith) continue;

            if (await this.#discoverNodes(client.options?.socket)) return;
        }

        throw new Error('None of the cluster nodes is available');
    }

    async #discoverNodes(socketOptions?: RedisSocketOptions): Promise<boolean> {
        const client = RedisClient.create({
            socket: socketOptions
        });

        await client.connect();

        try {
            await this.#reset(await client.clusterNodes());
            return true;
        } catch (err) {
            this.#onError(err);
            return false;
        } finally {
            if (client.isOpen) {
                await client.disconnect();
            }
        }
    }

    async #reset(masters: Array<RedisClusterMasterNode>): Promise<void> {
        // Override this.#slots and add not existing clients to this.#clientByKey
        const promises: Array<Promise<void>> = [],
            clientsInUse = new Set<string>();
        for (const master of masters) {
            const slot = {
                master: this.#initiateClientForNode(master, false, clientsInUse, promises),
                replicas: this.#options.useReplicas ?
                    master.replicas.map(replica => this.#initiateClientForNode(replica, true, clientsInUse, promises)) :
                    [],
                    clientIterator: undefined // will be initiated in use
            };

            for (const { from, to } of master.slots) {
                for (let i = from; i < to; i++) {
                    this.#slots[i] = slot;
                }
            }
        }

        // Remove unused clients from this.#clientBykey using clientsInUse
        for (const [url, { client }] of this.#nodeByUrl.entries()) {
            if (clientsInUse.has(url)) continue;

            promises.push(client.disconnect());
            this.#nodeByUrl.delete(url);
        }

        await Promise.all(promises);
    }

    #initiateClientForNode(nodeData: RedisClusterMasterNode | RedisClusterReplicaNode, readonly: boolean, clientsInUse: Set<string>, promises: Array<Promise<void>>): ClusterNode<M, S> {
        const url = `${nodeData.host}:${nodeData.port}`;
        clientsInUse.add(url);

        let node = this.#nodeByUrl.get(url);
        if (!node) {
            node = {
                id: nodeData.id,
                client: RedisClient.create({
                    socket: {
                        host: nodeData.host,
                        port: nodeData.port
                    },
                    readonly
                })
            };
            promises.push(node.client.connect());
            this.#nodeByUrl.set(url, node);
        }

        return node;
    }

    getSlotMaster(slot: number): ClusterNode<M, S> {
        return this.#slots[slot].master;
    }

    *#slotClientIterator(slotNumber: number): IterableIterator<RedisClientType<M, S>> {
        const slot = this.#slots[slotNumber];
        yield slot.master.client;

        for (const replica of slot.replicas) {
            yield replica.client;
        }
    }

    #getSlotClient(slotNumber: number): RedisClientType<M, S> {
        const slot = this.#slots[slotNumber];
        if (!slot.clientIterator) {
            slot.clientIterator = this.#slotClientIterator(slotNumber);
        }

        const {done, value} = slot.clientIterator.next();
        if (done) {
            slot.clientIterator = undefined;
            return this.#getSlotClient(slotNumber);
        }

        return value;
    }

    #randomClientIterator?: IterableIterator<ClusterNode<M, S>>;

    #getRandomClient(): RedisClientType<M, S> {
        if (!this.#nodeByUrl.size) {
            throw new Error('Cluster is not connected');
        }

        if (!this.#randomClientIterator) {
            this.#randomClientIterator = this.#nodeByUrl.values();
        }

        const {done, value} = this.#randomClientIterator.next();
        if (done) {
            this.#randomClientIterator = undefined;
            return this.#getRandomClient();
        }

        return value.client;
    }

    getClient(firstKey?: string, isReadonly?: boolean): RedisClientType<M, S> {
        if (!firstKey) {
            return this.#getRandomClient();
        }

        const slot = calculateSlot(firstKey);
        if (!isReadonly || !this.#options.useReplicas) {
            return this.getSlotMaster(slot).client;
        }

        return this.#getSlotClient(slot);
    }

    getMasters(): Array<ClusterNode<M, S>> {
        const masters = [];

        for (const node of this.#nodeByUrl.values()) {
            if (node.client.options?.readonly) continue;

            masters.push(node);
        }

        return masters;
    }

    getNodeByUrl(url: string): ClusterNode<M, S> | undefined {
        return this.#nodeByUrl.get(url);
    }

    async disconnect(): Promise<void> {
        await Promise.all(
            [...this.#nodeByUrl.values()].map(({ client }) => client.disconnect())
        );

        this.#nodeByUrl.clear();
        this.#slots.splice(0);
    }
}

1	import calculateSlot from 'cluster-key-slot';	6✔
2	import RedisClient, { RedisClientType } from './client';	6✔
3	import { RedisSocketOptions } from './socket';
4	import { RedisClusterMasterNode, RedisClusterReplicaNode } from './commands/CLUSTER_NODES';
5	import { RedisClusterOptions } from './cluster';
6	import { RedisModules } from './commands';
7	import { RedisLuaScripts } from './lua-script';
8
9	export interface ClusterNode<M extends RedisModules, S extends RedisLuaScripts> {
10	id: string;
11	client: RedisClientType<M, S>;
12	}
13
14	interface SlotNodes<M extends RedisModules, S extends RedisLuaScripts> {
15	master: ClusterNode<M, S>;
16	replicas: Array<ClusterNode<M, S>>;
17	clientIterator: IterableIterator<RedisClientType<M, S>> \| undefined;
18	}
19
20	type OnError = (err: unknown) => void;
21
22	export default class RedisClusterSlots<M extends RedisModules, S extends RedisLuaScripts> {	6✔
23	readonly #options: RedisClusterOptions;	411✔
24	readonly #onError: OnError;	411✔
25	readonly #nodeByUrl = new Map<string, ClusterNode<M, S>>();	411✔
26	readonly #slots: Array<SlotNodes<M, S>> = [];	411✔
27
28	constructor(options: RedisClusterOptions, onError: OnError) {
29	this.#options = options;	411✔
30	this.#onError = onError;	411✔
31	}
32
33	async connect(): Promise<void> {
34	for (const rootNode of this.#options.rootNodes) {	411✔
35	if (await this.#discoverNodes(rootNode)) return;	411!
36	}
37
38	throw new Error('None of the root nodes is available');	×
39	}
40
41	async discover(startWith: RedisClientType<M, S>): Promise<void> {
NEW 42	if (await this.#discoverNodes(startWith.options?.socket)) return;	×
43
44	for (const { client } of this.#nodeByUrl.values()) {	×
45	if (client === startWith) continue;	×
46
NEW 47	if (await this.#discoverNodes(client.options?.socket)) return;	×
48	}
49
50	throw new Error('None of the cluster nodes is available');	×
51	}
52
53	async #discoverNodes(socketOptions?: RedisSocketOptions): Promise<boolean> {	6✔
54	const client = RedisClient.create({	411✔
55	socket: socketOptions
56	});
57
58	await client.connect();	411✔
59
60	try {	411✔
61	await this.#reset(await client.clusterNodes());	411✔
62	return true;	411✔
63	} catch (err) {
NEW 64	this.#onError(err);	×
NEW 65	return false;	×
66	} finally {
67	if (client.isOpen) {	411!
68	await client.disconnect();	411✔
69	}
70	}
71	}
72
73	async #reset(masters: Array<RedisClusterMasterNode>): Promise<void> {
74	// Override this.#slots and add not existing clients to this.#clientByKey
75	const promises: Array<Promise<void>> = [],	411✔
76	clientsInUse = new Set<string>();	411✔
77	for (const master of masters) {	411✔
78	const slot = {	1,233✔
79	master: this.#initiateClientForNode(master, false, clientsInUse, promises),
80	replicas: this.#options.useReplicas ?
81	master.replicas.map(replica => this.#initiateClientForNode(replica, true, clientsInUse, promises)) :	✔
82	[],
83	clientIterator: undefined // will be initiated in use
84	};
85
86	for (const { from, to } of master.slots) {	1,233✔
87	for (let i = from; i < to; i++) {	1,233✔
88	this.#slots[i] = slot;	6,732,591✔
89	}
90	}
91	}
92
93	// Remove unused clients from this.#clientBykey using clientsInUse
94	for (const [url, { client }] of this.#nodeByUrl.entries()) {	411✔
95	if (clientsInUse.has(url)) continue;	1,233!
96
UNCOV 97	promises.push(client.disconnect());	×
98	this.#nodeByUrl.delete(url);	×
99	}
100
101	await Promise.all(promises);	411✔
102	}
103
104	#initiateClientForNode(nodeData: RedisClusterMasterNode \| RedisClusterReplicaNode, readonly: boolean, clientsInUse: Set<string>, promises: Array<Promise<void>>): ClusterNode<M, S> {
105	const url = `${nodeData.host}:${nodeData.port}`;	1,233✔
106	clientsInUse.add(url);	1,233✔
107
108	let node = this.#nodeByUrl.get(url);	1,233✔
109	if (!node) {	1,233!
110	node = {	1,233✔
111	id: nodeData.id,
112	client: RedisClient.create({
113	socket: {
114	host: nodeData.host,
115	port: nodeData.port
116	},
117	readonly
118	})
119	};
120	promises.push(node.client.connect());	1,233✔
121	this.#nodeByUrl.set(url, node);	1,233✔
122	}
123
124	return node;	1,233✔
125	}
126
127	getSlotMaster(slot: number): ClusterNode<M, S> {
128	return this.#slots[slot].master;	363✔
129	}
130
131	*#slotClientIterator(slotNumber: number): IterableIterator<RedisClientType<M, S>> {
132	const slot = this.#slots[slotNumber];	24✔
133	yield slot.master.client;	24✔
134
135	for (const replica of slot.replicas) {	18✔
136	yield replica.client;	×
137	}
138	}
139
140	#getSlotClient(slotNumber: number): RedisClientType<M, S> {
141	const slot = this.#slots[slotNumber];	42✔
142	if (!slot.clientIterator) {	42✔
143	slot.clientIterator = this.#slotClientIterator(slotNumber);	24✔
144	}
145
146	const {done, value} = slot.clientIterator.next();	42✔
147	if (done) {	42✔
148	slot.clientIterator = undefined;	18✔
149	return this.#getSlotClient(slotNumber);	18✔
150	}
151
152	return value;	24✔
153	}
154
155	#randomClientIterator?: IterableIterator<ClusterNode<M, S>>;	411✔
156
157	#getRandomClient(): RedisClientType<M, S> {
158	if (!this.#nodeByUrl.size) {	126!
159	throw new Error('Cluster is not connected');	×
160	}
161
162	if (!this.#randomClientIterator) {	126!
163	this.#randomClientIterator = this.#nodeByUrl.values();	126✔
164	}
165
166	const {done, value} = this.#randomClientIterator.next();	126✔
167	if (done) {	126!
168	this.#randomClientIterator = undefined;	×
169	return this.#getRandomClient();	×
170	}
171
172	return value.client;	126✔
173	}
174
175	getClient(firstKey?: string, isReadonly?: boolean): RedisClientType<M, S> {
176	if (!firstKey) {	507✔
177	return this.#getRandomClient();	126✔
178	}
179
180	const slot = calculateSlot(firstKey);	381✔
181	if (!isReadonly \|\| !this.#options.useReplicas) {	381✔
182	return this.getSlotMaster(slot).client;	357✔
183	}
184
185	return this.#getSlotClient(slot);	24✔
186	}
187
188	getMasters(): Array<ClusterNode<M, S>> {
189	const masters = [];	792✔
190
191	for (const node of this.#nodeByUrl.values()) {	792✔
192	if (node.client.options?.readonly) continue;	2,376!
193
194	masters.push(node);	2,376✔
195	}
196
197	return masters;	792✔
198	}
199
200	getNodeByUrl(url: string): ClusterNode<M, S> \| undefined {
201	return this.#nodeByUrl.get(url);	6✔
202	}
203
204	async disconnect(): Promise<void> {
205	await Promise.all(	411✔
206	[...this.#nodeByUrl.values()].map(({ client }) => client.disconnect())	1,233✔
207	);
208
209	this.#nodeByUrl.clear();	411✔
210	this.#slots.splice(0);	411✔
211	}
212	}

NodeRedis / node-redis / 1195241456

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous