You've already forked node-redis
mirror of
https://github.com/redis/node-redis.git
synced 2025-08-06 02:15:48 +03:00
fix isolationPool
after reconnect (#2409)
* fix #2406 - fix isolationPool after reconnect * revert breaking change * fix
This commit is contained in:
@@ -607,11 +607,41 @@ describe('Client', () => {
|
|||||||
}
|
}
|
||||||
});
|
});
|
||||||
|
|
||||||
testUtils.testWithClient('executeIsolated', async client => {
|
describe('isolationPool', () => {
|
||||||
const id = await client.clientId(),
|
testUtils.testWithClient('executeIsolated', async client => {
|
||||||
isolatedId = await client.executeIsolated(isolatedClient => isolatedClient.clientId());
|
const id = await client.clientId(),
|
||||||
assert.ok(id !== isolatedId);
|
isolatedId = await client.executeIsolated(isolatedClient => isolatedClient.clientId());
|
||||||
}, GLOBAL.SERVERS.OPEN);
|
assert.ok(id !== isolatedId);
|
||||||
|
}, GLOBAL.SERVERS.OPEN);
|
||||||
|
|
||||||
|
testUtils.testWithClient('should be able to use pool even before connect', async client => {
|
||||||
|
await client.executeIsolated(() => Promise.resolve());
|
||||||
|
// make sure to destroy isolation pool
|
||||||
|
await client.connect();
|
||||||
|
await client.disconnect();
|
||||||
|
}, {
|
||||||
|
...GLOBAL.SERVERS.OPEN,
|
||||||
|
disableClientSetup: true
|
||||||
|
});
|
||||||
|
|
||||||
|
testUtils.testWithClient('should work after reconnect (#2406)', async client => {
|
||||||
|
await client.disconnect();
|
||||||
|
await client.connect();
|
||||||
|
await client.executeIsolated(() => Promise.resolve());
|
||||||
|
}, GLOBAL.SERVERS.OPEN);
|
||||||
|
|
||||||
|
testUtils.testWithClient('should throw ClientClosedError after disconnect', async client => {
|
||||||
|
await client.connect();
|
||||||
|
await client.disconnect();
|
||||||
|
await assert.rejects(
|
||||||
|
client.executeIsolated(() => Promise.resolve()),
|
||||||
|
ClientClosedError
|
||||||
|
);
|
||||||
|
}, {
|
||||||
|
...GLOBAL.SERVERS.OPEN,
|
||||||
|
disableClientSetup: true
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
async function killClient<
|
async function killClient<
|
||||||
M extends RedisModules,
|
M extends RedisModules,
|
||||||
@@ -731,7 +761,7 @@ describe('Client', () => {
|
|||||||
members.map<MemberTuple>(member => [member.value, member.score]).sort(sort)
|
members.map<MemberTuple>(member => [member.value, member.score]).sort(sort)
|
||||||
);
|
);
|
||||||
}, GLOBAL.SERVERS.OPEN);
|
}, GLOBAL.SERVERS.OPEN);
|
||||||
|
|
||||||
describe('PubSub', () => {
|
describe('PubSub', () => {
|
||||||
testUtils.testWithClient('should be able to publish and subscribe to messages', async publisher => {
|
testUtils.testWithClient('should be able to publish and subscribe to messages', async publisher => {
|
||||||
function assertStringListener(message: string, channel: string) {
|
function assertStringListener(message: string, channel: string) {
|
||||||
|
@@ -15,7 +15,6 @@ import { ClientClosedError, ClientOfflineError, DisconnectsClientError } from '.
|
|||||||
import { URL } from 'url';
|
import { URL } from 'url';
|
||||||
import { TcpSocketConnectOpts } from 'net';
|
import { TcpSocketConnectOpts } from 'net';
|
||||||
import { PubSubType, PubSubListener, PubSubTypeListeners, ChannelListeners } from './pub-sub';
|
import { PubSubType, PubSubListener, PubSubTypeListeners, ChannelListeners } from './pub-sub';
|
||||||
import { callbackify } from 'util';
|
|
||||||
|
|
||||||
export interface RedisClientOptions<
|
export interface RedisClientOptions<
|
||||||
M extends RedisModules = RedisModules,
|
M extends RedisModules = RedisModules,
|
||||||
@@ -190,7 +189,7 @@ export default class RedisClient<
|
|||||||
readonly #options?: RedisClientOptions<M, F, S>;
|
readonly #options?: RedisClientOptions<M, F, S>;
|
||||||
readonly #socket: RedisSocket;
|
readonly #socket: RedisSocket;
|
||||||
readonly #queue: RedisCommandsQueue;
|
readonly #queue: RedisCommandsQueue;
|
||||||
readonly #isolationPool: Pool<RedisClientType<M, F, S>>;
|
#isolationPool?: Pool<RedisClientType<M, F, S>>;
|
||||||
readonly #v4: Record<string, any> = {};
|
readonly #v4: Record<string, any> = {};
|
||||||
#selectedDB = 0;
|
#selectedDB = 0;
|
||||||
|
|
||||||
@@ -223,16 +222,9 @@ export default class RedisClient<
|
|||||||
this.#options = this.#initiateOptions(options);
|
this.#options = this.#initiateOptions(options);
|
||||||
this.#queue = this.#initiateQueue();
|
this.#queue = this.#initiateQueue();
|
||||||
this.#socket = this.#initiateSocket();
|
this.#socket = this.#initiateSocket();
|
||||||
this.#isolationPool = createPool({
|
// should be initiated in connect, not here
|
||||||
create: async () => {
|
// TODO: consider breaking in v5
|
||||||
const duplicate = this.duplicate({
|
this.#isolationPool = this.#initiateIsolationPool();
|
||||||
isolationPoolOptions: undefined
|
|
||||||
}).on('error', err => this.emit('error', err));
|
|
||||||
await duplicate.connect();
|
|
||||||
return duplicate;
|
|
||||||
},
|
|
||||||
destroy: client => client.disconnect()
|
|
||||||
}, options?.isolationPoolOptions);
|
|
||||||
this.#legacyMode();
|
this.#legacyMode();
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -337,6 +329,19 @@ export default class RedisClient<
|
|||||||
.on('end', () => this.emit('end'));
|
.on('end', () => this.emit('end'));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#initiateIsolationPool() {
|
||||||
|
return createPool({
|
||||||
|
create: async () => {
|
||||||
|
const duplicate = this.duplicate({
|
||||||
|
isolationPoolOptions: undefined
|
||||||
|
}).on('error', err => this.emit('error', err));
|
||||||
|
await duplicate.connect();
|
||||||
|
return duplicate;
|
||||||
|
},
|
||||||
|
destroy: client => client.disconnect()
|
||||||
|
}, this.#options?.isolationPoolOptions);
|
||||||
|
}
|
||||||
|
|
||||||
#legacyMode(): void {
|
#legacyMode(): void {
|
||||||
if (!this.#options?.legacyMode) return;
|
if (!this.#options?.legacyMode) return;
|
||||||
|
|
||||||
@@ -422,6 +427,8 @@ export default class RedisClient<
|
|||||||
}
|
}
|
||||||
|
|
||||||
connect(): Promise<void> {
|
connect(): Promise<void> {
|
||||||
|
// see comment in constructor
|
||||||
|
this.#isolationPool ??= this.#initiateIsolationPool();
|
||||||
return this.#socket.connect();
|
return this.#socket.connect();
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -704,6 +711,7 @@ export default class RedisClient<
|
|||||||
}
|
}
|
||||||
|
|
||||||
executeIsolated<T>(fn: (client: RedisClientType<M, F, S>) => T | Promise<T>): Promise<T> {
|
executeIsolated<T>(fn: (client: RedisClientType<M, F, S>) => T | Promise<T>): Promise<T> {
|
||||||
|
if (!this.#isolationPool) return Promise.reject(new ClientClosedError());
|
||||||
return this.#isolationPool.use(fn);
|
return this.#isolationPool.use(fn);
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -802,8 +810,9 @@ export default class RedisClient<
|
|||||||
}
|
}
|
||||||
|
|
||||||
async #destroyIsolationPool(): Promise<void> {
|
async #destroyIsolationPool(): Promise<void> {
|
||||||
await this.#isolationPool.drain();
|
await this.#isolationPool!.drain();
|
||||||
await this.#isolationPool.clear();
|
await this.#isolationPool!.clear();
|
||||||
|
this.#isolationPool = undefined;
|
||||||
}
|
}
|
||||||
|
|
||||||
ref(): void {
|
ref(): void {
|
||||||
|
Reference in New Issue
Block a user