From 438199eb09d4cf4ab3422673a37a6455ef0811c3 Mon Sep 17 00:00:00 2001 From: menshibin Date: Wed, 11 Jun 2025 21:08:46 +0800 Subject: [PATCH 1/8] fix: clear invalid connections from the connection pool --- nodejs/src/client/wsConnectorPool.ts | 16 +++++++++++----- nodejs/test/bulkPulling/wsConnectPool.test.ts | 9 +++++++++ 2 files changed, 20 insertions(+), 5 deletions(-) diff --git a/nodejs/src/client/wsConnectorPool.ts b/nodejs/src/client/wsConnectorPool.ts index 2b13d31c..fd2b3f71 100644 --- a/nodejs/src/client/wsConnectorPool.ts +++ b/nodejs/src/client/wsConnectorPool.ts @@ -34,13 +34,19 @@ export class WebSocketConnectionPool { const unlock = await mutex.acquire() try { if (this.pool.has(connectAddr)) { - let connectors = this.pool.get(connectAddr); - if (connectors) { - if (connectors.length > 0) { - connector = connectors.pop(); + const connectors = this.pool.get(connectAddr); + while (connectors && connectors.length > 0) { + const candidate = connectors.pop(); + if (candidate && candidate.readyState() > 0) { // 1: OPEN + connector = candidate; + break; + } else if (candidate) { + Atomics.add(WebSocketConnectionPool.sharedArray, 0, -1); + candidate.close(); + logger.error(`getConnection, current connection status fail, url: ${connectAddr}`) } } - } + } if (connector) { logger.debug("get connection success:" + Atomics.load(WebSocketConnectionPool.sharedArray, 0)); diff --git a/nodejs/test/bulkPulling/wsConnectPool.test.ts b/nodejs/test/bulkPulling/wsConnectPool.test.ts index ec23cce4..fde6e2ef 100644 --- a/nodejs/test/bulkPulling/wsConnectPool.test.ts +++ b/nodejs/test/bulkPulling/wsConnectPool.test.ts @@ -149,6 +149,15 @@ describe('TDWebSocket.WsSql()', () => { await Promise.all(allp) console.log(stmtIds) }); + + test('retry connect', async() => { + for (let i =0; i < 20; i++) { + await tmqConnect() + await Sleep(1000); + } + console.log(stmtIds) + }); + }) afterAll(async () => { From b61c09c06319a996cd859d97930d4efd9e2401cf Mon Sep 17 00:00:00 2001 From: menshibin Date: Fri, 13 Jun 2025 10:51:39 +0800 Subject: [PATCH 2/8] fix: fix connection pool allocation with unavailable connections --- .github/workflows/build.yaml | 2 + nodejs/package-lock.json | 5 +- nodejs/package.json | 3 +- nodejs/src/client/wsClient.ts | 14 ++-- nodejs/src/client/wsConnectorPool.ts | 7 +- nodejs/src/tmq/wsTmq.ts | 2 +- nodejs/test/bulkPulling/cloud.tmq.test.ts | 66 +++++++++++++++++++ nodejs/test/bulkPulling/wsConnectPool.test.ts | 14 +--- 8 files changed, 88 insertions(+), 25 deletions(-) create mode 100644 nodejs/test/bulkPulling/cloud.tmq.test.ts diff --git a/.github/workflows/build.yaml b/.github/workflows/build.yaml index 233f3a5f..f85bc9ad 100644 --- a/.github/workflows/build.yaml +++ b/.github/workflows/build.yaml @@ -76,6 +76,8 @@ jobs: - name: test nodejs websocket working-directory: nodejs-connector/nodejs run: | + export WS_CLOUD_URL=${{ secrets.WS_CLOUD_URL }} + export WS_CLOUD_TOKEN=${{ secrets.WS_CLOUD_TOKEN }} ls -al npm install npm list diff --git a/nodejs/package-lock.json b/nodejs/package-lock.json index da154832..faf0a46a 100644 --- a/nodejs/package-lock.json +++ b/nodejs/package-lock.json @@ -26,7 +26,7 @@ "@types/node": "^18.0.0", "@types/uuid": "^9.0.8", "@types/websocket": "^1.0", - "jest": "^29.2.2", + "jest": "^29.7.0", "parcel": "^2.7.0", "qingwa": "^1.0.7", "ts-jest": "^29.0.3", @@ -3715,8 +3715,9 @@ }, "node_modules/jest": { "version": "29.7.0", + "resolved": "https://registry.npmjs.org/jest/-/jest-29.7.0.tgz", + "integrity": "sha512-NIy3oAFp9shda19hy4HK0HRTWKtPJmGdnvywu01nOqNC2vZg+Z+fvJDxpMQA88eb2I9EcafcdjYgsDthnYTvGw==", "dev": true, - "license": "MIT", "dependencies": { "@jest/core": "^29.7.0", "@jest/types": "^29.6.3", diff --git a/nodejs/package.json b/nodejs/package.json index ac67cdfa..4ffe1b57 100644 --- a/nodejs/package.json +++ b/nodejs/package.json @@ -47,7 +47,6 @@ "winston": "^3.17.0", "winston-daily-rotate-file": "^5.0.0" }, - "devDependencies": { "@parcel/packager-ts": "^2.7.0", "@parcel/transformer-typescript-types": "^2.7.0", @@ -56,7 +55,7 @@ "@types/node": "^18.0.0", "@types/uuid": "^9.0.8", "@types/websocket": "^1.0", - "jest": "^29.2.2", + "jest": "^29.7.0", "parcel": "^2.7.0", "qingwa": "^1.0.7", "ts-jest": "^29.0.3", diff --git a/nodejs/src/client/wsClient.ts b/nodejs/src/client/wsClient.ts index e94355fb..be029ecb 100644 --- a/nodejs/src/client/wsClient.ts +++ b/nodejs/src/client/wsClient.ts @@ -40,7 +40,7 @@ export class WsClient { }; this._wsConnector = await WebSocketConnectionPool.instance().getConnection(this._url, this._timeout); - if (this._wsConnector.readyState() > 0) { + if (this._wsConnector.readyState() === 1) { return; } try { @@ -63,7 +63,7 @@ export class WsClient { async execNoResp(queryMsg: string): Promise { logger.debug('[wsQueryInterface.query.queryMsg]===>' + queryMsg); - if (this._wsConnector && this._wsConnector.readyState() > 0) { + if (this._wsConnector && this._wsConnector.readyState() === 1) { await this._wsConnector.sendMsgNoResp(queryMsg) return; } @@ -74,7 +74,7 @@ export class WsClient { async exec(queryMsg: string, bSqlQuery:boolean = true): Promise { return new Promise((resolve, reject) => { logger.debug('[wsQueryInterface.query.queryMsg]===>' + queryMsg); - if (this._wsConnector && this._wsConnector.readyState() > 0) { + if (this._wsConnector && this._wsConnector.readyState() === 1) { this._wsConnector.sendMsg(queryMsg).then((e: any) => { if (e.msg.code == 0) { if (bSqlQuery) { @@ -96,7 +96,7 @@ export class WsClient { // need to construct Response. async sendBinaryMsg(reqId: bigint, action:string, message: ArrayBuffer, bSqlQuery:boolean = true, bResultBinary: boolean = false): Promise { return new Promise((resolve, reject) => { - if (this._wsConnector && this._wsConnector.readyState() > 0) { + if (this._wsConnector && this._wsConnector.readyState() === 1) { this._wsConnector.sendBinaryMsg(reqId, action, message).then((e: any) => { if (bResultBinary) { resolve(e); @@ -131,7 +131,7 @@ export class WsClient { async ready(): Promise { try { this._wsConnector = await WebSocketConnectionPool.instance().getConnection(this._url, this._timeout); - if (this._wsConnector.readyState() <= 0) { + if (this._wsConnector.readyState() !== 1) { await this._wsConnector.ready() } logger.debug("ready status ", this._url, this._wsConnector.readyState()) @@ -146,7 +146,7 @@ export class WsClient { async sendMsg(msg:string): Promise { return new Promise((resolve, reject) => { logger.debug("[wsQueryInterface.sendMsg]===>" + msg) - if (this._wsConnector && this._wsConnector.readyState() > 0) { + if (this._wsConnector && this._wsConnector.readyState() === 1) { this._wsConnector.sendMsg(msg).then((e: any) => { resolve(e); }).catch((e) => reject(e)); @@ -167,7 +167,7 @@ export class WsClient { return new Promise((resolve, reject) => { let jsonStr = JSONBig.stringify(freeResultMsg); logger.debug("[wsQueryInterface.freeResult.freeResultMsg]===>" + jsonStr) - if (this._wsConnector && this._wsConnector.readyState() > 0) { + if (this._wsConnector && this._wsConnector.readyState() === 1) { this._wsConnector.sendMsg(jsonStr, false) .then((e: any) => {resolve(e);}) .catch((e) => reject(e)); diff --git a/nodejs/src/client/wsConnectorPool.ts b/nodejs/src/client/wsConnectorPool.ts index fd2b3f71..7d35923e 100644 --- a/nodejs/src/client/wsConnectorPool.ts +++ b/nodejs/src/client/wsConnectorPool.ts @@ -37,7 +37,10 @@ export class WebSocketConnectionPool { const connectors = this.pool.get(connectAddr); while (connectors && connectors.length > 0) { const candidate = connectors.pop(); - if (candidate && candidate.readyState() > 0) { // 1: OPEN + if (!candidate) { + continue; + } + if (candidate && candidate.readyState() === 1) { // 1: OPEN connector = candidate; break; } else if (candidate) { @@ -68,7 +71,7 @@ export class WebSocketConnectionPool { if (connector) { const unlock = await mutex.acquire(); try { - if (connector.readyState() > 0) { + if (connector.readyState() === 1) { let url = connector.getWsURL(); let connectAddr = url.origin.concat(url.pathname).concat(url.search) let connectors = this.pool.get(connectAddr); diff --git a/nodejs/src/tmq/wsTmq.ts b/nodejs/src/tmq/wsTmq.ts index a2be0998..9c3d4199 100644 --- a/nodejs/src/tmq/wsTmq.ts +++ b/nodejs/src/tmq/wsTmq.ts @@ -32,7 +32,7 @@ export class WsConsumer { await wsSql.connect(); await wsSql.checkVersion(); await this._wsClient.ready(); - }else { + } else { throw(new TDWebSocketClientError(ErrorCode.ERR_WEBSOCKET_CONNECTION_FAIL, `connection creation failed, url: ${this._wsConfig.url}`)); } }catch (e: any) { diff --git a/nodejs/test/bulkPulling/cloud.tmq.test.ts b/nodejs/test/bulkPulling/cloud.tmq.test.ts new file mode 100644 index 00000000..92952495 --- /dev/null +++ b/nodejs/test/bulkPulling/cloud.tmq.test.ts @@ -0,0 +1,66 @@ + +import { TMQConstants } from "../../src/tmq/constant"; +import { WsConsumer } from "../../src/tmq/wsTmq"; +import { WebSocketConnectionPool } from "../../src/client/wsConnectorPool"; +import logger, { setLevel } from "../../src/common/log" + +beforeAll(async () => { + +}) + + +describe('TDWebSocket.Tmq()', () => { + jest.setTimeout(20 * 1000) + + test('normal connect', async() => { + const url = `wss://${process.env.TDENGINE_CLOUD_URL}?token=${process.env.TDENGINE_CLOUD_TOKEN}`; + // const TDENGINE_CLOUD_URL = 'wss://gw.cloud.taosdata.com?token=1eb78307be0681ac2fc07c2817ba8a9719641fb9'; + const topic = 'chenyu'; + const topics = [topic]; + const groupId = 'group1'; + const clientId = 'client1'; + let configMap = new Map([ + [TMQConstants.GROUP_ID, groupId], + [TMQConstants.CLIENT_ID, clientId], + [TMQConstants.AUTO_OFFSET_RESET, 'earliest'], + [TMQConstants.WS_URL, url], + [TMQConstants.ENABLE_AUTO_COMMIT, 'true'], + [TMQConstants.AUTO_COMMIT_INTERVAL_MS, '1000'], + ]) + + setLevel("debug"); + // create consumer + let consumer = await WsConsumer.newConsumer(configMap); + console.log( + `Create consumer successfully, host: ${url}, groupId: ${groupId}, clientId: ${clientId}` + ); + // subscribe + await consumer.subscribe(topics); + console.log(`Subscribe topics successfully, topics: ${topics}`); + let res = new Map(); + while (res.size == 0) { + // poll + res = await consumer.poll(1000); + for (let [key, value] of res) { + // Add your data processing logic here + console.log(`data: ${key} ${value}`); + } + // commit + await consumer.commit(); + } + + // seek + let assignment = await consumer.assignment(); + await consumer.seekToBeginning(assignment); + console.log('Assignment seek to beginning successfully'); + + // clean + await consumer.unsubscribe(); + await consumer.close(); + + }); +}) + +afterAll(async () => { + WebSocketConnectionPool.instance().destroyed() +}) \ No newline at end of file diff --git a/nodejs/test/bulkPulling/wsConnectPool.test.ts b/nodejs/test/bulkPulling/wsConnectPool.test.ts index fde6e2ef..3e70166f 100644 --- a/nodejs/test/bulkPulling/wsConnectPool.test.ts +++ b/nodejs/test/bulkPulling/wsConnectPool.test.ts @@ -5,7 +5,7 @@ import { WsSql } from "../../src/sql/wsSql"; import { TMQConstants } from "../../src/tmq/constant"; import { WsConsumer } from "../../src/tmq/wsTmq"; import { Sleep } from "../utils"; -import logger, { setLevel } from "../../src/common/log" +import { setLevel } from "../../src/common/log" let dsn = 'ws://root:taosdata@localhost:6041'; let tags = ['California.SanFrancisco', 3]; @@ -86,7 +86,7 @@ async function tmqConnect() { consumer = await WsConsumer.newConsumer(configMap); await consumer.subscribe(topics); - let res = await consumer.poll(500); + let res = await consumer.poll(100); for (let [key, value] of res) { console.log(key, value.getMeta()); let data = value.getData(); @@ -130,7 +130,7 @@ beforeAll(async () => { }) describe('TDWebSocket.WsSql()', () => { - jest.setTimeout(20 * 1000) + jest.setTimeout(60 * 1000) test('ReqId', async()=> { const allp:any[] = [] for (let i =0; i < 10; i++) { @@ -150,14 +150,6 @@ describe('TDWebSocket.WsSql()', () => { console.log(stmtIds) }); - test('retry connect', async() => { - for (let i =0; i < 20; i++) { - await tmqConnect() - await Sleep(1000); - } - console.log(stmtIds) - }); - }) afterAll(async () => { From 6bc38033fd17794f0a7794bc8df702615fc46a9b Mon Sep 17 00:00:00 2001 From: menshibin Date: Fri, 13 Jun 2025 11:11:47 +0800 Subject: [PATCH 3/8] fix: modify cloud tmq case for insert data --- .github/workflows/build.yaml | 4 ++-- nodejs/test/bulkPulling/cloud.tmq.test.ts | 27 ++++++++++++++++++++--- 2 files changed, 26 insertions(+), 5 deletions(-) diff --git a/.github/workflows/build.yaml b/.github/workflows/build.yaml index f85bc9ad..ac1f27cd 100644 --- a/.github/workflows/build.yaml +++ b/.github/workflows/build.yaml @@ -76,8 +76,8 @@ jobs: - name: test nodejs websocket working-directory: nodejs-connector/nodejs run: | - export WS_CLOUD_URL=${{ secrets.WS_CLOUD_URL }} - export WS_CLOUD_TOKEN=${{ secrets.WS_CLOUD_TOKEN }} + export TDENGINE_CLOUD_URL=${{ secrets.TDENGINE_CLOUD_URL }} + export TDENGINE_CLOUD_TOKEN=${{ secrets.TDENGINE_CLOUD_TOKEN }} ls -al npm install npm list diff --git a/nodejs/test/bulkPulling/cloud.tmq.test.ts b/nodejs/test/bulkPulling/cloud.tmq.test.ts index 92952495..93d7798c 100644 --- a/nodejs/test/bulkPulling/cloud.tmq.test.ts +++ b/nodejs/test/bulkPulling/cloud.tmq.test.ts @@ -3,19 +3,40 @@ import { TMQConstants } from "../../src/tmq/constant"; import { WsConsumer } from "../../src/tmq/wsTmq"; import { WebSocketConnectionPool } from "../../src/client/wsConnectorPool"; import logger, { setLevel } from "../../src/common/log" +import { WSConfig } from "../../src/common/config"; +import { WsSql } from "../../src/sql/wsSql"; beforeAll(async () => { - +const url = `wss://${process.env.TDENGINE_CLOUD_URL}?token=${process.env.TDENGINE_CLOUD_TOKEN}`; +let wsSql = null; + try { + const conf = new WSConfig(url); + conf.setUser('root') + conf.setPwd('taosdata') + wsSql = await WsSql.open(conf) + let sql = `INSERT INTO dmeters.d1001 USING dmeters.meters (groupid, location) TAGS(2, 'SanFrancisco') + VALUES (NOW + 1a, 10.30000, 219, 0.31000) (NOW + 2a, 12.60000, 218, 0.33000) (NOW + 3a, 12.30000, 221, 0.31000) + dmeters.d1002 USING dmeters.meters (groupid, location) TAGS(3, 'SanFrancisco') + VALUES (NOW + 1a, 10.30000, 218, 0.25000)` + let res = await wsSql.exec(sql); + console.log(res); + expect(res.getAffectRows()).toBeGreaterThanOrEqual(3); + } catch (err) { + throw err; + } finally { + if (wsSql) { + await wsSql.close(); + } + } }) - describe('TDWebSocket.Tmq()', () => { jest.setTimeout(20 * 1000) test('normal connect', async() => { const url = `wss://${process.env.TDENGINE_CLOUD_URL}?token=${process.env.TDENGINE_CLOUD_TOKEN}`; // const TDENGINE_CLOUD_URL = 'wss://gw.cloud.taosdata.com?token=1eb78307be0681ac2fc07c2817ba8a9719641fb9'; - const topic = 'chenyu'; + const topic = 'topic_meters'; const topics = [topic]; const groupId = 'group1'; const clientId = 'client1'; From 81b208bbf2a81870cd56aa5502c5cc2b808343b4 Mon Sep 17 00:00:00 2001 From: menshibin Date: Fri, 13 Jun 2025 14:55:28 +0800 Subject: [PATCH 4/8] fix: delete invalid code --- nodejs/src/client/wsConnector.ts | 7 ------- nodejs/src/tmq/config.ts | 2 +- 2 files changed, 1 insertion(+), 8 deletions(-) diff --git a/nodejs/src/client/wsConnector.ts b/nodejs/src/client/wsConnector.ts index b0a5d8d3..e08f0e81 100644 --- a/nodejs/src/client/wsConnector.ts +++ b/nodejs/src/client/wsConnector.ts @@ -59,13 +59,6 @@ export class WebSocketConnector { WsEventCallback.instance().handleEventCallback({id:id, action:'', req_id:BigInt(0)}, OnMessageType.MESSAGE_TYPE_ARRAYBUFFER, data); - } else if (Object.prototype.toString.call(data) === '[object Blob]') { - data.arrayBuffer().then((d: ArrayBuffer) => { - let id = new DataView(d, 8, 8).getBigUint64(0, true); - WsEventCallback.instance().handleEventCallback({id:id, action:'', req_id:BigInt(0)}, - OnMessageType.MESSAGE_TYPE_BLOB, d); - }) - } else if (Object.prototype.toString.call(data) === '[object String]') { let msg = JSON.parse(data) logger.debug("[_onmessage.stringType]==>:" + data); diff --git a/nodejs/src/tmq/config.ts b/nodejs/src/tmq/config.ts index 6fa00481..272f8b5d 100644 --- a/nodejs/src/tmq/config.ts +++ b/nodejs/src/tmq/config.ts @@ -10,7 +10,7 @@ export class TmqConfig { client_id: string | null = null; offset_rest: string | null = null; topics?: Array; - auto_commit: boolean = false; + auto_commit: boolean = true; auto_commit_interval_ms: number = 5 * 1000; timeout: number = 5000; otherConfigs: Map; From cb0f20f2728eb3a74fe6de23584a2a72c7c41a99 Mon Sep 17 00:00:00 2001 From: menshibin Date: Fri, 13 Jun 2025 16:15:07 +0800 Subject: [PATCH 5/8] fix: add config params test --- nodejs/test/bulkPulling/sql.test.ts | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/nodejs/test/bulkPulling/sql.test.ts b/nodejs/test/bulkPulling/sql.test.ts index b798b372..9e138aa5 100644 --- a/nodejs/test/bulkPulling/sql.test.ts +++ b/nodejs/test/bulkPulling/sql.test.ts @@ -26,10 +26,12 @@ describe('TDWebSocket.WsSql()', () => { jest.setTimeout(20 * 1000) test('normal connect', async() => { let wsSql = null; - let conf :WSConfig = new WSConfig(dns) - conf.setUser('root') - conf.setPwd('taosdata') - conf.setDb('power') + let conf :WSConfig = new WSConfig() + conf.setUrl(dns); + conf.setUser('root'); + conf.setPwd('taosdata'); + conf.setDb('power'); + conf.setTimeOut(6000); wsSql = await WsSql.open(conf) expect(wsSql.state()).toBeGreaterThan(0) await wsSql.close(); From a870c25beaf7a5675442d983640f96f91071769d Mon Sep 17 00:00:00 2001 From: menshibin Date: Fri, 13 Jun 2025 16:21:34 +0800 Subject: [PATCH 6/8] fix: add config url params test --- nodejs/test/bulkPulling/sql.test.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/nodejs/test/bulkPulling/sql.test.ts b/nodejs/test/bulkPulling/sql.test.ts index 9e138aa5..dfd9c2e6 100644 --- a/nodejs/test/bulkPulling/sql.test.ts +++ b/nodejs/test/bulkPulling/sql.test.ts @@ -26,7 +26,7 @@ describe('TDWebSocket.WsSql()', () => { jest.setTimeout(20 * 1000) test('normal connect', async() => { let wsSql = null; - let conf :WSConfig = new WSConfig() + let conf :WSConfig = new WSConfig(''); conf.setUrl(dns); conf.setUser('root'); conf.setPwd('taosdata'); From 5a85e7cc33ee666d9d3dae56623e2252049d9035 Mon Sep 17 00:00:00 2001 From: menshibin Date: Fri, 13 Jun 2025 16:55:09 +0800 Subject: [PATCH 7/8] fix: Delete invalid connection classes --- nodejs/src/client/wsResponse.ts | 24 ------------------------ 1 file changed, 24 deletions(-) diff --git a/nodejs/src/client/wsResponse.ts b/nodejs/src/client/wsResponse.ts index 205cb43c..40cbd764 100644 --- a/nodejs/src/client/wsResponse.ts +++ b/nodejs/src/client/wsResponse.ts @@ -106,27 +106,3 @@ export class WSFetchBlockResponse { } } - -interface IWSConnResponse { - code: number; - message: string; - action: string; - req_id: number; - timing: bigint; -} - -export class WSConnResponse { - code: number; - message: string; - action: string; - req_id: number; - timing: bigint; - - constructor(msg: IWSConnResponse) { - this.code = msg.code; - this.message = msg.message; - this.action = msg.action; - this.req_id = msg.req_id; - this.timing = BigInt(msg.timing); - } -} From 6606b9c86e8104068fe9253f358ffc28a3965b08 Mon Sep 17 00:00:00 2001 From: menshibin Date: Fri, 13 Jun 2025 17:48:33 +0800 Subject: [PATCH 8/8] fix: use w3cwebsocket.OPEN for status judgment --- nodejs/src/client/wsClient.ts | 17 +++++++++-------- nodejs/src/client/wsConnector.ts | 6 +++--- nodejs/src/client/wsConnectorPool.ts | 5 +++-- 3 files changed, 15 insertions(+), 13 deletions(-) diff --git a/nodejs/src/client/wsClient.ts b/nodejs/src/client/wsClient.ts index be029ecb..ecc1d87b 100644 --- a/nodejs/src/client/wsClient.ts +++ b/nodejs/src/client/wsClient.ts @@ -9,6 +9,7 @@ import { import { ReqId } from '../common/reqid'; import logger from '../common/log'; import { safeDecodeURIComponent, compareVersions} from '../common/utils'; +import { w3cwebsocket } from 'websocket'; export class WsClient { private _wsConnector?: WebSocketConnector; @@ -40,7 +41,7 @@ export class WsClient { }; this._wsConnector = await WebSocketConnectionPool.instance().getConnection(this._url, this._timeout); - if (this._wsConnector.readyState() === 1) { + if (this._wsConnector.readyState() === w3cwebsocket.OPEN) { return; } try { @@ -63,7 +64,7 @@ export class WsClient { async execNoResp(queryMsg: string): Promise { logger.debug('[wsQueryInterface.query.queryMsg]===>' + queryMsg); - if (this._wsConnector && this._wsConnector.readyState() === 1) { + if (this._wsConnector && this._wsConnector.readyState() === w3cwebsocket.OPEN) { await this._wsConnector.sendMsgNoResp(queryMsg) return; } @@ -74,7 +75,7 @@ export class WsClient { async exec(queryMsg: string, bSqlQuery:boolean = true): Promise { return new Promise((resolve, reject) => { logger.debug('[wsQueryInterface.query.queryMsg]===>' + queryMsg); - if (this._wsConnector && this._wsConnector.readyState() === 1) { + if (this._wsConnector && this._wsConnector.readyState() === w3cwebsocket.OPEN) { this._wsConnector.sendMsg(queryMsg).then((e: any) => { if (e.msg.code == 0) { if (bSqlQuery) { @@ -96,7 +97,7 @@ export class WsClient { // need to construct Response. async sendBinaryMsg(reqId: bigint, action:string, message: ArrayBuffer, bSqlQuery:boolean = true, bResultBinary: boolean = false): Promise { return new Promise((resolve, reject) => { - if (this._wsConnector && this._wsConnector.readyState() === 1) { + if (this._wsConnector && this._wsConnector.readyState() === w3cwebsocket.OPEN) { this._wsConnector.sendBinaryMsg(reqId, action, message).then((e: any) => { if (bResultBinary) { resolve(e); @@ -131,7 +132,7 @@ export class WsClient { async ready(): Promise { try { this._wsConnector = await WebSocketConnectionPool.instance().getConnection(this._url, this._timeout); - if (this._wsConnector.readyState() !== 1) { + if (this._wsConnector.readyState() !== w3cwebsocket.OPEN) { await this._wsConnector.ready() } logger.debug("ready status ", this._url, this._wsConnector.readyState()) @@ -146,7 +147,7 @@ export class WsClient { async sendMsg(msg:string): Promise { return new Promise((resolve, reject) => { logger.debug("[wsQueryInterface.sendMsg]===>" + msg) - if (this._wsConnector && this._wsConnector.readyState() === 1) { + if (this._wsConnector && this._wsConnector.readyState() === w3cwebsocket.OPEN) { this._wsConnector.sendMsg(msg).then((e: any) => { resolve(e); }).catch((e) => reject(e)); @@ -167,7 +168,7 @@ export class WsClient { return new Promise((resolve, reject) => { let jsonStr = JSONBig.stringify(freeResultMsg); logger.debug("[wsQueryInterface.freeResult.freeResultMsg]===>" + jsonStr) - if (this._wsConnector && this._wsConnector.readyState() === 1) { + if (this._wsConnector && this._wsConnector.readyState() === w3cwebsocket.OPEN) { this._wsConnector.sendMsg(jsonStr, false) .then((e: any) => {resolve(e);}) .catch((e) => reject(e)); @@ -187,7 +188,7 @@ export class WsClient { if (this._wsConnector) { try { - if (this._wsConnector.readyState() <= 0) { + if (this._wsConnector.readyState() !== w3cwebsocket.OPEN) { await this._wsConnector.ready(); } let result:any = await this._wsConnector.sendMsg(JSONBig.stringify(versionMsg)); diff --git a/nodejs/src/client/wsConnector.ts b/nodejs/src/client/wsConnector.ts index e08f0e81..84323a0d 100644 --- a/nodejs/src/client/wsConnector.ts +++ b/nodejs/src/client/wsConnector.ts @@ -91,7 +91,7 @@ export class WebSocketConnector { } return new Promise((resolve, reject) => { - if (this._wsConn && this._wsConn.readyState > 0) { + if (this._wsConn && this._wsConn.readyState === w3cwebsocket.OPEN) { this._wsConn.send(message) resolve() } else { @@ -110,7 +110,7 @@ export class WebSocketConnector { } return new Promise((resolve, reject) => { - if (this._wsConn && this._wsConn.readyState > 0) { + if (this._wsConn && this._wsConn.readyState === w3cwebsocket.OPEN) { if (register) { WsEventCallback.instance().registerCallback({ action: msg.action, req_id: msg.args.req_id, timeout:this._timeout, id: msg.args.id === undefined ? msg.args.id : BigInt(msg.args.id) }, @@ -127,7 +127,7 @@ export class WebSocketConnector { async sendBinaryMsg(reqId: bigint, action:string, message: ArrayBuffer, register: Boolean = true) { return new Promise((resolve, reject) => { - if (this._wsConn && this._wsConn.readyState > 0) { + if (this._wsConn && this._wsConn.readyState === w3cwebsocket.OPEN) { if (register) { WsEventCallback.instance().registerCallback({ action: action, req_id: reqId, timeout:this._timeout, id: reqId}, resolve, reject); diff --git a/nodejs/src/client/wsConnectorPool.ts b/nodejs/src/client/wsConnectorPool.ts index 7d35923e..6124993c 100644 --- a/nodejs/src/client/wsConnectorPool.ts +++ b/nodejs/src/client/wsConnectorPool.ts @@ -2,6 +2,7 @@ import { Mutex } from "async-mutex"; import { WebSocketConnector } from "./wsConnector"; import { ErrorCode, TDWebSocketClientError } from "../common/wsError"; import logger from "../common/log"; +import { w3cwebsocket } from "websocket"; const mutex = new Mutex(); @@ -40,7 +41,7 @@ export class WebSocketConnectionPool { if (!candidate) { continue; } - if (candidate && candidate.readyState() === 1) { // 1: OPEN + if (candidate && candidate.readyState() === w3cwebsocket.OPEN) { connector = candidate; break; } else if (candidate) { @@ -71,7 +72,7 @@ export class WebSocketConnectionPool { if (connector) { const unlock = await mutex.acquire(); try { - if (connector.readyState() === 1) { + if (connector.readyState() === w3cwebsocket.OPEN) { let url = connector.getWsURL(); let connectAddr = url.origin.concat(url.pathname).concat(url.search) let connectors = this.pool.get(connectAddr);