This is an automated email from the ASF dual-hosted git repository.
HTHou pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/iotdb-client-nodejs.git
The following commit(s) were added to refs/heads/develop by this push:
new 96ba79c fix: make session open idempotent (#23)
96ba79c is described below
commit 96ba79c1090717b2f373b252962f6c93f985f17c
Author: CritasWang <[email protected]>
AuthorDate: Fri Jul 31 10:16:01 2026 +0800
fix: make session open idempotent (#23)
---
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();
+ });
});
