Skip to content

Commit e742001

Browse files
qevolgqevolg
committed
feat: support user_app and user_ip
1 parent 6a014fa commit e742001

14 files changed

Lines changed: 363 additions & 46 deletions

File tree

‎AGENTS.md‎

Lines changed: 57 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,57 @@
1+
# AGENTS.md — taos-connector-node
2+
3+
Think in English, but always provide your final response in Chinese.
4+
5+
## Build, test, and lint commands
6+
7+
Run all commands from `nodejs/`.
8+
9+
- Install deps: `npm install`
10+
- Build (TypeScript compile): `npm run build`
11+
- Run full tests: `npm run test`
12+
- Run a single test file: `npm run test -- test/common/dsn.test.ts`
13+
- Run one test by name: `npm run test -- -t "deduplicates concurrent reconnect triggers with reconnect lock"`
14+
- Run integration example used in CI: `npm run example`
15+
16+
Test prerequisites from project docs/CI:
17+
- Local `taosd` and `taosAdapter` must be running.
18+
- CI runs on Node.js 16.x and 20.x.
19+
20+
There is currently no dedicated lint script in `nodejs/package.json`.
21+
22+
## High-level architecture
23+
24+
- Public entrypoint: `nodejs/src/index.ts` exposes `sqlConnect`, `tmqConnect`, logging level control, and pool teardown; root `index.ts` re-exports the library surface.
25+
- Connection stack:
26+
- `WSConfig` + `getDsn()` (`src/common/config.ts`, `src/common/utils.ts`) normalize connection inputs.
27+
- `parse()` in `src/common/dsn.ts` supports multi-address DSNs and endpoint-aware paths.
28+
- `WsClient` (`src/client/wsClient.ts`) handles connect/auth/version checks and delegates transport work.
29+
- `WebSocketConnectionPool` + `WebSocketConnector` (`src/client/wsConnectorPool.ts`, `src/client/wsConnector.ts`) provide pooled sockets, reconnect/failover, and in-flight request replay.
30+
- `WsEventCallback` (`src/client/wsEventCallback.ts`) is the async request/response registry keyed by req_id/id/action.
31+
- SQL path:
32+
- `WsSql` (`src/sql/wsSql.ts`) is the high-level SQL API.
33+
- `WSRows` (`src/sql/wsRows.ts`) streams blocks incrementally.
34+
- `TaosResult` + `parseBlock()` (`src/common/taosResult.ts`) decode binary protocol blocks into row/meta structures.
35+
- Statement path:
36+
- `WsSql.stmtInit()` switches between `WsStmt1` and `WsStmt2` based on server version (`compareVersions` + `minStmt2Version`).
37+
- `WsStmt2` includes explicit network-recovery flow for stmt lifecycle steps.
38+
- TMQ path:
39+
- `TmqConfig` (`src/tmq/config.ts`) parses config map and builds TMQ/SQL DSNs.
40+
- `WsConsumer` (`src/tmq/wsTmq.ts`) wraps subscribe/poll/commit and restores subscriptions after reconnect.
41+
42+
## Key repository conventions
43+
44+
- DSN handling is central and non-trivial:
45+
- Multi-address URLs are first-class (`host1,host2,[::1]`), with host deduplication and cloud-aware default ports.
46+
- Endpoint determines websocket path (`sql -> ws`, `tmq -> rest/tmq`).
47+
- Pooling identity is normalized:
48+
- Connection pool keys sort addresses and append endpoint path + SHA-256 auth scope hash (`username/password/token/bearer_token`), so equivalent DSNs share pools while auth contexts stay isolated.
49+
- Reconnect behavior is selective:
50+
- Retry/backoff comes from DSN params (`retries`, `retry_backoff_ms`, `retry_backoff_max_ms`).
51+
- Only safe actions are replayed after reconnect (`insert`, `options_connection`, `poll`, `subscribe`, and selected binary op codes).
52+
- Credentials must stay masked in logs:
53+
- Keep using existing masking helpers (`Dsn.toString()`, `maskSensitiveForLog`, `maskUrlForLog`, `maskTmqConfigForLog`) when adding logs around DSNs/messages/config.
54+
- Test conventions:
55+
- Jest + `ts-jest`; path aliases are `@src/*` and `@test-helpers/*`.
56+
- Tests are mostly integration-style and typically create/drop DB resources in `beforeAll`/`afterAll`.
57+

