This is an automated email from the ASF dual-hosted git repository. CritasWang pushed a commit to branch fix/session-idempotent-open in repository https://gitbox.apache.org/repos/asf/iotdb-client-nodejs.git
commit 8e49c0ff3d23397284bad12078b08d4e79bc237d Author: CritasWang <[email protected]> AuthorDate: Thu Jul 30 18:48:51 2026 +0800 fix: make session open idempotent --- eslint.config.mjs | 5 +++ src/connection/Connection.ts | 22 ++++++++- tests/unit/Connection.test.ts | 102 ++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 128 insertions(+), 1 deletion(-) diff --git a/eslint.config.mjs b/eslint.config.mjs index c44ac59..4f4a530 100644 --- a/eslint.config.mjs +++ b/eslint.config.mjs @@ -27,6 +27,11 @@ export default [ '@typescript-eslint/no-explicit-any': 'warn', '@typescript-eslint/explicit-function-return-type': 'off', '@typescript-eslint/no-unused-vars': ['error', { argsIgnorePattern: '^_' }], + // Thrift generates CommonJS modules whose declaration files are not ES modules. + '@typescript-eslint/no-require-imports': [ + 'error', + { allow: ['/thrift/generated/'] }, + ], }, }, { diff --git a/src/connection/Connection.ts b/src/connection/Connection.ts index 38969d0..5c88e84 100644 --- a/src/connection/Connection.ts +++ b/src/connection/Connection.ts @@ -31,12 +31,32 @@ export class Connection { private sessionId: number | null = null; private statementId: number | null = null; private isConnected: boolean = false; + private openingPromise: Promise<void> | null = null; constructor(config: InternalConfig) { this.config = config; } async open(): Promise<void> { + if (this.isConnected) { + return; + } + + if (!this.openingPromise) { + this.openingPromise = this.establishConnection(); + } + + const openingPromise = this.openingPromise; + try { + await openingPromise; + } finally { + if (this.openingPromise === openingPromise) { + this.openingPromise = null; + } + } + } + + private async establishConnection(): Promise<void> { try { if (!this.config.host || !this.config.port) { throw new Error("Host and port are required for connection"); @@ -216,7 +236,7 @@ export class Connection { }); // Use a timeout handle that we can clear - let timeoutHandle: NodeJS.Timeout | null = null; + let timeoutHandle: ReturnType<typeof setTimeout> | null = null; await Promise.race([ new Promise<void>((resolve, reject) => { diff --git a/tests/unit/Connection.test.ts b/tests/unit/Connection.test.ts index 7d5b01b..5642a5f 100644 --- a/tests/unit/Connection.test.ts +++ b/tests/unit/Connection.test.ts @@ -130,6 +130,78 @@ describe("Connection", () => { await connection.close(); }); + test("Should not create another connection when open is called repeatedly", async () => { + const config: InternalConfig = { + host: "localhost", + port: 6667, + username: "root", + password: "root", + enableSSL: false, + sqlDialect: "tree", + }; + const connection = new Connection(config); + + await connection.open(); + await connection.open(); + + expect(thriftMock.createConnection).toHaveBeenCalledTimes(1); + expect(thriftMock.createClient).toHaveBeenCalledTimes(1); + + await connection.close(); + }); + + test("Should share the connection attempt between concurrent open calls", async () => { + let completeOpenSession!: (error: Error | null, response: unknown) => void; + const openSession = jest.fn( + ( + _req: unknown, + callback: (error: Error | null, response: unknown) => void, + ) => { + completeOpenSession = callback; + }, + ); + const requestStatementId = jest.fn( + ( + _sessionId: unknown, + callback: (error: Error | null, statementId: number) => void, + ) => callback(null, 456), + ); + const closeSession = jest.fn( + ( + _req: unknown, + callback: (error: Error | null, response: unknown) => void, + ) => callback(null, { status: { code: 200 } }), + ); + thriftMock.createClient.mockReturnValueOnce({ + openSession, + requestStatementId, + closeSession, + }); + + const connection = new Connection({ + host: "localhost", + port: 6667, + username: "root", + password: "root", + enableSSL: false, + sqlDialect: "tree", + }); + + const firstOpen = connection.open(); + const secondOpen = connection.open(); + + expect(thriftMock.createConnection).toHaveBeenCalledTimes(1); + expect(openSession).toHaveBeenCalledTimes(1); + + completeOpenSession(null, { status: { code: 200 }, sessionId: 123 }); + await Promise.all([firstOpen, secondOpen]); + + expect(requestStatementId).toHaveBeenCalledTimes(1); + expect(connection.isOpen()).toBe(true); + + await connection.close(); + }); + test("Should tear down the socket when session setup fails", async () => { // openSession rejects after the TCP connection was established. thriftMock.createClient.mockReturnValueOnce({ @@ -193,4 +265,34 @@ describe("Connection", () => { // close()); getSessionId() throws once the id is cleared. expect(() => connection.getSessionId()).toThrow("Session is not open"); }); + + test("Should allow open to be retried after a failed attempt", async () => { + thriftMock.createClient.mockReturnValueOnce({ + openSession: jest.fn( + ( + _req: unknown, + callback: (error: Error | null, response: unknown) => void, + ) => callback(new Error("temporary failure"), null), + ), + requestStatementId: jest.fn(), + closeSession: jest.fn(), + }); + + const connection = new Connection({ + host: "localhost", + port: 6667, + username: "root", + password: "root", + enableSSL: false, + sqlDialect: "tree", + }); + + await expect(connection.open()).rejects.toThrow("temporary failure"); + await connection.open(); + + expect(thriftMock.createConnection).toHaveBeenCalledTimes(2); + expect(connection.isOpen()).toBe(true); + + await connection.close(); + }); });
