Skip to content

Commit 8c7a6ce

Browse files
menshibinmenshibin
authored
fix: network anomalies where the connections allocated by the connection pool are unavailable (#85)
* fix: clear invalid connections from the connection pool * fix: fix connection pool allocation with unavailable connections * fix: modify cloud tmq case for insert data * fix: delete invalid code * fix: add config params test * fix: add config url params test * fix: Delete invalid connection classes * fix: use w3cwebsocket.OPEN for status judgment
1 parent ee4f4e9 commit 8c7a6ce

12 files changed

Lines changed: 133 additions & 61 deletions

File tree

‎.github/workflows/build.yaml‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -76,6 +76,8 @@ jobs:
7676
- name: test nodejs websocket
7777
working-directory: nodejs-connector/nodejs
7878
run: |
79+
export TDENGINE_CLOUD_URL=${{ secrets.TDENGINE_CLOUD_URL }}
80+
export TDENGINE_CLOUD_TOKEN=${{ secrets.TDENGINE_CLOUD_TOKEN }}
7981
ls -al
8082
npm install
8183
npm list

‎nodejs/package-lock.json‎

Lines changed: 3 additions & 2 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎nodejs/package.json‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -47,7 +47,6 @@
4747
"winston": "^3.17.0",
4848
"winston-daily-rotate-file": "^5.0.0"
4949
},
50-
5150
"devDependencies": {
5251
"@parcel/packager-ts": "^2.7.0",
5352
"@parcel/transformer-typescript-types": "^2.7.0",
@@ -56,7 +55,7 @@
5655
"@types/node": "^18.0.0",
5756
"@types/uuid": "^9.0.8",
5857
"@types/websocket": "^1.0",
59-
"jest": "^29.2.2",
58+
"jest": "^29.7.0",
6059
"parcel": "^2.7.0",
6160
"qingwa": "^1.0.7",
6261
"ts-jest": "^29.0.3",

‎nodejs/src/client/wsClient.ts‎

Lines changed: 9 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@ import {
99
import { ReqId } from '../common/reqid';
1010
import logger from '../common/log';
1111
import { safeDecodeURIComponent, compareVersions} from '../common/utils';
12+
import { w3cwebsocket } from 'websocket';
1213

1314
export class WsClient {
1415
private _wsConnector?: WebSocketConnector;
@@ -40,7 +41,7 @@ export class WsClient {
4041
};
4142

4243
this._wsConnector = await WebSocketConnectionPool.instance().getConnection(this._url, this._timeout);
43-
if (this._wsConnector.readyState() > 0) {
44+
if (this._wsConnector.readyState() === w3cwebsocket.OPEN) {
4445
return;
4546
}
4647
try {
@@ -63,7 +64,7 @@ export class WsClient {
6364

6465
async execNoResp(queryMsg: string): Promise<void> {
6566
logger.debug('[wsQueryInterface.query.queryMsg]===>' + queryMsg);
66-
if (this._wsConnector && this._wsConnector.readyState() > 0) {
67+
if (this._wsConnector && this._wsConnector.readyState() === w3cwebsocket.OPEN) {
6768
await this._wsConnector.sendMsgNoResp(queryMsg)
6869
return;
6970
}
@@ -74,7 +75,7 @@ export class WsClient {
7475
async exec(queryMsg: string, bSqlQuery:boolean = true): Promise<any> {
7576
return new Promise((resolve, reject) => {
7677
logger.debug('[wsQueryInterface.query.queryMsg]===>' + queryMsg);
77-
if (this._wsConnector && this._wsConnector.readyState() > 0) {
78+
if (this._wsConnector && this._wsConnector.readyState() === w3cwebsocket.OPEN) {
7879
this._wsConnector.sendMsg(queryMsg).then((e: any) => {
7980
if (e.msg.code == 0) {
8081
if (bSqlQuery) {
@@ -96,7 +97,7 @@ export class WsClient {
9697
// need to construct Response.
9798
async sendBinaryMsg(reqId: bigint, action:string, message: ArrayBuffer, bSqlQuery:boolean = true, bResultBinary: boolean = false): Promise<any> {
9899
return new Promise((resolve, reject) => {
99-
if (this._wsConnector && this._wsConnector.readyState() > 0) {
100+
if (this._wsConnector && this._wsConnector.readyState() === w3cwebsocket.OPEN) {
100101
this._wsConnector.sendBinaryMsg(reqId, action, message).then((e: any) => {
101102
if (bResultBinary) {
102103
resolve(e);
@@ -131,7 +132,7 @@ export class WsClient {
131132
async ready(): Promise<void> {
132133
try {
133134
this._wsConnector = await WebSocketConnectionPool.instance().getConnection(this._url, this._timeout);
134-
if (this._wsConnector.readyState() <= 0) {
135+
if (this._wsConnector.readyState() !== w3cwebsocket.OPEN) {
135136
await this._wsConnector.ready()
136137
}
137138
logger.debug("ready status ", this._url, this._wsConnector.readyState())
@@ -146,7 +147,7 @@ export class WsClient {
146147
async sendMsg(msg:string): Promise<any> {
147148
return new Promise((resolve, reject) => {
148149
logger.debug("[wsQueryInterface.sendMsg]===>" + msg)
149-
if (this._wsConnector && this._wsConnector.readyState() > 0) {
150+
if (this._wsConnector && this._wsConnector.readyState() === w3cwebsocket.OPEN) {
150151
this._wsConnector.sendMsg(msg).then((e: any) => {
151152
resolve(e);
152153
}).catch((e) => reject(e));
@@ -167,7 +168,7 @@ export class WsClient {
167168
return new Promise((resolve, reject) => {
168169
let jsonStr = JSONBig.stringify(freeResultMsg);
169170
logger.debug("[wsQueryInterface.freeResult.freeResultMsg]===>" + jsonStr)
170-
if (this._wsConnector && this._wsConnector.readyState() > 0) {
171+
if (this._wsConnector && this._wsConnector.readyState() === w3cwebsocket.OPEN) {
171172
this._wsConnector.sendMsg(jsonStr, false)
172173
.then((e: any) => {resolve(e);})
173174
.catch((e) => reject(e));
@@ -187,7 +188,7 @@ export class WsClient {
187188

188189
if (this._wsConnector) {
189190
try {
190-
if (this._wsConnector.readyState() <= 0) {
191+
if (this._wsConnector.readyState() !== w3cwebsocket.OPEN) {
191192
await this._wsConnector.ready();
192193
}
193194
let result:any = await this._wsConnector.sendMsg(JSONBig.stringify(versionMsg));

‎nodejs/src/client/wsConnector.ts‎

Lines changed: 3 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -59,13 +59,6 @@ export class WebSocketConnector {
5959
WsEventCallback.instance().handleEventCallback({id:id, action:'', req_id:BigInt(0)},
6060
OnMessageType.MESSAGE_TYPE_ARRAYBUFFER, data);
6161

62-
} else if (Object.prototype.toString.call(data) === '[object Blob]') {
63-
data.arrayBuffer().then((d: ArrayBuffer) => {
64-
let id = new DataView(d, 8, 8).getBigUint64(0, true);
65-
WsEventCallback.instance().handleEventCallback({id:id, action:'', req_id:BigInt(0)},
66-
OnMessageType.MESSAGE_TYPE_BLOB, d);
67-
})
68-
6962
} else if (Object.prototype.toString.call(data) === '[object String]') {
7063
let msg = JSON.parse(data)
7164
logger.debug("[_onmessage.stringType]==>:" + data);
@@ -98,7 +91,7 @@ export class WebSocketConnector {
9891
}
9992

10093
return new Promise((resolve, reject) => {
101-
if (this._wsConn && this._wsConn.readyState > 0) {
94+
if (this._wsConn && this._wsConn.readyState === w3cwebsocket.OPEN) {
10295
this._wsConn.send(message)
10396
resolve()
10497
} else {
@@ -117,7 +110,7 @@ export class WebSocketConnector {
117110
}
118111

119112
return new Promise((resolve, reject) => {
120-
if (this._wsConn && this._wsConn.readyState > 0) {
113+
if (this._wsConn && this._wsConn.readyState === w3cwebsocket.OPEN) {
121114
if (register) {
122115
WsEventCallback.instance().registerCallback({ action: msg.action, req_id: msg.args.req_id,
123116
timeout:this._timeout, id: msg.args.id === undefined ? msg.args.id : BigInt(msg.args.id) },
@@ -134,7 +127,7 @@ export class WebSocketConnector {
134127

135128
async sendBinaryMsg(reqId: bigint, action:string, message: ArrayBuffer, register: Boolean = true) {
136129
return new Promise((resolve, reject) => {
137-
if (this._wsConn && this._wsConn.readyState > 0) {
130+
if (this._wsConn && this._wsConn.readyState === w3cwebsocket.OPEN) {
138131
if (register) {
139132
WsEventCallback.instance().registerCallback({ action: action, req_id: reqId,
140133
timeout:this._timeout, id: reqId}, resolve, reject);

‎nodejs/src/client/wsConnectorPool.ts‎

Lines changed: 16 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@ import { Mutex } from "async-mutex";
22
import { WebSocketConnector } from "./wsConnector";
33
import { ErrorCode, TDWebSocketClientError } from "../common/wsError";
44
import logger from "../common/log";
5+
import { w3cwebsocket } from "websocket";
56

67
const mutex = new Mutex();
78

@@ -34,13 +35,22 @@ export class WebSocketConnectionPool {
3435
const unlock = await mutex.acquire()
3536
try {
3637
if (this.pool.has(connectAddr)) {
37-
let connectors = this.pool.get(connectAddr);
38-
if (connectors) {
39-
if (connectors.length > 0) {
40-
connector = connectors.pop();
38+
const connectors = this.pool.get(connectAddr);
39+
while (connectors && connectors.length > 0) {
40+
const candidate = connectors.pop();
41+
if (!candidate) {
42+
continue;
43+
}
44+
if (candidate && candidate.readyState() === w3cwebsocket.OPEN) {
45+
connector = candidate;
46+
break;
47+
} else if (candidate) {
48+
Atomics.add(WebSocketConnectionPool.sharedArray, 0, -1);
49+
candidate.close();
50+
logger.error(`getConnection, current connection status fail, url: ${connectAddr}`)
4151
}
4252
}
43-
}
53+
}
4454

4555
if (connector) {
4656
logger.debug("get connection success:" + Atomics.load(WebSocketConnectionPool.sharedArray, 0));
@@ -62,7 +72,7 @@ export class WebSocketConnectionPool {
6272
if (connector) {
6373
const unlock = await mutex.acquire();
6474
try {
65-
if (connector.readyState() > 0) {
75+
if (connector.readyState() === w3cwebsocket.OPEN) {
6676
let url = connector.getWsURL();
6777
let connectAddr = url.origin.concat(url.pathname).concat(url.search)
6878
let connectors = this.pool.get(connectAddr);

‎nodejs/src/client/wsResponse.ts‎

Lines changed: 0 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -106,27 +106,3 @@ export class WSFetchBlockResponse {
106106

107107
}
108108
}
109-
110-
interface IWSConnResponse {
111-
code: number;
112-
message: string;
113-
action: string;
114-
req_id: number;
115-
timing: bigint;
116-
}
117-
118-
export class WSConnResponse {
119-
code: number;
120-
message: string;
121-
action: string;
122-
req_id: number;
123-
timing: bigint;
124-
125-
constructor(msg: IWSConnResponse) {
126-
this.code = msg.code;
127-
this.message = msg.message;
128-
this.action = msg.action;
129-
this.req_id = msg.req_id;
130-
this.timing = BigInt(msg.timing);
131-
}
132-
}

‎nodejs/src/tmq/config.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,7 @@ export class TmqConfig {
1010
client_id: string | null = null;
1111
offset_rest: string | null = null;
1212
topics?: Array<string>;
13-
auto_commit: boolean = false;
13+
auto_commit: boolean = true;
1414
auto_commit_interval_ms: number = 5 * 1000;
1515
timeout: number = 5000;
1616
otherConfigs: Map<string, any>;

‎nodejs/src/tmq/wsTmq.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -32,7 +32,7 @@ export class WsConsumer {
3232
await wsSql.connect();
3333
await wsSql.checkVersion();
3434
await this._wsClient.ready();
35-
}else {
35+
} else {
3636
throw(new TDWebSocketClientError(ErrorCode.ERR_WEBSOCKET_CONNECTION_FAIL, `connection creation failed, url: ${this._wsConfig.url}`));
3737
}
3838
}catch (e: any) {

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

Lines changed: 87 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,87 @@
1+
2+
import { TMQConstants } from "../../src/tmq/constant";
3+
import { WsConsumer } from "../../src/tmq/wsTmq";
4+
import { WebSocketConnectionPool } from "../../src/client/wsConnectorPool";
5+
import logger, { setLevel } from "../../src/common/log"
6+
import { WSConfig } from "../../src/common/config";
7+
import { WsSql } from "../../src/sql/wsSql";
8+
9+
beforeAll(async () => {
10+
const url = `wss://${process.env.TDENGINE_CLOUD_URL}?token=${process.env.TDENGINE_CLOUD_TOKEN}`;
11+
let wsSql = null;
12+
try {
13+
const conf = new WSConfig(url);
14+
conf.setUser('root')
15+
conf.setPwd('taosdata')
16+
wsSql = await WsSql.open(conf)
17+
let sql = `INSERT INTO dmeters.d1001 USING dmeters.meters (groupid, location) TAGS(2, 'SanFrancisco')
18+
VALUES (NOW + 1a, 10.30000, 219, 0.31000) (NOW + 2a, 12.60000, 218, 0.33000) (NOW + 3a, 12.30000, 221, 0.31000)
19+
dmeters.d1002 USING dmeters.meters (groupid, location) TAGS(3, 'SanFrancisco')
20+
VALUES (NOW + 1a, 10.30000, 218, 0.25000)`
21+
let res = await wsSql.exec(sql);
22+
console.log(res);
23+
expect(res.getAffectRows()).toBeGreaterThanOrEqual(3);
24+
} catch (err) {
25+
throw err;
26+
} finally {
27+
if (wsSql) {
28+
await wsSql.close();
29+
}
30+
}
31+
})
32+
33+
describe('TDWebSocket.Tmq()', () => {
34+
jest.setTimeout(20 * 1000)
35+
36+
test('normal connect', async() => {
37+
const url = `wss://${process.env.TDENGINE_CLOUD_URL}?token=${process.env.TDENGINE_CLOUD_TOKEN}`;
38+
// const TDENGINE_CLOUD_URL = 'wss://gw.cloud.taosdata.com?token=1eb78307be0681ac2fc07c2817ba8a9719641fb9';
39+
const topic = 'topic_meters';
40+
const topics = [topic];
41+
const groupId = 'group1';
42+
const clientId = 'client1';
43+
let configMap = new Map([
44+
[TMQConstants.GROUP_ID, groupId],
45+
[TMQConstants.CLIENT_ID, clientId],
46+
[TMQConstants.AUTO_OFFSET_RESET, 'earliest'],
47+
[TMQConstants.WS_URL, url],
48+
[TMQConstants.ENABLE_AUTO_COMMIT, 'true'],
49+
[TMQConstants.AUTO_COMMIT_INTERVAL_MS, '1000'],
50+
])
51+
52+
setLevel("debug");
53+
// create consumer
54+
let consumer = await WsConsumer.newConsumer(configMap);
55+
console.log(
56+
`Create consumer successfully, host: ${url}, groupId: ${groupId}, clientId: ${clientId}`
57+
);
58+
// subscribe
59+
await consumer.subscribe(topics);
60+
console.log(`Subscribe topics successfully, topics: ${topics}`);
61+
let res = new Map();
62+
while (res.size == 0) {
63+
// poll
64+
res = await consumer.poll(1000);
65+
for (let [key, value] of res) {
66+
// Add your data processing logic here
67+
console.log(`data: ${key} ${value}`);
68+
}
69+
// commit
70+
await consumer.commit();
71+
}
72+
73+
// seek
74+
let assignment = await consumer.assignment();
75+
await consumer.seekToBeginning(assignment);
76+
console.log('Assignment seek to beginning successfully');
77+
78+
// clean
79+
await consumer.unsubscribe();
80+
await consumer.close();
81+
82+
});
83+
})
84+
85+
afterAll(async () => {
86+
WebSocketConnectionPool.instance().destroyed()
87+
})

0 commit comments

Comments
 (0)