‎docs/specs/2026-04-14-user-app-ip-design.md‎

Lines changed: 31 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -90,25 +90,25 @@ private buildConnMessage(database?: string | undefined | null) {
9090

9191
### 4. `nodejs/src/sql/wsSql.ts` — WsSql.open() 中 options_connection 设置
9292

93-
在连接成功、设置 timezone 之后，追加 `user_app` 和 `user_ip` 的 options_connection 调用：
93+
在连接成功、设置 timezone 之后，**无条件**下发 `user_app` 和 `user_ip` 的 options_connection 调用。当值未设置时发送 `null` 清空，防止连接池复用导致旧值残留：
9494

9595
```typescript
9696
let userApp = wsConfig.getUserApp();
97-
if (userApp && userApp.length > 0) {
98-
await wsSql._wsClient.setOptionConnection(
99-
TSDB_OPTION_CONNECTION.TSDB_OPTION_CONNECTION_USER_APP,
100-
userApp
101-
);
102-
}
97+
await wsSql._wsClient.setOptionConnection(
98+
TSDB_OPTION_CONNECTION.TSDB_OPTION_CONNECTION_USER_APP,
99+
userApp && userApp.length > 0 ? userApp : null
100+
);
103101
let userIp = wsConfig.getUserIp();
104-
if (userIp && userIp.length > 0) {
105-
await wsSql._wsClient.setOptionConnection(
106-
TSDB_OPTION_CONNECTION.TSDB_OPTION_CONNECTION_USER_IP,
107-
userIp
108-
);
109-
}
102+
await wsSql._wsClient.setOptionConnection(
103+
TSDB_OPTION_CONNECTION.TSDB_OPTION_CONNECTION_USER_IP,
104+
userIp && userIp.length > 0 ? userIp : null
105+
);
110106
```
111107

108+
> **设计考量**：连接池 key 不含 `user_app`/`user_ip`（避免不必要的池碎片化）。因此同一池中复用的连接可能携带前一会话的值。无条件下发 options_connection（有值设值、无值清空）确保每次会话都拥有正确的状态。
109+
110+
**失败策略**：遵循现有 timezone 的 fail-close 模式 — 若 `setOptionConnection` 失败，异常被 `WsSql.open()` 的 try/catch 捕获，连接关闭并向调用方抛出错误。不做降级或重试。
111+
112112
### 5. TMQ 路径
113113

114114
#### 5a. `nodejs/src/tmq/constant.ts` — TMQConstants 新增常量
@@ -138,26 +138,26 @@ case TMQConstants.USER_IP:
138138
break;
139139
```
140140

141-
如果 config map 未设置但 DSN query params 中有值，则回退获取：
141+
如果 config map 未设置（值仍为 `null`）但 DSN query params 中有值，则回退获取。使用 `== null` 判断（覆盖 `null` 和 `undefined`），避免将用户显式传入的空字符串 `""` 误判为"未设置"：
142142

143143
```typescript
144-
if (!this.userApp && this.dsn?.params.has("user_app")) {
144+
if (this.userApp == null && this.dsn?.params.has("user_app")) {
145145
this.userApp = this.dsn.params.get("user_app") || null;
146146
}
147-
if (!this.userIp && this.dsn?.params.has("user_ip")) {
147+
if (this.userIp == null && this.dsn?.params.has("user_ip")) {
148148
this.userIp = this.dsn.params.get("user_ip") || null;
149149
}
150150
```
151151

152152
#### 5c. `nodejs/src/tmq/wsTmq.ts` — subscribe 请求包含 app 和 ip
153153

154-
在 `buildSubscribeMessage()` 的 args 中新增：
154+
在 `buildSubscribeMessage()` 的 args 中，使用条件展开（与 `buildConnMessage()` 保持一致），仅在有效值时发送字段：
155155

156156
```typescript
157157
args: {
158158
// ... 现有字段 ...
159-
app: this._config.userApp,
160-
ip: this._config.userIp,
159+
...(this._config.userApp && { app: this._config.userApp }),
160+
...(this._config.userIp && { ip: this._config.userIp }),
161161
connector: ConnectorInfo,
162162
}
163163
```
@@ -217,4 +217,15 @@ const consumer2 = await tmqConnect(configMap2);
217217

218218
- 不在连接器端截断 `app` 名称（由 taosAdapter 服务端处理）
219219
- 不做 IP 格式校验（由 taosAdapter 服务端处理）
220-
- 不修改连接池 key 逻辑（`user_app` 和 `user_ip` 不影响连接池标识）
220+
- 不修改连接池 key 逻辑（通过无条件下发 options_connection 清空/设置解决状态残留，避免池碎片化）
221+
222+
## 错误处理策略
223+
224+
| 场景 | 行为 |
225+
|---|---|
226+
| `setOptionConnection` 设置 `user_app` 失败 | Fail-close：关闭连接，抛出错误，WsSql.open() 整体失败 |
227+
| `setOptionConnection` 设置 `user_ip` 失败 | 同上 |
228+
| `setOptionConnection` 清空（value: null）失败 | 同上 |
229+
| TMQ subscribe 时 `app`/`ip` 被服务端拒绝 | subscribe 调用失败，错误传播至调用方 |
230+
231+
此策略与现有 `timezone` 的 fail-close 行为一致（`WsSql.open()` 中 try/catch 统一处理）。

‎nodejs/index.ts‎

Lines changed: 34 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -1,25 +1,34 @@
1-
export * from "./src/client/wsClient";
2-
export * from "./src/client/wsConnector";
3-
export * from "./src/client/wsConnectorPool";
4-
export * from "./src/client/wsEventCallback";
5-
export * from "./src/client/wsResponse";
6-
export * from "./src/common/config";
7-
export * from "./src/common/constant";
8-
export * from "./src/common/log";
9-
export * from "./src/common/reqid";
10-
export * from "./src/common/taosResult";
11-
export * from "./src/common/ut8Helper";
12-
export * from "./src/common/utils";
13-
export * from "./src/common/wsError";
14-
export * from "./src/common/wsOptions";
15-
export * from "./src/index";
16-
export * from "./src/sql/wsProto";
17-
export * from "./src/sql/wsRows";
18-
export * from "./src/sql/wsSql";
19-
export * from "./src/stmt/wsParamsBase";
20-
export * from "./src/stmt/wsProto";
21-
export * from "./src/stmt/wsStmt";
22-
export * from "./src/tmq/config";
23-
export * from "./src/tmq/constant";
24-
export * from "./src/tmq/tmqResponse";
25-
export * from "./src/tmq/wsTmq";
1+
export * from "./src/client/wsClient"
2+
export * from "./src/client/wsConnector"
3+
export * from "./src/client/wsConnectorPool"
4+
export * from "./src/client/wsEventCallback"
5+
export * from "./src/client/wsResponse"
6+
export * from "./src/common/addressConnectionTracker"
7+
export * from "./src/common/config"
8+
export * from "./src/common/constant"
9+
export * from "./src/common/dsn"
10+
export * from "./src/common/log"
11+
export * from "./src/common/reqid"
12+
export * from "./src/common/taosResult"
13+
export * from "./src/common/ut8Helper"
14+
export * from "./src/common/utils"
15+
export * from "./src/common/wsError"
16+
export * from "./src/common/wsOptions"
17+
export * from "./src/index"
18+
export * from "./src/sql/wsProto"
19+
export * from "./src/sql/wsRows"
20+
export * from "./src/sql/wsSql"
21+
export * from "./src/stmt/FieldBindParams"
22+
export * from "./src/stmt/wsColumnInfo"
23+
export * from "./src/stmt/wsParams1"
24+
export * from "./src/stmt/wsParams2"
25+
export * from "./src/stmt/wsParamsBase"
26+
export * from "./src/stmt/wsProto"
27+
export * from "./src/stmt/wsStmt"
28+
export * from "./src/stmt/wsStmt1"
29+
export * from "./src/stmt/wsStmt2"
30+
export * from "./src/stmt/wsTableInfo"
31+
export * from "./src/tmq/config"
32+
export * from "./src/tmq/constant"
33+
export * from "./src/tmq/tmqResponse"
34+
export * from "./src/tmq/wsTmq"

‎nodejs/src/client/wsClient.ts‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,8 @@ export class WsClient {
2727
private _wsConnector?: WebSocketConnector;
2828
private _timeout?: number | undefined | null;
2929
private _timezone?: string | undefined | null;
30+
private _userApp?: string | undefined | null;
31+
private _userIp?: string | undefined | null;
3032
private readonly _dsn: Dsn;
3133
private static readonly _minVersion = "3.3.2.0";
3234
private _version?: string | undefined | null;
@@ -42,6 +44,12 @@ export class WsClient {
4244
if (this._dsn.params.has("timezone")) {
4345
this._timezone = this._dsn.params.get("timezone") || undefined;
4446
}
47+
if (this._dsn.params.has("user_app")) {
48+
this._userApp = this._dsn.params.get("user_app") || undefined;
49+
}
50+
if (this._dsn.params.has("user_ip")) {
51+
this._userIp = this._dsn.params.get("user_ip") || undefined;
52+
}
4553
if (this._dsn.params.has("bearer_token")) {
4654
this._bearerToken = this._dsn.params.get("bearer_token") || undefined;
4755
}
@@ -57,6 +65,8 @@ export class WsClient {
5765
db: database,
5866
connector: ConnectorInfo,
5967
...(this._timezone && { tz: this._timezone }),
68+
...(this._userApp && { app: this._userApp }),
69+
...(this._userIp && { ip: this._userIp }),
6070
...(this._bearerToken && { bearer_token: this._bearerToken }),
6171
},
6272
};

‎nodejs/src/common/config.ts‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,8 @@ export class WSConfig {
88
private _timeout: number | undefined | null;
99
private _token: string | undefined | null;
1010
private _timezone: string | undefined | null;
11+
private _userApp: string | undefined | null;
12+
private _userIp: string | undefined | null;
1113
private _minStmt2Version: string;
1214
private _bearerToken: string | undefined | null;
1315

@@ -76,6 +78,22 @@ export class WSConfig {
7678
this._timezone = timezone;
7779
}
7880

81+
public getUserApp(): string | undefined | null {
82+
return this._userApp;
83+
}
84+
85+
public setUserApp(userApp: string) {
86+
this._userApp = userApp;
87+
}
88+
89+
public getUserIp(): string | undefined | null {
90+
return this._userIp;
91+
}
92+
93+
public setUserIp(userIp: string) {
94+
this._userIp = userIp;
95+
}
96+
7997
public getBearerToken(): string | undefined | null {
8098
return this._bearerToken;
8199
}

‎nodejs/src/common/utils.ts‎

Lines changed: 14 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,20 @@ export function getDsn(wsConfig: WSConfig): Dsn {
3737
wsConfig.setTimezone(dsn.params.get("timezone") || "");
3838
}
3939

40+
const userApp = wsConfig.getUserApp();
41+
if (userApp) {
42+
dsn.params.set("user_app", userApp);
43+
} else if (dsn.params.has("user_app")) {
44+
wsConfig.setUserApp(dsn.params.get("user_app") || "");
45+
}
46+
47+
const userIp = wsConfig.getUserIp();
48+
if (userIp) {
49+
dsn.params.set("user_ip", userIp);
50+
} else if (dsn.params.has("user_ip")) {
51+
wsConfig.setUserIp(dsn.params.get("user_ip") || "");
52+
}
53+
4054
const db = wsConfig.getDb();
4155
if (db && db.length > 0) {
4256
dsn.database = db;
@@ -240,4 +254,3 @@ export function maskTmqConfigForLog(config: TmqConfig): string {
240254
}
241255
});
242256
}
243-

‎nodejs/src/sql/wsSql.ts‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -57,6 +57,16 @@ export class WsSql {
5757
timezone
5858
);
5959
}
60+
let userApp = wsConfig.getUserApp();
61+
await wsSql._wsClient.setOptionConnection(
62+
TSDB_OPTION_CONNECTION.TSDB_OPTION_CONNECTION_USER_APP,
63+
userApp && userApp.length > 0 ? userApp : null
64+
);
65+
let userIp = wsConfig.getUserIp();
66+
await wsSql._wsClient.setOptionConnection(
67+
TSDB_OPTION_CONNECTION.TSDB_OPTION_CONNECTION_USER_IP,
68+
userIp && userIp.length > 0 ? userIp : null
69+
);
6070
return wsSql;
6171
} catch (e: any) {
6272
logger.error(`WsSql open is failed, ${e.code}, ${e.message}`);

‎nodejs/src/tmq/config.ts‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,8 @@ export class TmqConfig {
77
user: string | null = null;
88
password: string | null = null;
99
token: string | null = null;
10+
userApp: string | null = null;
11+
userIp: string | null = null;
1012
group_id: string | null = null;
1113
client_id: string | null = null;
1214
offset_rest: string | null = null;
@@ -34,6 +36,12 @@ export class TmqConfig {
3436
this.token = value;
3537
this.otherConfigs.set(key, value);
3638
break;
39+
case TMQConstants.USER_APP:
40+
this.userApp = value;
41+
break;
42+
case TMQConstants.USER_IP:
43+
this.userIp = value;
44+
break;
3745
case TMQConstants.GROUP_ID:
3846
this.group_id = value;
3947
break;
@@ -82,6 +90,13 @@ export class TmqConfig {
8290
}
8391
}
8492

93+
if (this.userApp == null && this.dsn.params.has("user_app")) {
94+
this.userApp = this.dsn.params.get("user_app") || null;
95+
}
96+
if (this.userIp == null && this.dsn.params.has("user_ip")) {
97+
this.userIp = this.dsn.params.get("user_ip") || null;
98+
}
99+
85100
this.sqlDsn = new Dsn(
86101
this.dsn.scheme,
87102
this.dsn.username,

‎nodejs/src/tmq/constant.ts‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -87,6 +87,10 @@ export class TMQConstants {
8787
* only valid in websocket
8888
*/
8989
public static CONNECT_MESSAGE_TIMEOUT: string = "messageWaitTimeout";
90+
91+
public static USER_APP: string = "user_app";
92+
93+
public static USER_IP: string = "user_ip";
9094
}
9195

9296
export class TMQMessageType {

‎nodejs/src/tmq/wsTmq.ts‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -111,6 +111,8 @@ export class WsConsumer {
111111
auto_commit: String(this._config.auto_commit),
112112
auto_commit_interval_ms: String(this._config.auto_commit_interval_ms),
113113
config: Object.fromEntries(this._config.otherConfigs),
114+
...(this._config.userApp && { app: this._config.userApp }),
115+
...(this._config.userIp && { ip: this._config.userIp }),
114116
connector: ConnectorInfo,
115117
},
116118
};

0 commit comments

Comments
 (0)