fix: network anomalies where the connections allocated by the connection pool are unavailable

.github/workflows/build.yaml

@@ -76,6 +76,8 @@ jobs:
     - name: test nodejs websocket
       working-directory: nodejs-connector/nodejs
       run: |
+        export TDENGINE_CLOUD_URL=${{ secrets.TDENGINE_CLOUD_URL }}
+        export TDENGINE_CLOUD_TOKEN=${{ secrets.TDENGINE_CLOUD_TOKEN }}
         ls -al
         npm install
         npm list

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",

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() > 0) {
+        if (this._wsConnector.readyState() === w3cwebsocket.OPEN) {
             return;
         } 
         try {
@@ -63,7 +64,7 @@ export class WsClient {
 
     async execNoResp(queryMsg: string): Promise<void> {
         logger.debug('[wsQueryInterface.query.queryMsg]===>' + queryMsg);
-        if (this._wsConnector && this._wsConnector.readyState() > 0) {
+        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<any> {
         return new Promise((resolve, reject) => {
             logger.debug('[wsQueryInterface.query.queryMsg]===>' + queryMsg);
-            if (this._wsConnector && this._wsConnector.readyState() > 0) {
+            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<any> {
         return new Promise((resolve, reject) => {
-            if (this._wsConnector && this._wsConnector.readyState() > 0) {
+            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<void> {
         try {
             this._wsConnector = await WebSocketConnectionPool.instance().getConnection(this._url, this._timeout);
-            if (this._wsConnector.readyState() <= 0) {
+            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<any> {
         return new Promise((resolve, reject) => {
             logger.debug("[wsQueryInterface.sendMsg]===>" + msg)
-            if (this._wsConnector && this._wsConnector.readyState() > 0) {
+            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() > 0) {
+            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)); 

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);
@@ -98,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 {
@@ -117,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) }, 
@@ -134,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);

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();
 
@@ -34,13 +35,22 @@ 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) {
+                        continue;
+                    }
+                    if (candidate && candidate.readyState() === w3cwebsocket.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));
@@ -62,7 +72,7 @@ export class WebSocketConnectionPool {
         if (connector) {
             const unlock = await mutex.acquire();
             try {
-                if (connector.readyState() > 0) {
+                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);

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);
-    }
-}

nodejs/src/tmq/config.ts

@@ -10,7 +10,7 @@ export class TmqConfig {
     client_id: string | null = null;
     offset_rest: string | null = null;
     topics?: Array<string>;
-    auto_commit: boolean = false;
+    auto_commit: boolean = true;
     auto_commit_interval_ms: number = 5 * 1000;
     timeout: number = 5000;
     otherConfigs: Map<string, any>;

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) {

nodejs/test/bulkPulling/cloud.tmq.test.ts

@@ -0,0 +1,87 @@
+
+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 = 'topic_meters';
+        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()
+})

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();

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++) {
@@ -149,6 +149,7 @@ describe('TDWebSocket.WsSql()', () => {
         await Promise.all(allp)
         console.log(stmtIds)
     });
+
 })
 
 afterAll(async () => {